[Commit-gnuradio] [gnuradio] 07/29: zmq: all source blocks should now support tag headers

[git]

This is an automated email from the git hooks/post-receive script.

jcorgan pushed a commit to branch master
in repository gnuradio.

commit b161a35d47f4143b8fe460fee9ffd38a348d08e3
Author: Tim O'Shea <address@hidden>
Date:   Mon Oct 27 21:21:47 2014 -0400

    zmq: all source blocks should now support tag headers
---
 gr-zeromq/include/gnuradio/zeromq/pull_source.h |  2 +-
 gr-zeromq/include/gnuradio/zeromq/req_source.h  |  2 +-
 gr-zeromq/lib/pull_source_impl.cc               | 31 ++++++++++++++++++-------
 gr-zeromq/lib/pull_source_impl.h                |  3 ++-
 gr-zeromq/lib/req_source_impl.cc                | 27 ++++++++++++++++-----
 gr-zeromq/lib/req_source_impl.h                 |  3 ++-
 gr-zeromq/lib/sub_source_impl.cc                |  2 +-
 7 files changed, 51 insertions(+), 19 deletions(-)

diff --git a/gr-zeromq/include/gnuradio/zeromq/pull_source.h 
b/gr-zeromq/include/gnuradio/zeromq/pull_source.h
index 5c1d37d..6dec9c2 100644
--- a/gr-zeromq/include/gnuradio/zeromq/pull_source.h
+++ b/gr-zeromq/include/gnuradio/zeromq/pull_source.h
@@ -51,7 +51,7 @@ namespace gr {
        * \param timeout  Receive timeout in seconds, default is 100ms, 1us 
increments
        *
        */
-      static sptr make(size_t itemsize, size_t vlen, char *address, int 
timeout=100);
+      static sptr make(size_t itemsize, size_t vlen, char *address, int 
timeout=100, bool pass_tags=false);
     };
 
   } // namespace zeromq
diff --git a/gr-zeromq/include/gnuradio/zeromq/req_source.h 
b/gr-zeromq/include/gnuradio/zeromq/req_source.h
index c272c28..d9f55d3 100644
--- a/gr-zeromq/include/gnuradio/zeromq/req_source.h
+++ b/gr-zeromq/include/gnuradio/zeromq/req_source.h
@@ -52,7 +52,7 @@ namespace gr {
        * \param timeout  Receive timeout in seconds, default is 100ms, 1us 
increments
        *
        */
-      static sptr make(size_t itemsize, size_t vlen, char *address, int 
timeout=100);
+      static sptr make(size_t itemsize, size_t vlen, char *address, int 
timeout=100, bool pass_tags=false);
     };
 
   } // namespace zeromq
diff --git a/gr-zeromq/lib/pull_source_impl.cc 
b/gr-zeromq/lib/pull_source_impl.cc
index 479b15b..96aa9d8 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 {
diff --git a/gr-zeromq/lib/pull_source_impl.h b/gr-zeromq/lib/pull_source_impl.h
index e69de81..f25013b 100644
--- a/gr-zeromq/lib/pull_source_impl.h
+++ b/gr-zeromq/lib/pull_source_impl.h
@@ -37,9 +37,10 @@ namespace gr {
       int             d_timeout; // microseconds, -1 is blocking
       zmq::context_t  *d_context;
       zmq::socket_t   *d_socket;
+      bool            d_pass_tags;
 
     public:
-      pull_source_impl(size_t itemsize, size_t vlen, char *address, int 
timeout);
+      pull_source_impl(size_t itemsize, size_t vlen, char *address, int 
timeout, bool pass_tags=false);
       ~pull_source_impl();
 
       int work(int noutput_items,
diff --git a/gr-zeromq/lib/req_source_impl.cc b/gr-zeromq/lib/req_source_impl.cc
index 54cd659..2cc61e4 100644
--- a/gr-zeromq/lib/req_source_impl.cc
+++ b/gr-zeromq/lib/req_source_impl.cc
@@ -26,22 +26,23 @@
 
 #include <gnuradio/io_signature.h>
 #include "req_source_impl.h"
+#include "tag_headers.h"
 
 namespace gr {
   namespace zeromq {
 
     req_source::sptr
-    req_source::make(size_t itemsize, size_t vlen, char *address, int timeout)
+    req_source::make(size_t itemsize, size_t vlen, char *address, int timeout, 
bool pass_tags)
     {
       return gnuradio::get_initial_sptr
-        (new req_source_impl(itemsize, vlen, address, timeout));
+        (new req_source_impl(itemsize, vlen, address, timeout, pass_tags));
     }
 
-    req_source_impl::req_source_impl(size_t itemsize, size_t vlen, char 
*address, int timeout)
+    req_source_impl::req_source_impl(size_t itemsize, size_t vlen, char 
*address, int timeout, bool pass_tags)
       : gr::sync_block("req_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);
@@ -89,9 +90,23 @@ namespace gr {
         zmq::message_t reply;
         d_socket->recv(&reply);
 
+        // Deserialize header data / tags
+        std::string buf(static_cast<char*>(reply.data()), reply.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
-        memcpy(out, (void *)reply.data(), reply.size());
-        return reply.size()/(d_itemsize*d_vlen);
+        memcpy(out, (void *)&buf[0], buf.size());
+        return buf.size()/(d_itemsize*d_vlen);
       }
 
       return 0;
diff --git a/gr-zeromq/lib/req_source_impl.h b/gr-zeromq/lib/req_source_impl.h
index f61b1d1..d37d73b 100644
--- a/gr-zeromq/lib/req_source_impl.h
+++ b/gr-zeromq/lib/req_source_impl.h
@@ -37,9 +37,10 @@ namespace gr {
       int             d_timeout;
       zmq::context_t  *d_context;
       zmq::socket_t   *d_socket;
+      bool            d_pass_tags;
 
     public:
-      req_source_impl(size_t itemsize, size_t vlen, char *address, int 
timeout);
+      req_source_impl(size_t itemsize, size_t vlen, char *address, int 
timeout, bool pass_tags=false);
       ~req_source_impl();
 
       int work(int noutput_items,
diff --git a/gr-zeromq/lib/sub_source_impl.cc b/gr-zeromq/lib/sub_source_impl.cc
index 9276a5e..f1aeb91 100644
--- a/gr-zeromq/lib/sub_source_impl.cc
+++ b/gr-zeromq/lib/sub_source_impl.cc
@@ -35,7 +35,7 @@ namespace gr {
     sub_source::make(size_t itemsize, size_t vlen, char *address, int timeout, 
bool pass_tags)
     {
       return gnuradio::get_initial_sptr
-        (new sub_source_impl(itemsize, vlen, address, timeout));
+        (new sub_source_impl(itemsize, vlen, address, timeout, pass_tags));
     }
 
     sub_source_impl::sub_source_impl(size_t itemsize, size_t vlen, char 
*address, int timeout, bool pass_tags)