summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorTim O'Shea <tim.oshea753@gmail.com>2014-10-27 21:21:47 -0400
committerTim O'Shea <tim.oshea753@gmail.com>2014-10-27 21:21:47 -0400
commitb161a35d47f4143b8fe460fee9ffd38a348d08e3 (patch)
treee18e23c4a55bd54db29d43ad9b90b9d98688fb38
parent169a6c796d7fd3ff5dfc5114e38e6a05000068d5 (diff)
zmq: all source blocks should now support tag headers
Diffstat
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/pull_source.h2
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/req_source.h2
-rw-r--r--gr-zeromq/lib/pull_source_impl.cc31
-rw-r--r--gr-zeromq/lib/pull_source_impl.h3
-rw-r--r--gr-zeromq/lib/req_source_impl.cc27
-rw-r--r--gr-zeromq/lib/req_source_impl.h3
-rw-r--r--gr-zeromq/lib/sub_source_impl.cc2
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 5c1d37d353..6dec9c2b34 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 c272c28510..d9f55d3c72 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 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 {
diff --git a/gr-zeromq/lib/pull_source_impl.h b/gr-zeromq/lib/pull_source_impl.h
index e69de81dcd..f25013b373 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 54cd659972..2cc61e40d9 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 f61b1d1ce4..d37d73b525 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 9276a5e281..f1aeb91dd0 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)
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-21 19:18:52 +0000