summaryrefslogtreecommitdiff
path: root/gr-zeromq/lib/push_msg_sink_impl.cc
diff options
context:
space:
mode:
Diffstat (limited to 'gr-zeromq/lib/push_msg_sink_impl.cc')
-rw-r--r--gr-zeromq/lib/push_msg_sink_impl.cc60
1 files changed, 13 insertions, 47 deletions
diff --git a/gr-zeromq/lib/push_msg_sink_impl.cc b/gr-zeromq/lib/push_msg_sink_impl.cc
index 6266cd6a57..e9cc5bc435 100644
--- a/gr-zeromq/lib/push_msg_sink_impl.cc
+++ b/gr-zeromq/lib/push_msg_sink_impl.cc
@@ -39,25 +39,28 @@ namespace gr {
}
push_msg_sink_impl::push_msg_sink_impl(char *address, int timeout)
- : gr::sync_block("push_msg_sink",
- gr::io_signature::make(0, 0, 0),
- gr::io_signature::make(0, 0, 0)),
+ : gr::block("push_msg_sink",
+ gr::io_signature::make(0, 0, 0),
+ gr::io_signature::make(0, 0, 0)),
d_timeout(timeout)
{
int major, minor, patch;
zmq::version (&major, &minor, &patch);
+
if (major < 3) {
d_timeout = timeout*1000;
}
+
d_context = new zmq::context_t(1);
d_socket = new zmq::socket_t(*d_context, ZMQ_PUSH);
+
int time = 0;
d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time));
- d_socket->bind (address);
+ d_socket->bind(address);
message_port_register_in(pmt::mp("in"));
- set_msg_handler( pmt::mp("in"),
- boost::bind(&push_msg_sink_impl::handler, this, _1));
+ set_msg_handler(pmt::mp("in"),
+ boost::bind(&push_msg_sink_impl::handler, this, _1));
}
push_msg_sink_impl::~push_msg_sink_impl()
@@ -67,52 +70,15 @@ namespace gr {
delete d_context;
}
- void push_msg_sink_impl::handler(pmt::pmt_t msg){
+ void push_msg_sink_impl::handler(pmt::pmt_t msg)
+ {
std::stringbuf sb("");
pmt::serialize( msg, sb );
std::string s = sb.str();
zmq::message_t zmsg(s.size());
- memcpy( zmsg.data(), s.c_str(), s.size() );
- d_socket->send(zmsg);
- }
-
- int
- push_msg_sink_impl::work(int noutput_items,
- gr_vector_const_void_star &input_items,
- gr_vector_void_star &output_items)
- {
- return noutput_items;
-/* const char *in = (const char *) input_items[0];
-
- zmq::pollitem_t itemsout[] = { { *d_socket, 0, ZMQ_POLLOUT, 0 } };
- zmq::poll (&itemsout[0], 1, d_timeout);
-
- // If we got a reply, process
- if (itemsout[0].revents & ZMQ_POLLOUT) {
-
- // encode the current offset, # tags, and tags into header
- std::string header("");
-
- if(d_pass_tags){
- uint64_t offset = nitems_read(0);
- std::vector<gr::tag_t> tags;
- get_tags_in_range(tags, 0, nitems_read(0), nitems_read(0)+noutput_items);
- header = gen_tag_header( offset, tags );
- }
-
- // create message copy and send
- zmq::message_t msg(header.length() + d_itemsize*d_vlen*noutput_items);
- 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);
- d_socket->send(msg);
-
- return noutput_items;
- }
- else {
- return 0;
- }*/
+ memcpy(zmsg.data(), s.c_str(), s.size());
+ d_socket->send(zmsg);
}
} /* namespace zeromq */