1 files changed, 23 insertions, 8 deletions
diff --git a/gr-zeromq/lib/pull_source_impl.cc b/gr-zeromq/lib/pull_source_impl.cc
index 479b15bb32..96aa9d853a 100644
--- a/gr-zeromq/lib/pull_source_impl.cc
+++ b/gr-zeromq/lib/pull_source_impl.cc
@@ -26,22 +26,23 @@
 
 #include <gnuradio/io_signature.h>
 #include "pull_source_impl.h"
+#include "tag_headers.h"
 
 namespace gr {
   namespace zeromq {
 
     pull_source::sptr
-    pull_source::make(size_t itemsize, size_t vlen, char *address, int timeout)
+    pull_source::make(size_t itemsize, size_t vlen, char *address, int timeout, bool pass_tags)
     {
       return gnuradio::get_initial_sptr
-        (new pull_source_impl(itemsize, vlen, address, timeout));
+        (new pull_source_impl(itemsize, vlen, address, timeout, pass_tags));
     }
 
-    pull_source_impl::pull_source_impl(size_t itemsize, size_t vlen, char *address, int timeout)
+    pull_source_impl::pull_source_impl(size_t itemsize, size_t vlen, char *address, int timeout, bool pass_tags)
       : gr::sync_block("pull_source",
                        gr::io_signature::make(0, 0, 0),
                        gr::io_signature::make(1, 1, itemsize * vlen)),
-        d_itemsize(itemsize), d_vlen(vlen), d_timeout(timeout)
+        d_itemsize(itemsize), d_vlen(vlen), d_timeout(timeout), d_pass_tags(pass_tags)
     {
       int major, minor, patch;
       zmq::version (&major, &minor, &patch);
@@ -81,16 +82,30 @@ namespace gr {
         // Receive data
         zmq::message_t msg;
         d_socket->recv(&msg);
+
+        // check header for tags...
+        std::string buf(static_cast<char*>(msg.data()), msg.size());
+        if(d_pass_tags){
+            uint64_t rcv_offset;
+            std::vector<gr::tag_t> tags;
+            buf = parse_tag_header(buf, rcv_offset, tags);
+            for(size_t i=0; i<tags.size(); i++){
+                tags[i].offset -= rcv_offset - nitems_read(0);
+                add_item_tag(0, tags[i]);
+                }
+            }
+
+
         // Copy to ouput buffer and return
-        if (msg.size() >= d_itemsize*d_vlen*noutput_items) {
-          memcpy(out, (void *)msg.data(), d_itemsize*d_vlen*noutput_items);
+        if (buf.size() >= d_itemsize*d_vlen*noutput_items) {
+          memcpy(out, (void *)&buf[0], d_itemsize*d_vlen*noutput_items);
 
           return noutput_items;
         }
         else {
-          memcpy(out, (void *)msg.data(), msg.size());
+          memcpy(out, (void *)&buf[0], buf.size());
 
-          return msg.size()/(d_itemsize*d_vlen);
+          return buf.size()/(d_itemsize*d_vlen);
         }
       }
       else {