diff options
Diffstat (limited to 'gr-zeromq/lib/push_msg_sink_impl.cc')
-rw-r--r-- | gr-zeromq/lib/push_msg_sink_impl.cc | 60 |
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 */ |