diff options
24 files changed, 123 insertions, 42 deletions
diff --git a/gr-zeromq/grc/zeromq_pub_msg_sink.block.yml b/gr-zeromq/grc/zeromq_pub_msg_sink.block.yml index 306a954796..5f1e9f38cd 100644 --- a/gr-zeromq/grc/zeromq_pub_msg_sink.block.yml +++ b/gr-zeromq/grc/zeromq_pub_msg_sink.block.yml @@ -11,6 +11,12 @@ parameters: label: Timeout (msec) dtype: float default: '100' +- id: bind + category: Advanced + label: Bind + dtype: bool + default: 'True' + hide: part inputs: - domain: message @@ -19,7 +25,7 @@ inputs: templates: imports: from gnuradio import zeromq - make: zeromq.pub_msg_sink(${address}, ${timeout}) + make: zeromq.pub_msg_sink(${address}, ${timeout}, ${bind}) cpp_templates: includes: [ '#include <gnuradio/zeromq/pub_msg_sink.h>' ] diff --git a/gr-zeromq/grc/zeromq_pull_msg_source.block.yml b/gr-zeromq/grc/zeromq_pull_msg_source.block.yml index ad4fa53e78..70e60dff62 100644 --- a/gr-zeromq/grc/zeromq_pull_msg_source.block.yml +++ b/gr-zeromq/grc/zeromq_pull_msg_source.block.yml @@ -11,6 +11,12 @@ parameters: label: Timeout (msec) dtype: float default: '100' +- id: bind + category: Advanced + label: Bind + dtype: bool + default: 'False' + hide: part outputs: - domain: message @@ -19,7 +25,7 @@ outputs: templates: imports: from gnuradio import zeromq - make: zeromq.pull_msg_source(${address}, ${timeout}) + make: zeromq.pull_msg_source(${address}, ${timeout}, ${bind}) cpp_templates: includes: [ '#include <gnuradio/zeromq/pull_msg_source.h>' ] diff --git a/gr-zeromq/grc/zeromq_push_msg_sink.block.yml b/gr-zeromq/grc/zeromq_push_msg_sink.block.yml index 87599f7a6d..8f5b89304f 100644 --- a/gr-zeromq/grc/zeromq_push_msg_sink.block.yml +++ b/gr-zeromq/grc/zeromq_push_msg_sink.block.yml @@ -11,6 +11,12 @@ parameters: label: Timeout (msec) dtype: float default: '100' +- id: bind + category: Advanced + label: Bind + dtype: bool + default: 'True' + hide: part inputs: - domain: message @@ -19,7 +25,7 @@ inputs: templates: imports: from gnuradio import zeromq - make: zeromq.push_msg_sink(${address}, ${timeout}) + make: zeromq.push_msg_sink(${address}, ${timeout}, ${bind}) cpp_templates: includes: [ '#include <gnuradio/zeromq/push_msg_sink.h>' ] diff --git a/gr-zeromq/grc/zeromq_rep_msg_sink.block.yml b/gr-zeromq/grc/zeromq_rep_msg_sink.block.yml index 9a01131f27..0a89fc81e0 100644 --- a/gr-zeromq/grc/zeromq_rep_msg_sink.block.yml +++ b/gr-zeromq/grc/zeromq_rep_msg_sink.block.yml @@ -11,6 +11,12 @@ parameters: label: Timeout (msec) dtype: float default: '100' +- id: bind + category: Advanced + label: Bind + dtype: bool + default: 'True' + hide: part inputs: - domain: message @@ -19,7 +25,7 @@ inputs: templates: imports: from gnuradio import zeromq - make: zeromq.rep_msg_sink(${address}, ${timeout}) + make: zeromq.rep_msg_sink(${address}, ${timeout}, ${bind}) cpp_templates: includes: [ '#include <gnuradio/zeromq/rep_msg_sink.h>' ] diff --git a/gr-zeromq/grc/zeromq_req_msg_source.block.yml b/gr-zeromq/grc/zeromq_req_msg_source.block.yml index 25a25a2c63..4a681cf6e3 100644 --- a/gr-zeromq/grc/zeromq_req_msg_source.block.yml +++ b/gr-zeromq/grc/zeromq_req_msg_source.block.yml @@ -11,6 +11,12 @@ parameters: label: Timeout (msec) dtype: float default: '100' +- id: bind + category: Advanced + label: Bind + dtype: bool + default: 'False' + hide: part outputs: - domain: message @@ -19,7 +25,7 @@ outputs: templates: imports: from gnuradio import zeromq - make: zeromq.req_msg_source(${address}, ${timeout}) + make: zeromq.req_msg_source(${address}, ${timeout}, ${bind}) cpp_templates: includes: [ '#include <gnuradio/zeromq/req_msg_source.h>' ] diff --git a/gr-zeromq/grc/zeromq_sub_msg_source.block.yml b/gr-zeromq/grc/zeromq_sub_msg_source.block.yml index 55f0323e5e..aabb7f9bfe 100644 --- a/gr-zeromq/grc/zeromq_sub_msg_source.block.yml +++ b/gr-zeromq/grc/zeromq_sub_msg_source.block.yml @@ -11,6 +11,12 @@ parameters: label: Timeout (msec) dtype: float default: '100' +- id: bind + category: Advanced + label: Bind + dtype: bool + default: 'False' + hide: part outputs: - domain: message @@ -19,7 +25,7 @@ outputs: templates: imports: from gnuradio import zeromq - make: zeromq.sub_msg_source(${address}, ${timeout}) + make: zeromq.sub_msg_source(${address}, ${timeout}, ${bind}) cpp_templates: includes: [ '#include <gnuradio/zeromq/sub_msg_source.h>' ] diff --git a/gr-zeromq/include/gnuradio/zeromq/pub_msg_sink.h b/gr-zeromq/include/gnuradio/zeromq/pub_msg_sink.h index 6bccc19001..a1dbdd53e0 100644 --- a/gr-zeromq/include/gnuradio/zeromq/pub_msg_sink.h +++ b/gr-zeromq/include/gnuradio/zeromq/pub_msg_sink.h @@ -38,8 +38,11 @@ public: * * \param address ZMQ socket address specifier * \param timeout Receive timeout in milliseconds, default is 100ms, 1us increments + * \param bind If true this block will bind to the address, otherwise it will + * connect; the default is to bind + * */ - static sptr make(char* address, int timeout = 100); + static sptr make(char* address, int timeout = 100, bool bind = true); /*! * \brief Return a std::string of ZMQ_LAST_ENDPOINT from the underlying ZMQ socket. diff --git a/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h index 5a4e1a88a2..ec6ec3c5ae 100644 --- a/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h +++ b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h @@ -35,9 +35,11 @@ public: * * \param address ZMQ socket address specifier * \param timeout Receive timeout in milliseconds, default is 100ms, 1us increments + * \param bind If true this block will bind to the address, otherwise it will + * connect; the default is to connect * */ - static sptr make(char* address, int timeout = 100); + static sptr make(char* address, int timeout = 100, bool bind = false); /*! * \brief Return a std::string of ZMQ_LAST_ENDPOINT from the underlying ZMQ socket. diff --git a/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h b/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h index 41c5fba042..5cd3218015 100644 --- a/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h +++ b/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h @@ -37,9 +37,11 @@ public: * * \param address ZMQ socket address specifier * \param timeout Receive timeout in milliseconds, default is 100ms, 1us increments + * \param bind If true this block will bind to the address, otherwise it will + * connect; the default is to bind * */ - static sptr make(char* address, int timeout = 100); + static sptr make(char* address, int timeout = 100, bool bind = true); /*! * \brief Return a std::string of ZMQ_LAST_ENDPOINT from the underlying ZMQ socket. diff --git a/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h b/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h index 4c3d54adac..1bcac43c26 100644 --- a/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h +++ b/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h @@ -37,9 +37,11 @@ public: * * \param address ZMQ socket address specifier * \param timeout Receive timeout in milliseconds, default is 100ms, 1us increments + * \param bind If true this block will bind to the address, otherwise it will + * connect; the default is to bind * */ - static sptr make(char* address, int timeout = 100); + static sptr make(char* address, int timeout = 100, bool bind = true); /*! * \brief Return a std::string of ZMQ_LAST_ENDPOINT from the underlying ZMQ socket. diff --git a/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h b/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h index 09d316004a..908229a1d8 100644 --- a/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h +++ b/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h @@ -35,9 +35,11 @@ public: * * \param address ZMQ socket address specifier * \param timeout Receive timeout in milliseconds, default is 100ms, 1us increments + * \param bind If true this block will bind to the address, otherwise it will + * connect; the default is to connect * */ - static sptr make(char* address, int timeout = 100); + static sptr make(char* address, int timeout = 100, bool bind = false); /*! * \brief Return a std::string of ZMQ_LAST_ENDPOINT from the underlying ZMQ socket. diff --git a/gr-zeromq/include/gnuradio/zeromq/sub_msg_source.h b/gr-zeromq/include/gnuradio/zeromq/sub_msg_source.h index 3d0cdc525b..c3e82d825a 100644 --- a/gr-zeromq/include/gnuradio/zeromq/sub_msg_source.h +++ b/gr-zeromq/include/gnuradio/zeromq/sub_msg_source.h @@ -35,9 +35,11 @@ public: * * \param address ZMQ socket address specifier * \param timeout Receive timeout in milliseconds, default is 100ms, 1us increments + * \param bind If true this block will bind to the address, otherwise it will + * connect; the default is to connect * */ - static sptr make(char* address, int timeout = 100); + static sptr make(char* address, int timeout = 100, bool bind = false); /*! * \brief Return a std::string of ZMQ_LAST_ENDPOINT from the underlying ZMQ socket. diff --git a/gr-zeromq/lib/pub_msg_sink_impl.cc b/gr-zeromq/lib/pub_msg_sink_impl.cc index d4a0179f09..ca78a8ea8d 100644 --- a/gr-zeromq/lib/pub_msg_sink_impl.cc +++ b/gr-zeromq/lib/pub_msg_sink_impl.cc @@ -19,12 +19,12 @@ namespace gr { namespace zeromq { -pub_msg_sink::sptr pub_msg_sink::make(char* address, int timeout) +pub_msg_sink::sptr pub_msg_sink::make(char* address, int timeout, bool bind) { - return gnuradio::get_initial_sptr(new pub_msg_sink_impl(address, timeout)); + return gnuradio::get_initial_sptr(new pub_msg_sink_impl(address, timeout, bind)); } -pub_msg_sink_impl::pub_msg_sink_impl(char* address, int timeout) +pub_msg_sink_impl::pub_msg_sink_impl(char* address, int timeout, bool bind) : gr::block("pub_msg_sink", gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), @@ -32,15 +32,22 @@ pub_msg_sink_impl::pub_msg_sink_impl(char* address, int 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_PUB); + int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); - d_socket->bind(address); + + if (bind) { + d_socket->bind(address); + } else { + d_socket->connect(address); + } message_port_register_in(pmt::mp("in")); set_msg_handler(pmt::mp("in"), boost::bind(&pub_msg_sink_impl::handler, this, _1)); diff --git a/gr-zeromq/lib/pub_msg_sink_impl.h b/gr-zeromq/lib/pub_msg_sink_impl.h index 6ee13ccf0b..1c278ce70a 100644 --- a/gr-zeromq/lib/pub_msg_sink_impl.h +++ b/gr-zeromq/lib/pub_msg_sink_impl.h @@ -25,7 +25,7 @@ private: zmq::socket_t* d_socket; public: - pub_msg_sink_impl(char* address, int timeout); + pub_msg_sink_impl(char* address, int timeout, bool bind); ~pub_msg_sink_impl(); void handler(pmt::pmt_t msg); diff --git a/gr-zeromq/lib/pull_msg_source_impl.cc b/gr-zeromq/lib/pull_msg_source_impl.cc index 651bc530fa..0520ba5abc 100644 --- a/gr-zeromq/lib/pull_msg_source_impl.cc +++ b/gr-zeromq/lib/pull_msg_source_impl.cc @@ -21,12 +21,12 @@ namespace gr { namespace zeromq { -pull_msg_source::sptr pull_msg_source::make(char* address, int timeout) +pull_msg_source::sptr pull_msg_source::make(char* address, int timeout, bool bind) { - return gnuradio::get_initial_sptr(new pull_msg_source_impl(address, timeout)); + return gnuradio::get_initial_sptr(new pull_msg_source_impl(address, timeout, bind)); } -pull_msg_source_impl::pull_msg_source_impl(char* address, int timeout) +pull_msg_source_impl::pull_msg_source_impl(char* address, int timeout, bool bind) : gr::block("pull_msg_source", gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), @@ -45,7 +45,12 @@ pull_msg_source_impl::pull_msg_source_impl(char* address, int timeout) int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); - d_socket->connect(address); + + if (bind) { + d_socket->bind(address); + } else { + d_socket->connect(address); + } message_port_register_out(d_port); } diff --git a/gr-zeromq/lib/pull_msg_source_impl.h b/gr-zeromq/lib/pull_msg_source_impl.h index 9115f3f1a2..4d84b11119 100644 --- a/gr-zeromq/lib/pull_msg_source_impl.h +++ b/gr-zeromq/lib/pull_msg_source_impl.h @@ -31,7 +31,7 @@ private: public: bool d_finished; - pull_msg_source_impl(char* address, int timeout); + pull_msg_source_impl(char* address, int timeout, bool bind); ~pull_msg_source_impl(); bool start() override; diff --git a/gr-zeromq/lib/push_msg_sink_impl.cc b/gr-zeromq/lib/push_msg_sink_impl.cc index a194c03b80..b9964ce6ab 100644 --- a/gr-zeromq/lib/push_msg_sink_impl.cc +++ b/gr-zeromq/lib/push_msg_sink_impl.cc @@ -19,12 +19,12 @@ namespace gr { namespace zeromq { -push_msg_sink::sptr push_msg_sink::make(char* address, int timeout) +push_msg_sink::sptr push_msg_sink::make(char* address, int timeout, bool bind) { - return gnuradio::get_initial_sptr(new push_msg_sink_impl(address, timeout)); + return gnuradio::get_initial_sptr(new push_msg_sink_impl(address, timeout, bind)); } -push_msg_sink_impl::push_msg_sink_impl(char* address, int timeout) +push_msg_sink_impl::push_msg_sink_impl(char* address, int timeout, bool bind) : gr::block("push_msg_sink", gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), @@ -42,7 +42,12 @@ push_msg_sink_impl::push_msg_sink_impl(char* address, int timeout) int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); - d_socket->bind(address); + + if (bind) { + d_socket->bind(address); + } else { + d_socket->connect(address); + } message_port_register_in(pmt::mp("in")); set_msg_handler(pmt::mp("in"), boost::bind(&push_msg_sink_impl::handler, this, _1)); diff --git a/gr-zeromq/lib/push_msg_sink_impl.h b/gr-zeromq/lib/push_msg_sink_impl.h index dd417a96b8..a197e68569 100644 --- a/gr-zeromq/lib/push_msg_sink_impl.h +++ b/gr-zeromq/lib/push_msg_sink_impl.h @@ -25,7 +25,7 @@ private: zmq::socket_t* d_socket; public: - push_msg_sink_impl(char* address, int timeout); + push_msg_sink_impl(char* address, int timeout, bool bind); ~push_msg_sink_impl(); void handler(pmt::pmt_t msg); diff --git a/gr-zeromq/lib/rep_msg_sink_impl.cc b/gr-zeromq/lib/rep_msg_sink_impl.cc index 8fec3c3ca6..5bf7d634ce 100644 --- a/gr-zeromq/lib/rep_msg_sink_impl.cc +++ b/gr-zeromq/lib/rep_msg_sink_impl.cc @@ -19,12 +19,12 @@ namespace gr { namespace zeromq { -rep_msg_sink::sptr rep_msg_sink::make(char* address, int timeout) +rep_msg_sink::sptr rep_msg_sink::make(char* address, int timeout, bool bind) { - return gnuradio::get_initial_sptr(new rep_msg_sink_impl(address, timeout)); + return gnuradio::get_initial_sptr(new rep_msg_sink_impl(address, timeout, bind)); } -rep_msg_sink_impl::rep_msg_sink_impl(char* address, int timeout) +rep_msg_sink_impl::rep_msg_sink_impl(char* address, int timeout, bool bind) : gr::block("rep_msg_sink", gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), @@ -43,7 +43,12 @@ rep_msg_sink_impl::rep_msg_sink_impl(char* address, int timeout) int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); - d_socket->bind(address); + + if (bind) { + d_socket->bind(address); + } else { + d_socket->connect(address); + } message_port_register_in(d_port); } diff --git a/gr-zeromq/lib/rep_msg_sink_impl.h b/gr-zeromq/lib/rep_msg_sink_impl.h index c3e04c472d..d42bc8077b 100644 --- a/gr-zeromq/lib/rep_msg_sink_impl.h +++ b/gr-zeromq/lib/rep_msg_sink_impl.h @@ -31,7 +31,7 @@ private: void readloop(); public: - rep_msg_sink_impl(char* address, int timeout); + rep_msg_sink_impl(char* address, int timeout, bool bind); ~rep_msg_sink_impl(); bool start() override; diff --git a/gr-zeromq/lib/req_msg_source_impl.cc b/gr-zeromq/lib/req_msg_source_impl.cc index 0118b4d084..d55829a815 100644 --- a/gr-zeromq/lib/req_msg_source_impl.cc +++ b/gr-zeromq/lib/req_msg_source_impl.cc @@ -21,12 +21,12 @@ namespace gr { namespace zeromq { -req_msg_source::sptr req_msg_source::make(char* address, int timeout) +req_msg_source::sptr req_msg_source::make(char* address, int timeout, bool bind) { - return gnuradio::get_initial_sptr(new req_msg_source_impl(address, timeout)); + return gnuradio::get_initial_sptr(new req_msg_source_impl(address, timeout, bind)); } -req_msg_source_impl::req_msg_source_impl(char* address, int timeout) +req_msg_source_impl::req_msg_source_impl(char* address, int timeout, bool bind) : gr::block("req_msg_source", gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), @@ -45,7 +45,12 @@ req_msg_source_impl::req_msg_source_impl(char* address, int timeout) int time = 0; d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time)); - d_socket->connect(address); + + if (bind) { + d_socket->bind(address); + } else { + d_socket->connect(address); + } message_port_register_out(d_port); } diff --git a/gr-zeromq/lib/req_msg_source_impl.h b/gr-zeromq/lib/req_msg_source_impl.h index 3696ae176f..3ea44c632d 100644 --- a/gr-zeromq/lib/req_msg_source_impl.h +++ b/gr-zeromq/lib/req_msg_source_impl.h @@ -31,7 +31,7 @@ private: public: bool d_finished; - req_msg_source_impl(char* address, int timeout); + req_msg_source_impl(char* address, int timeout, bool bind); ~req_msg_source_impl(); bool start() override; diff --git a/gr-zeromq/lib/sub_msg_source_impl.cc b/gr-zeromq/lib/sub_msg_source_impl.cc index 3ead4025f1..281c5b1995 100644 --- a/gr-zeromq/lib/sub_msg_source_impl.cc +++ b/gr-zeromq/lib/sub_msg_source_impl.cc @@ -21,12 +21,12 @@ namespace gr { namespace zeromq { -sub_msg_source::sptr sub_msg_source::make(char* address, int timeout) +sub_msg_source::sptr sub_msg_source::make(char* address, int timeout, bool bind) { - return gnuradio::get_initial_sptr(new sub_msg_source_impl(address, timeout)); + return gnuradio::get_initial_sptr(new sub_msg_source_impl(address, timeout, bind)); } -sub_msg_source_impl::sub_msg_source_impl(char* address, int timeout) +sub_msg_source_impl::sub_msg_source_impl(char* address, int timeout, bool bind) : gr::block("sub_msg_source", gr::io_signature::make(0, 0, 0), gr::io_signature::make(0, 0, 0)), @@ -44,7 +44,12 @@ sub_msg_source_impl::sub_msg_source_impl(char* address, int timeout) d_socket = new zmq::socket_t(*d_context, ZMQ_SUB); d_socket->setsockopt(ZMQ_SUBSCRIBE, "", 0); - d_socket->connect(address); + + if (bind) { + d_socket->bind(address); + } else { + d_socket->connect(address); + } message_port_register_out(d_port); } diff --git a/gr-zeromq/lib/sub_msg_source_impl.h b/gr-zeromq/lib/sub_msg_source_impl.h index d0779f33d2..6707a217de 100644 --- a/gr-zeromq/lib/sub_msg_source_impl.h +++ b/gr-zeromq/lib/sub_msg_source_impl.h @@ -31,7 +31,7 @@ private: public: bool d_finished; - sub_msg_source_impl(char* address, int timeout); + sub_msg_source_impl(char* address, int timeout, bool bind); ~sub_msg_source_impl(); bool start() override; |