diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2015-01-12 14:24:48 -0800 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2015-01-12 14:24:48 -0800 |
| commit | 3a443cfe307c4f2c5f03806146d4ad1664faa288 (patch) | |
| tree | 294dd854a825dbed9e209720d1bbfa01ed41381a /gr-zeromq | |
| parent | 7b77431d3a4971ac62c6e71a813151404c1ae5e3 (diff) | |
zeromq: cleanup and made pull_msg_source derive from gr::block
Diffstat (limited to 'gr-zeromq')
| -rw-r--r-- | gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h | 10 | ||||
| -rw-r--r-- | gr-zeromq/lib/pull_msg_source_impl.cc | 59 | ||||
| -rw-r--r-- | gr-zeromq/lib/pull_msg_source_impl.h | 9 |
3 files changed, 35 insertions, 43 deletions
diff --git a/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h index 7d87ba1574..17495152b8 100644 --- a/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h +++ b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h @@ -24,20 +24,20 @@ #define INCLUDED_ZEROMQ_PULL_MSG_SOURCE_H #include <gnuradio/zeromq/api.h> -#include <gnuradio/sync_block.h> +#include <gnuradio/block.h> namespace gr { namespace zeromq { /*! - * \brief Receive messages on ZMQ PULL socket and source stream + * \brief Receive messages on ZMQ PULL socket and output async messages * \ingroup zeromq * * \details - * This block will connect to a ZMQ PUSH socket, then produce all - * incoming messages as streaming output. + * This block will connect to a ZMQ PUSH socket, then convert + * received messages to outgoing async messages. */ - class ZEROMQ_API pull_msg_source : virtual public gr::sync_block + class ZEROMQ_API pull_msg_source : virtual public gr::block { public: typedef boost::shared_ptr<pull_msg_source> sptr; diff --git a/gr-zeromq/lib/pull_msg_source_impl.cc b/gr-zeromq/lib/pull_msg_source_impl.cc index 0c848fc771..ca496ef216 100644 --- a/gr-zeromq/lib/pull_msg_source_impl.cc +++ b/gr-zeromq/lib/pull_msg_source_impl.cc @@ -39,18 +39,21 @@ namespace gr { } pull_msg_source_impl::pull_msg_source_impl(char *address, int timeout) - : gr::sync_block("pull_msg_source", - gr::io_signature::make(0, 0, 0), - gr::io_signature::make(0, 0, 0)), + : gr::block("pull_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); + if (major < 3) { d_timeout = timeout*1000; } + d_context = new zmq::context_t(1); d_socket = new zmq::socket_t(*d_context, ZMQ_PULL); + int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); d_socket->connect (address); @@ -65,53 +68,43 @@ namespace gr { delete d_context; } - bool pull_msg_source_impl::start(){ + bool pull_msg_source_impl::start() + { d_finished = false; - d_thread = new boost::thread( boost::bind( &pull_msg_source_impl::readloop , this ) ); + d_thread = new boost::thread(boost::bind(&pull_msg_source_impl::readloop, this)); return true; } - bool pull_msg_source_impl::stop(){ + bool pull_msg_source_impl::stop() + { d_finished = true; - d_thread->join(); + d_thread->join(); return true; } - void pull_msg_source_impl::readloop(){ + void pull_msg_source_impl::readloop() + { while(!d_finished){ - //std::cout << "readloop\n"; - + zmq::pollitem_t items[] = { { *d_socket, 0, ZMQ_POLLIN, 0 } }; zmq::poll (&items[0], 1, d_timeout); // If we got a reply, process if (items[0].revents & ZMQ_POLLIN) { - // 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); - //std::cout << m << "\n"; - message_port_pub(pmt::mp("out"), m); - - } else { - usleep(100); - } - } - } + // Receive data + zmq::message_t msg; + d_socket->recv(&msg); + 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); - int - pull_msg_source_impl::work(int noutput_items, - gr_vector_const_void_star &input_items, - gr_vector_void_star &output_items) - { - return noutput_items; + } else { + usleep(100); + } + } } } /* namespace zeromq */ diff --git a/gr-zeromq/lib/pull_msg_source_impl.h b/gr-zeromq/lib/pull_msg_source_impl.h index 9ff89ef154..6d8791d613 100644 --- a/gr-zeromq/lib/pull_msg_source_impl.h +++ b/gr-zeromq/lib/pull_msg_source_impl.h @@ -35,19 +35,18 @@ namespace gr { int d_timeout; // microseconds, -1 is blocking zmq::context_t *d_context; zmq::socket_t *d_socket; + boost::thread *d_thread; + void readloop(); - boost::thread *d_thread; public: + bool d_finished; + pull_msg_source_impl(char *address, int timeout); ~pull_msg_source_impl(); bool start(); bool stop(); - bool d_finished; - int work(int noutput_items, - gr_vector_const_void_star &input_items, - gr_vector_void_star &output_items); }; } // namespace zeromq |