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/req_source_impl.cc | 27 +++++++++++++++++++++------
1 file changed, 21 insertions(+), 6 deletions(-)
(limited to 'gr-zeromq/lib/req_source_impl.cc')
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;
--
cgit v1.2.3