summaryrefslogtreecommitdiff
path: root/gr-zeromq/lib
diff options
context:
space:
mode:
authorJacob Gilbert <mrjacobagilbert@gmail.com>2020-04-10 18:54:14 -0600
committerMarcus Müller <marcus@hostalia.de>2020-04-10 21:45:25 +0200
commitc17b2dabf050698532ab38c5460d44bab853ee76 (patch)
tree62937133ba1ecaf8769d4cb6b86195a16b87379e /gr-zeromq/lib
parentdaf936e44c85a36e76663f4bb82c6a8756e13812 (diff)
zeromq: Add bind argument for message based zeromq blocks.
This is needed to support N:1 patterns. To avoid confusion, since this is probably a feature only needed by those who Know What They Are Doing, the option is added to the Advanced block property tab.
Diffstat (limited to 'gr-zeromq/lib')
-rw-r--r--gr-zeromq/lib/pub_msg_sink_impl.cc15
-rw-r--r--gr-zeromq/lib/pub_msg_sink_impl.h2
-rw-r--r--gr-zeromq/lib/pull_msg_source_impl.cc13
-rw-r--r--gr-zeromq/lib/pull_msg_source_impl.h2
-rw-r--r--gr-zeromq/lib/push_msg_sink_impl.cc13
-rw-r--r--gr-zeromq/lib/push_msg_sink_impl.h2
-rw-r--r--gr-zeromq/lib/rep_msg_sink_impl.cc13
-rw-r--r--gr-zeromq/lib/rep_msg_sink_impl.h2
-rw-r--r--gr-zeromq/lib/req_msg_source_impl.cc13
-rw-r--r--gr-zeromq/lib/req_msg_source_impl.h2
-rw-r--r--gr-zeromq/lib/sub_msg_source_impl.cc13
-rw-r--r--gr-zeromq/lib/sub_msg_source_impl.h2
12 files changed, 62 insertions, 30 deletions
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;
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-25 13:05:31 +0000