diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2015-01-12 16:04:45 -0800 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2015-01-12 16:06:41 -0800 |
| commit | 5de5b1a87f71bdbed9925f80648756c566c3a91b (patch) | |
| tree | 9ce1394c23245358974d9b32338c46c35d49c72b /gr-zeromq/lib/req_msg_source_impl.cc | |
| parent | 7bf8b05bde26bfb1e2df684e2ab55b878b25350c (diff) | |
zeromq: cleanup and made req_msg_source derive from gr::block
Diffstat (limited to 'gr-zeromq/lib/req_msg_source_impl.cc')
| -rw-r--r-- | gr-zeromq/lib/req_msg_source_impl.cc | 67 |
1 files changed, 29 insertions, 38 deletions
diff --git a/gr-zeromq/lib/req_msg_source_impl.cc b/gr-zeromq/lib/req_msg_source_impl.cc index 2dda15208f..b30ef2679d 100644 --- a/gr-zeromq/lib/req_msg_source_impl.cc +++ b/gr-zeromq/lib/req_msg_source_impl.cc @@ -39,18 +39,21 @@ namespace gr { } req_msg_source_impl::req_msg_source_impl(char *address, int timeout) - : gr::sync_block("req_msg_source", - gr::io_signature::make(0, 0, 0), - gr::io_signature::make(0, 0, 0)), - d_timeout(timeout) + : gr::block("req_msg_source", + 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); + 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_REQ); + int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); d_socket->connect (address); @@ -65,67 +68,55 @@ namespace gr { delete d_context; } - bool req_msg_source_impl::start(){ + bool req_msg_source_impl::start() + { d_finished = false; - d_thread = new boost::thread( boost::bind( &req_msg_source_impl::readloop , this ) ); + d_thread = new boost::thread(boost::bind(&req_msg_source_impl::readloop, this)); return true; } - bool req_msg_source_impl::stop(){ + bool req_msg_source_impl::stop() + { d_finished = true; d_thread->join(); return true; } - void req_msg_source_impl::readloop(){ + void req_msg_source_impl::readloop() + { while(!d_finished){ //std::cout << "readloop\n"; - + zmq::pollitem_t itemsout[] = { { *d_socket, 0, ZMQ_POLLOUT, 0 } }; - zmq::poll (&itemsout[0], 1, d_timeout); - + zmq::poll(&itemsout[0], 1, d_timeout); + // If we got a reply, process if (itemsout[0].revents & ZMQ_POLLOUT) { // Request data, FIXME non portable? int nmsg = 1; zmq::message_t request(sizeof(int)); - memcpy ((void *) request.data (), &nmsg, sizeof(int)); + memcpy((void *) request.data (), &nmsg, sizeof(int)); d_socket->send(request); - //std::cout << "sent request...\n"; } zmq::pollitem_t items[] = { { *d_socket, 0, ZMQ_POLLIN, 0 } }; zmq::poll (&items[0], 1, d_timeout); - //std::cout << "rx response...\n"; // If we got a reply, process if (items[0].revents & ZMQ_POLLIN) { - //std::cout << "rx response... got data\n"; - - // Receive data - zmq::message_t msg; - d_socket->recv(&msg); + // Receive data + zmq::message_t msg; + d_socket->recv(&msg); - //std::cout << "got msg...\n"; + std::string buf(static_cast<char*>(msg.data()), msg.size()); + std::stringbuf sb(buf); + pmt::pmt_t m = pmt::deserialize(sb); + message_port_pub(pmt::mp("out"), m); - std::string buf(static_cast<char*>(msg.data()), msg.size()); - std::stringbuf sb(buf); - pmt::pmt_t m = pmt::deserialize(sb); - //std::cout << m << "\n"; - message_port_pub(pmt::mp("out"), m); - - } else { - usleep(100); - } + } else { + usleep(100); } - } - - int - req_msg_source_impl::work(int noutput_items, - gr_vector_const_void_star &input_items, - gr_vector_void_star &output_items) - { - return noutput_items; + } } } /* namespace zeromq */ |