From b161a35d47f4143b8fe460fee9ffd38a348d08e3 Mon Sep 17 00:00:00 2001
From: Tim O'Shea <tim.oshea753@gmail.com>
Date: Mon, 27 Oct 2014 21:21:47 -0400
Subject: zmq: all source blocks should now support tag headers
---
gr-zeromq/lib/pull_source_impl.cc | 31 +++++++++++++++++++++++--------
1 file changed, 23 insertions(+), 8 deletions(-)
(limited to 'gr-zeromq/lib/pull_source_impl.cc')
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 {
--
cgit v1.2.3