[Commit-gnuradio] [gnuradio] 12/29: zmq: stream tag passing now works

[git]

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

jcorgan pushed a commit to branch master
in repository gnuradio.

commit 585cc65be340fe1ed4c99ff397c3d71f49b621c5
Author: Tim O'Shea <address@hidden>
Date:   Fri Dec 26 19:03:51 2014 +0100

    zmq: stream tag passing now works
---
 gr-zeromq/lib/pub_sink_impl.cc    |  1 +
 gr-zeromq/lib/pull_source_impl.cc |  2 +-
 gr-zeromq/lib/req_source_impl.cc  |  2 +-
 gr-zeromq/lib/sub_source_impl.cc  |  5 ++++-
 gr-zeromq/lib/tag_headers.cc      | 26 ++++++++++----------------
 gr-zeromq/lib/tag_headers.h       |  2 +-
 6 files changed, 18 insertions(+), 20 deletions(-)

diff --git a/gr-zeromq/lib/pub_sink_impl.cc b/gr-zeromq/lib/pub_sink_impl.cc
index 6804c1d..5afcb72 100644
--- a/gr-zeromq/lib/pub_sink_impl.cc
+++ b/gr-zeromq/lib/pub_sink_impl.cc
@@ -81,6 +81,7 @@ namespace gr {
 
       // create message copy and send
       zmq::message_t msg(header.length() + d_itemsize*d_vlen*noutput_items);
+      //std::cout << "PUB: Header Len: " << header.length() << ", Data Length: 
" << d_itemsize*d_vlen*noutput_items << "\n";
       if(d_pass_tags)
         memcpy((void*) msg.data(), header.c_str(), header.length() );
       memcpy((uint8_t *)msg.data() + header.length(), in, 
d_itemsize*d_vlen*noutput_items);
diff --git a/gr-zeromq/lib/pull_source_impl.cc 
b/gr-zeromq/lib/pull_source_impl.cc
index 96aa9d8..87b330a 100644
--- a/gr-zeromq/lib/pull_source_impl.cc
+++ b/gr-zeromq/lib/pull_source_impl.cc
@@ -90,7 +90,7 @@ namespace gr {
             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);
+                tags[i].offset -= rcv_offset - nitems_written(0);
                 add_item_tag(0, tags[i]);
                 }
             }
diff --git a/gr-zeromq/lib/req_source_impl.cc b/gr-zeromq/lib/req_source_impl.cc
index 6e2ad3d..5c5071e 100644
--- a/gr-zeromq/lib/req_source_impl.cc
+++ b/gr-zeromq/lib/req_source_impl.cc
@@ -98,7 +98,7 @@ namespace gr {
             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);
+                tags[i].offset -= rcv_offset - nitems_written(0);
                 add_item_tag(0, tags[i]);
                 }
             }
diff --git a/gr-zeromq/lib/sub_source_impl.cc b/gr-zeromq/lib/sub_source_impl.cc
index f1aeb91..813ff5a 100644
--- a/gr-zeromq/lib/sub_source_impl.cc
+++ b/gr-zeromq/lib/sub_source_impl.cc
@@ -89,9 +89,12 @@ namespace gr {
         if(d_pass_tags){
             uint64_t rcv_offset;
             std::vector<gr::tag_t> tags;
+            //int olen = buf.size();
             buf = parse_tag_header(buf, rcv_offset, tags);
+            //std::cout << "SUB: Header Len = " << olen - buf.size() << ", 
data len = " << buf.size() << "\n";
             for(size_t i=0; i<tags.size(); i++){
-                tags[i].offset -= rcv_offset - nitems_read(0);
+                //std::cout << "add item tag ... (offset = " << tags[i].offset 
<< " rcv_offset = " << rcv_offset << " nitems_read(0) = " << nitems_written(0) 
<< "\n";
+                tags[i].offset -= rcv_offset - nitems_written(0);
                 add_item_tag(0, tags[i]);
                 }
             }
diff --git a/gr-zeromq/lib/tag_headers.cc b/gr-zeromq/lib/tag_headers.cc
index 42a65be..c47bac7 100644
--- a/gr-zeromq/lib/tag_headers.cc
+++ b/gr-zeromq/lib/tag_headers.cc
@@ -36,9 +36,9 @@ namespace gr {
       ss.write( reinterpret_cast< const char* >( &offset ), sizeof(uint64_t) 
);  // offset
       ss.write( reinterpret_cast< const char* >( &ntags ), sizeof(size_t) );   
   // num tags
       std::stringbuf sb("");
-      std::cout << "TX TAGS: (offset="<<offset<<" ntags="<<ntags<<")\n";
+      //std::cout << "TX TAGS: (offset="<<offset<<" ntags="<<ntags<<")\n";
       for(size_t i=0; i<tags.size(); i++){
-        std::cout << "TX TAG: (" << tags[i].offset << ", " << tags[i].key << 
", " << tags[i].value << ", " << tags[i].srcid << ")\n";
+        //std::cout << "TX TAG: (" << tags[i].offset << ", " << tags[i].key << 
", " << tags[i].value << ", " << tags[i].srcid << ")\n";
         ss.write( reinterpret_cast< const char* >( &tags[i].offset ), 
sizeof(uint64_t) );   // offset
         sb.str("");
         pmt::serialize( tags[i].key, sb );                                     
      // key
@@ -51,35 +51,29 @@ namespace gr {
     }
 
     std::string
-    parse_tag_header(std::string &buf_in, uint64_t &offset_out, 
std::vector<gr::tag_t> tags_out) {
+    parse_tag_header(std::string &buf_in, uint64_t &offset_out, 
std::vector<gr::tag_t> &tags_out) {
 
-      std::cout << "sblen: " << buf_in.size() <<"\n";
       std::istringstream iss( buf_in );
       size_t   rcv_ntags;
       iss.read( (char*)&offset_out, sizeof(uint64_t ) );
       iss.read( (char*)&rcv_ntags,  sizeof(size_t   ) );
-      std::cout << "RX TAGS: (offset="<<offset_out<<" 
ntags="<<rcv_ntags<<")\n";
+      //std::cout << "RX TAGS: (offset="<<offset_out<<" 
ntags="<<rcv_ntags<<")\n";
       int rd_offset = sizeof(uint64_t) + sizeof(size_t);
+      std::stringbuf sb( iss.str().substr(rd_offset) );
+
       for(size_t i=0; i<rcv_ntags; i++){
         gr::tag_t newtag;       
-        iss.read( (char*)&newtag.offset, sizeof(uint64_t ) );
-        rd_offset += sizeof(uint64_t);
-
-        std::stringbuf sb( iss.str().substr(rd_offset) );
-        std::cout << "sblen: " << sb.str().size() << "\n";
-        std::cout << "sbloc: " << sb.getloc() << "\n";
-        //std::stringbuf sb( iss.str() );
-
+        sb.sgetn( (char*) &(newtag.offset), sizeof(uint64_t) );
         newtag.key   = pmt::deserialize( sb );
         newtag.value = pmt::deserialize( sb );
         newtag.srcid = pmt::deserialize( sb );
-        std::cout << "sblen(after): " << sb.str().size() << "\n";
-        std::cout << "sbloc(after): " << sb.getloc() << "\n";
+        //std::cout << "RX TAG: (" << newtag.offset << ", " << newtag.key << 
", " << newtag.value << ", " << newtag.srcid << ")\n";
         tags_out.push_back(newtag);
         iss.str(sb.str());
       }
 
-      return iss.str();
+      int ndata = sb.in_avail();
+      return iss.str().substr(iss.str().size() - ndata);
     }
   } /* namespace zeromq */
 } /* namespace gr */
diff --git a/gr-zeromq/lib/tag_headers.h b/gr-zeromq/lib/tag_headers.h
index 03300f8..4c7a812 100644
--- a/gr-zeromq/lib/tag_headers.h
+++ b/gr-zeromq/lib/tag_headers.h
@@ -32,7 +32,7 @@ namespace gr {
   namespace zeromq {
 
     std::string gen_tag_header(uint64_t &offset, std::vector<gr::tag_t> &tags);
-    std::string parse_tag_header(std::string &buf_in, uint64_t &offset_out, 
std::vector<gr::tag_t> tags_out);
+    std::string parse_tag_header(std::string &buf_in, uint64_t &offset_out, 
std::vector<gr::tag_t> &tags_out);
     
   } /* namespace zeromq */
 } /* namespace gr */