summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorTim O'Shea <tim.oshea753@gmail.com>2014-12-26 19:03:51 +0100
committerTim O'Shea <tim.oshea753@gmail.com>2014-12-26 19:05:08 +0100
commit585cc65be340fe1ed4c99ff397c3d71f49b621c5 (patch)
tree604b62f2edcad8a554005b29c59191fc4a1d3451
parentc1f42c12af1b4deedf7d72c477b40d4cc369ee41 (diff)
zmq: stream tag passing now works
Diffstat
-rw-r--r--gr-zeromq/lib/pub_sink_impl.cc1
-rw-r--r--gr-zeromq/lib/pull_source_impl.cc2
-rw-r--r--gr-zeromq/lib/req_source_impl.cc2
-rw-r--r--gr-zeromq/lib/sub_source_impl.cc5
-rw-r--r--gr-zeromq/lib/tag_headers.cc26
-rw-r--r--gr-zeromq/lib/tag_headers.h2
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 6804c1d9ce..5afcb722b1 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 96aa9d853a..87b330a862 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 6e2ad3daf4..5c5071e15e 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 f1aeb91dd0..813ff5a1c0 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 42a65be5af..c47bac7ae1 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 03300f881a..4c7a81238e 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 */
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-21 20:37:48 +0000