[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Commit-gnuradio] [gnuradio] 15/29: zmq: building working versions of ad
|
From: |
git |
|
Subject: |
[Commit-gnuradio] [gnuradio] 15/29: zmq: building working versions of additional zmq message blocks in place (push/pull + rep/req) |
|
Date: |
Tue, 13 Jan 2015 01:04:28 +0000 (UTC) |
This is an automated email from the git hooks/post-receive script.
jcorgan pushed a commit to branch master
in repository gnuradio.
commit 1ee94434c6d81d04cc1d5435cd5bd8c2a5875f89
Author: Tim O'Shea <address@hidden>
Date: Tue Dec 30 17:48:09 2014 +0100
zmq: building working versions of additional zmq message blocks in place
(push/pull + rep/req)
---
gr-zeromq/grc/CMakeLists.txt | 4 +
gr-zeromq/grc/zeromq_pull_msg_source.xml | 28 +++++
gr-zeromq/grc/zeromq_push_msg_sink.xml | 28 +++++
gr-zeromq/grc/zeromq_rep_msg_sink.xml | 28 +++++
gr-zeromq/grc/zeromq_req_msg_source.xml | 28 +++++
gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt | 4 +
.../gnuradio/zeromq/pull_msg_source.h} | 49 +++++----
gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h | 62 +++++++++++
gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h | 60 ++++++++++
gr-zeromq/include/gnuradio/zeromq/req_msg_source.h | 59 ++++++++++
gr-zeromq/lib/CMakeLists.txt | 4 +
gr-zeromq/lib/pull_msg_source_impl.cc | 120 ++++++++++++++++++++
...ub_msg_source_impl.h => pull_msg_source_impl.h} | 30 ++---
gr-zeromq/lib/push_msg_sink_impl.cc | 119 ++++++++++++++++++++
...{sub_msg_source_impl.h => push_msg_sink_impl.h} | 33 +++---
gr-zeromq/lib/rep_msg_sink_impl.cc | 121 +++++++++++++++++++++
.../{sub_msg_source_impl.h => rep_msg_sink_impl.h} | 33 +++---
gr-zeromq/lib/req_msg_source_impl.cc | 121 +++++++++++++++++++++
...sub_msg_source_impl.h => req_msg_source_impl.h} | 32 +++---
gr-zeromq/lib/sub_msg_source_impl.cc | 2 +
gr-zeromq/lib/sub_msg_source_impl.h | 1 -
gr-zeromq/swig/zeromq_swig.i | 12 ++
22 files changed, 878 insertions(+), 100 deletions(-)
diff --git a/gr-zeromq/grc/CMakeLists.txt b/gr-zeromq/grc/CMakeLists.txt
index 95d7d6c..7807bcf 100644
--- a/gr-zeromq/grc/CMakeLists.txt
+++ b/gr-zeromq/grc/CMakeLists.txt
@@ -23,9 +23,13 @@ install(FILES
zeromq_sub_source.xml
zeromq_sub_msg_source.xml
zeromq_push_sink.xml
+ zeromq_push_msg_sink.xml
zeromq_pull_source.xml
+ zeromq_pull_msg_source.xml
zeromq_rep_sink.xml
+ zeromq_rep_msg_sink.xml
zeromq_req_source.xml
+ zeromq_req_msg_source.xml
DESTINATION share/gnuradio/grc/blocks
)
diff --git a/gr-zeromq/grc/zeromq_pull_msg_source.xml
b/gr-zeromq/grc/zeromq_pull_msg_source.xml
new file mode 100644
index 0000000..6b4fddd
--- /dev/null
+++ b/gr-zeromq/grc/zeromq_pull_msg_source.xml
@@ -0,0 +1,28 @@
+<?xml version="1.0"?>
+<block>
+ <name>ZMQ PULL Source</name>
+ <key>zeromq_pull_msg_source</key>
+ <category>ZeroMQ Interfaces</category>
+ <import>from gnuradio import zeromq</import>
+ <make>zeromq.pull_msg_source($address, $timeout)</make>
+
+ <param>
+ <name>Address</name>
+ <key>address</key>
+ <type>string</type>
+ </param>
+
+ <param>
+ <name>Timeout (msec)</name>
+ <key>timeout</key>
+ <value>100</value>
+ <type>float</type>
+ </param>
+
+ <source>
+ <name>out</name>
+ <type>message</type>
+ <optional>1</optional>
+ </source>
+
+</block>
diff --git a/gr-zeromq/grc/zeromq_push_msg_sink.xml
b/gr-zeromq/grc/zeromq_push_msg_sink.xml
new file mode 100644
index 0000000..28544f0
--- /dev/null
+++ b/gr-zeromq/grc/zeromq_push_msg_sink.xml
@@ -0,0 +1,28 @@
+<?xml version="1.0"?>
+<block>
+ <name>ZMQ PUSH Sink</name>
+ <key>zeromq_push_msg_sink</key>
+ <category>ZeroMQ Interfaces</category>
+ <import>from gnuradio import zeromq</import>
+ <make>zeromq.push_msg_sink($address, $timeout)</make>
+
+ <param>
+ <name>Address</name>
+ <key>address</key>
+ <type>string</type>
+ </param>
+
+ <param>
+ <name>Timeout (msec)</name>
+ <key>timeout</key>
+ <value>100</value>
+ <type>float</type>
+ </param>
+
+ <sink>
+ <name>in</name>
+ <type>message</type>
+ <optional>1</optional>
+ </sink>
+
+</block>
diff --git a/gr-zeromq/grc/zeromq_rep_msg_sink.xml
b/gr-zeromq/grc/zeromq_rep_msg_sink.xml
new file mode 100644
index 0000000..0eb6c06
--- /dev/null
+++ b/gr-zeromq/grc/zeromq_rep_msg_sink.xml
@@ -0,0 +1,28 @@
+<?xml version="1.0"?>
+<block>
+ <name>ZMQ REP Sink</name>
+ <key>zeromq_rep_msg_sink</key>
+ <category>ZeroMQ Interfaces</category>
+ <import>from gnuradio import zeromq</import>
+ <make>zeromq.rep_msg_sink($address, $timeout)</make>
+
+ <param>
+ <name>Address</name>
+ <key>address</key>
+ <type>string</type>
+ </param>
+
+ <param>
+ <name>Timeout (msec)</name>
+ <key>timeout</key>
+ <value>100</value>
+ <type>float</type>
+ </param>
+
+ <sink>
+ <name>in</name>
+ <type>message</type>
+ <optional>1</optional>
+ </sink>
+
+</block>
diff --git a/gr-zeromq/grc/zeromq_req_msg_source.xml
b/gr-zeromq/grc/zeromq_req_msg_source.xml
new file mode 100644
index 0000000..1d40bf5
--- /dev/null
+++ b/gr-zeromq/grc/zeromq_req_msg_source.xml
@@ -0,0 +1,28 @@
+<?xml version="1.0"?>
+<block>
+ <name>ZMQ REQ Source</name>
+ <key>zeromq_req_msg_source</key>
+ <category>ZeroMQ Interfaces</category>
+ <import>from gnuradio import zeromq</import>
+ <make>zeromq.req_msg_source($address, $timeout)</make>
+
+ <param>
+ <name>Address</name>
+ <key>address</key>
+ <type>string</type>
+ </param>
+
+ <param>
+ <name>Timeout (msec)</name>
+ <key>timeout</key>
+ <value>100</value>
+ <type>float</type>
+ </param>
+
+ <source>
+ <name>out</name>
+ <type>message</type>
+ <optional>1</optional>
+ </source>
+
+</block>
diff --git a/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt
b/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt
index c8d155f..d365532 100644
--- a/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt
+++ b/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt
@@ -27,9 +27,13 @@ install(FILES
sub_source.h
sub_msg_source.h
pull_source.h
+ pull_msg_source.h
push_sink.h
+ push_msg_sink.h
rep_sink.h
+ rep_msg_sink.h
req_source.h
+ req_msg_source.h
DESTINATION ${GR_INCLUDE_DIR}/gnuradio/zeromq
COMPONENT "zeromq_devel"
diff --git a/gr-zeromq/lib/sub_msg_source_impl.h
b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h
similarity index 50%
copy from gr-zeromq/lib/sub_msg_source_impl.h
copy to gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h
index 0db5440..7d87ba1 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.h
+++ b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h
@@ -20,38 +20,39 @@
* Boston, MA 02110-1301, USA.
*/
-#ifndef INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
-#define INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
+#ifndef INCLUDED_ZEROMQ_PULL_MSG_SOURCE_H
+#define INCLUDED_ZEROMQ_PULL_MSG_SOURCE_H
-#include <gnuradio/zeromq/sub_msg_source.h>
-#include "zmq.hpp"
+#include <gnuradio/zeromq/api.h>
+#include <gnuradio/sync_block.h>
namespace gr {
namespace zeromq {
- class sub_msg_source_impl : public sub_msg_source
+ /*!
+ * \brief Receive messages on ZMQ PULL socket and source stream
+ * \ingroup zeromq
+ *
+ * \details
+ * This block will connect to a ZMQ PUSH socket, then produce all
+ * incoming messages as streaming output.
+ */
+ class ZEROMQ_API pull_msg_source : virtual public gr::sync_block
{
- private:
- int d_timeout; // microseconds, -1 is blocking
- zmq::context_t *d_context;
- zmq::socket_t *d_socket;
- void readloop();
- boost::thread *d_thread;
-
- public:
- sub_msg_source_impl(char *address, int timeout);
- ~sub_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);
+ public:
+ typedef boost::shared_ptr<pull_msg_source> sptr;
+
+ /*!
+ * \brief Return a shared_ptr to a new instance of
gr::zeromq::pull_msg_source.
+ *
+ * \param address ZMQ socket address specifier
+ * \param timeout Receive timeout in seconds, default is 100ms, 1us
increments
+ *
+ */
+ static sptr make(char *address, int timeout=100);
};
} // namespace zeromq
} // namespace gr
-#endif /* INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H */
-
+#endif /* INCLUDED_ZEROMQ_PULL_MSG_SOURCE_H */
diff --git a/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h
b/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h
new file mode 100644
index 0000000..27a19b2
--- /dev/null
+++ b/gr-zeromq/include/gnuradio/zeromq/push_msg_sink.h
@@ -0,0 +1,62 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013,2014 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef INCLUDED_ZEROMQ_PUSH_MSG_SINK_H
+#define INCLUDED_ZEROMQ_PUSH_MSG_SINK_H
+
+#include <gnuradio/zeromq/api.h>
+#include <gnuradio/sync_block.h>
+
+namespace gr {
+ namespace zeromq {
+
+ /*!
+ * \brief Sink the contents of a stream to a ZMQ PUSH socket
+ * \ingroup zeromq
+ *
+ * \details
+ * This block acts a a streaming sink for a GNU Radio flowgraph
+ * and writes its contents to a ZMQ PUSH socket. A PUSH socket
+ * will round-robin send its messages to each connected ZMQ PULL
+ * socket, either another gr-zeromq source block or a regular,
+ * non-GNU Radio ZMQ socket.
+ *
+ */
+ class ZEROMQ_API push_msg_sink : virtual public gr::sync_block
+ {
+ public:
+ typedef boost::shared_ptr<push_msg_sink> sptr;
+
+ /*!
+ * \brief Return a shared_ptr to a new instance of
gr::zeromq::push_msg_sink
+ *
+ * \param address ZMQ socket address specifier
+ * \param timeout Receive timeout in seconds, default is 100ms, 1us
increments
+ *
+ */
+ static sptr make(char *address, int timeout=100);
+ };
+
+ } // namespace zeromq
+} // namespace gr
+
+#endif /* INCLUDED_ZEROMQ_PUSH_MSG_SINK_H */
diff --git a/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h
b/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h
new file mode 100644
index 0000000..b0fd23c
--- /dev/null
+++ b/gr-zeromq/include/gnuradio/zeromq/rep_msg_sink.h
@@ -0,0 +1,60 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013,2014 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef INCLUDED_ZEROMQ_REP_MSG_SINK_H
+#define INCLUDED_ZEROMQ_REP_MSG_SINK_H
+
+#include <gnuradio/zeromq/api.h>
+#include <gnuradio/sync_block.h>
+
+namespace gr {
+ namespace zeromq {
+
+ /*!
+ * \brief Sink the contents of a stream to a ZMQ REP socket
+ * \ingroup zeromq
+ *
+ * \details
+ * This block acts a a streaming sink for a GNU Radio flowgraph
+ * and writes its contents to a ZMQ REP socket. A REP socket will
+ * only send its contents to an attached REQ socket when it
+ * requests items.
+ */
+ class ZEROMQ_API rep_msg_sink : virtual public gr::sync_block
+ {
+ public:
+ typedef boost::shared_ptr<rep_msg_sink> sptr;
+
+ /*!
+ * \brief Return a shared_ptr to a new instance of zeromq::rep_msg_sink.
+ *
+ * \param address ZMQ socket address specifier
+ * \param timeout Receive timeout in seconds, default is 100ms, 1us
increments
+ *
+ */
+ static sptr make(char *address, int timeout=100);
+ };
+
+ } // namespace zeromq
+} // namespace gr
+
+#endif /* INCLUDED_ZEROMQ_REP_MSG_SINK_H */
diff --git a/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h
b/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h
new file mode 100644
index 0000000..cf91833
--- /dev/null
+++ b/gr-zeromq/include/gnuradio/zeromq/req_msg_source.h
@@ -0,0 +1,59 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef INCLUDED_ZEROMQ_REQ_MSG_SOURCE_H
+#define INCLUDED_ZEROMQ_REQ_MSG_SOURCE_H
+
+#include <gnuradio/zeromq/api.h>
+#include <gnuradio/sync_block.h>
+
+namespace gr {
+ namespace zeromq {
+
+ /*!
+ * \brief Receive messages on ZMQ REQ socket and source stream
+ * \ingroup zeromq
+ *
+ * \details
+ * This block will connect to a ZMQ REP socket, then produce all
+ * incoming messages as streaming output.
+ */
+ class ZEROMQ_API req_msg_source : virtual public gr::sync_block
+ {
+ public:
+ typedef boost::shared_ptr<req_msg_source> sptr;
+
+ /*!
+ * \brief Return a shared_ptr to a new instance of
zeromq::req_msg_source.
+ *
+ *
+ * \param address ZMQ socket address specifier
+ * \param timeout Receive timeout in seconds, default is 100ms, 1us
increments
+ *
+ */
+ static sptr make(char *address, int timeout=100);
+ };
+
+ } // namespace zeromq
+} // namespace gr
+
+#endif /* INCLUDED_ZEROMQ_SOURCE_REQREP_H */
diff --git a/gr-zeromq/lib/CMakeLists.txt b/gr-zeromq/lib/CMakeLists.txt
index 0964d81..941e5ff 100644
--- a/gr-zeromq/lib/CMakeLists.txt
+++ b/gr-zeromq/lib/CMakeLists.txt
@@ -42,9 +42,13 @@ list(APPEND zeromq_sources
sub_source_impl.cc
sub_msg_source_impl.cc
pull_source_impl.cc
+ pull_msg_source_impl.cc
push_sink_impl.cc
+ push_msg_sink_impl.cc
rep_sink_impl.cc
+ rep_msg_sink_impl.cc
req_source_impl.cc
+ req_msg_source_impl.cc
tag_headers.cc
)
diff --git a/gr-zeromq/lib/pull_msg_source_impl.cc
b/gr-zeromq/lib/pull_msg_source_impl.cc
new file mode 100644
index 0000000..5b207ce
--- /dev/null
+++ b/gr-zeromq/lib/pull_msg_source_impl.cc
@@ -0,0 +1,120 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013,2014 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gnuradio/io_signature.h>
+#include "pull_msg_source_impl.h"
+#include "tag_headers.h"
+
+namespace gr {
+ namespace zeromq {
+
+ pull_msg_source::sptr
+ pull_msg_source::make(char *address, int timeout)
+ {
+ return gnuradio::get_initial_sptr
+ (new pull_msg_source_impl(address, timeout));
+ }
+
+ 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)),
+ 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);
+
+ message_port_register_out(pmt::mp("out"));
+ }
+
+ pull_msg_source_impl::~pull_msg_source_impl()
+ {
+ d_socket->close();
+ delete d_socket;
+ delete d_context;
+ }
+
+ 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;
+ /*
+ char *out = (char*)output_items[0];
+
+ 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);
+
+ // check header for tags...
+ std::string buf(static_cast<char*>(msg.data()), msg.size());
+ if(d_pass_tags){
+ uint64_t rcv_offset;
+ std::vector<gr::tag_t> tags;
+ buf = parse_tag_header(buf, rcv_offset, tags);
+ for(size_t i=0; i<tags.size(); i++){
+ tags[i].offset -= rcv_offset - nitems_written(0);
+ add_item_tag(0, tags[i]);
+ }
+ }
+
+
+ // Copy to ouput buffer and return
+ if (buf.size() >= d_itemsize*d_vlen*noutput_items) {
+ memcpy(out, (void *)&buf[0], d_itemsize*d_vlen*noutput_items);
+
+ return noutput_items;
+ }
+ else {
+ memcpy(out, (void *)&buf[0], buf.size());
+
+ return buf.size()/(d_itemsize*d_vlen);
+ }
+ }
+ else {
+ return 0; // FIXME: someday when the scheduler does all the
poll/selects
+ }
+
+ */
+ }
+
+ } /* namespace zeromq */
+} /* namespace gr */
diff --git a/gr-zeromq/lib/sub_msg_source_impl.h
b/gr-zeromq/lib/pull_msg_source_impl.h
similarity index 64%
copy from gr-zeromq/lib/sub_msg_source_impl.h
copy to gr-zeromq/lib/pull_msg_source_impl.h
index 0db5440..fb9237c 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.h
+++ b/gr-zeromq/lib/pull_msg_source_impl.h
@@ -20,38 +20,32 @@
* Boston, MA 02110-1301, USA.
*/
-#ifndef INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
-#define INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
+#ifndef INCLUDED_ZEROMQ_PULL_MSG_SOURCE_IMPL_H
+#define INCLUDED_ZEROMQ_PULL_MSG_SOURCE_IMPL_H
-#include <gnuradio/zeromq/sub_msg_source.h>
-#include "zmq.hpp"
+#include <gnuradio/zeromq/pull_msg_source.h>
+#include <zmq.hpp>
namespace gr {
namespace zeromq {
- class sub_msg_source_impl : public sub_msg_source
+ class pull_msg_source_impl : public pull_msg_source
{
- private:
+ private:
int d_timeout; // microseconds, -1 is blocking
zmq::context_t *d_context;
zmq::socket_t *d_socket;
- void readloop();
- boost::thread *d_thread;
- public:
- sub_msg_source_impl(char *address, int timeout);
- ~sub_msg_source_impl();
+ public:
+ 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);
+ gr_vector_const_void_star &input_items,
+ gr_vector_void_star &output_items);
};
} // namespace zeromq
} // namespace gr
-#endif /* INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H */
-
+#endif /* INCLUDED_ZEROMQ_PULL_MSG_SOURCE_IMPL_H */
diff --git a/gr-zeromq/lib/push_msg_sink_impl.cc
b/gr-zeromq/lib/push_msg_sink_impl.cc
new file mode 100644
index 0000000..6266cd6
--- /dev/null
+++ b/gr-zeromq/lib/push_msg_sink_impl.cc
@@ -0,0 +1,119 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013,2014 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gnuradio/io_signature.h>
+#include "push_msg_sink_impl.h"
+#include "tag_headers.h"
+
+namespace gr {
+ namespace zeromq {
+
+ push_msg_sink::sptr
+ push_msg_sink::make(char *address, int timeout)
+ {
+ return gnuradio::get_initial_sptr
+ (new push_msg_sink_impl(address, timeout));
+ }
+
+ 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)),
+ 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);
+
+ message_port_register_in(pmt::mp("in"));
+ set_msg_handler( pmt::mp("in"),
+ boost::bind(&push_msg_sink_impl::handler, this, _1));
+ }
+
+ push_msg_sink_impl::~push_msg_sink_impl()
+ {
+ d_socket->close();
+ delete d_socket;
+ delete d_context;
+ }
+
+ 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;
+ }*/
+ }
+
+ } /* namespace zeromq */
+} /* namespace gr */
diff --git a/gr-zeromq/lib/sub_msg_source_impl.h
b/gr-zeromq/lib/push_msg_sink_impl.h
similarity index 61%
copy from gr-zeromq/lib/sub_msg_source_impl.h
copy to gr-zeromq/lib/push_msg_sink_impl.h
index 0db5440..b77c998 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.h
+++ b/gr-zeromq/lib/push_msg_sink_impl.h
@@ -20,38 +20,33 @@
* Boston, MA 02110-1301, USA.
*/
-#ifndef INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
-#define INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
+#ifndef INCLUDED_ZEROMQ_PUSH_MSG_SINK_IMPL_H
+#define INCLUDED_ZEROMQ_PUSH_MSG_SINK_IMPL_H
-#include <gnuradio/zeromq/sub_msg_source.h>
-#include "zmq.hpp"
+#include <gnuradio/zeromq/push_msg_sink.h>
+#include <zmq.hpp>
namespace gr {
namespace zeromq {
- class sub_msg_source_impl : public sub_msg_source
+ class push_msg_sink_impl : public push_msg_sink
{
- private:
- int d_timeout; // microseconds, -1 is blocking
+ private:
+ float d_timeout;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
- void readloop();
- boost::thread *d_thread;
- public:
- sub_msg_source_impl(char *address, int timeout);
- ~sub_msg_source_impl();
+ public:
+ push_msg_sink_impl(char *address, int timeout);
+ ~push_msg_sink_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);
+ gr_vector_const_void_star &input_items,
+ gr_vector_void_star &output_items);
+ void handler(pmt::pmt_t msg);
};
} // namespace zeromq
} // namespace gr
-#endif /* INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H */
-
+#endif /* INCLUDED_ZEROMQ_ZMQ_PUSH_MSG_SINK_IMPL_H */
diff --git a/gr-zeromq/lib/rep_msg_sink_impl.cc
b/gr-zeromq/lib/rep_msg_sink_impl.cc
new file mode 100644
index 0000000..72dd5fb
--- /dev/null
+++ b/gr-zeromq/lib/rep_msg_sink_impl.cc
@@ -0,0 +1,121 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013,2014 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gnuradio/io_signature.h>
+#include "rep_msg_sink_impl.h"
+#include "tag_headers.h"
+
+namespace gr {
+ namespace zeromq {
+
+ rep_msg_sink::sptr
+ rep_msg_sink::make(char *address, int timeout)
+ {
+ return gnuradio::get_initial_sptr
+ (new rep_msg_sink_impl(address, timeout));
+ }
+
+ rep_msg_sink_impl::rep_msg_sink_impl(char *address, int timeout)
+ : gr::sync_block("rep_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_REP);
+ int time = 0;
+ d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time));
+ d_socket->bind (address);
+
+ message_port_register_in(pmt::mp("in"));
+ set_msg_handler( pmt::mp("in"),
+ boost::bind(&rep_msg_sink_impl::handler, this, _1));
+ }
+
+ rep_msg_sink_impl::~rep_msg_sink_impl()
+ {
+ d_socket->close();
+ delete d_socket;
+ delete d_context;
+ }
+
+ void rep_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
+ rep_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 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 request
+ zmq::message_t request;
+ d_socket->recv(&request);
+ int req_output_items = *(static_cast<int*>(request.data()));
+ int nitems_send = std::min(noutput_items, req_output_items);
+
+ // 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*nitems_send);
+ 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*nitems_send);
+ d_socket->send(msg);
+
+ return nitems_send;
+ }
+
+ return 0;*/
+ }
+ } /* namespace zeromq */
+} /* namespace gr */
diff --git a/gr-zeromq/lib/sub_msg_source_impl.h
b/gr-zeromq/lib/rep_msg_sink_impl.h
similarity index 61%
copy from gr-zeromq/lib/sub_msg_source_impl.h
copy to gr-zeromq/lib/rep_msg_sink_impl.h
index 0db5440..40d7969 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.h
+++ b/gr-zeromq/lib/rep_msg_sink_impl.h
@@ -20,38 +20,33 @@
* Boston, MA 02110-1301, USA.
*/
-#ifndef INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
-#define INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
+#ifndef INCLUDED_ZEROMQ_REP_MSG_SINK_IMPL_H
+#define INCLUDED_ZEROMQ_REP_MSG_SINK_IMPL_H
-#include <gnuradio/zeromq/sub_msg_source.h>
-#include "zmq.hpp"
+#include <gnuradio/zeromq/rep_msg_sink.h>
+#include <zmq.hpp>
namespace gr {
namespace zeromq {
- class sub_msg_source_impl : public sub_msg_source
+ class rep_msg_sink_impl : public rep_msg_sink
{
- private:
- int d_timeout; // microseconds, -1 is blocking
+ private:
+ int d_timeout;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
- void readloop();
- boost::thread *d_thread;
- public:
- sub_msg_source_impl(char *address, int timeout);
- ~sub_msg_source_impl();
+ public:
+ rep_msg_sink_impl(char *address, int timeout);
+ ~rep_msg_sink_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);
+ gr_vector_const_void_star &input_items,
+ gr_vector_void_star &output_items);
+ void handler(pmt::pmt_t msg);
};
} // namespace zeromq
} // namespace gr
-#endif /* INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H */
-
+#endif /* INCLUDED_ZEROMQ_REP_MSG_SINK_IMPL_H */
diff --git a/gr-zeromq/lib/req_msg_source_impl.cc
b/gr-zeromq/lib/req_msg_source_impl.cc
new file mode 100644
index 0000000..92e26e9
--- /dev/null
+++ b/gr-zeromq/lib/req_msg_source_impl.cc
@@ -0,0 +1,121 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013,2014 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio.
+ *
+ * This is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this software; see the file COPYING. If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gnuradio/io_signature.h>
+#include "req_msg_source_impl.h"
+#include "tag_headers.h"
+
+namespace gr {
+ namespace zeromq {
+
+ req_msg_source::sptr
+ req_msg_source::make(char *address, int timeout)
+ {
+ return gnuradio::get_initial_sptr
+ (new req_msg_source_impl(address, timeout));
+ }
+
+ 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)
+ {
+ 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_REQ);
+ int time = 0;
+ d_socket->setsockopt(ZMQ_LINGER, &time, sizeof(time));
+ d_socket->connect (address);
+
+ message_port_register_out(pmt::mp("out"));
+ }
+
+ req_msg_source_impl::~req_msg_source_impl()
+ {
+ d_socket->close();
+ delete d_socket;
+ delete d_context;
+ }
+
+ 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;
+/*
+ char *out = (char*)output_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) {
+ // Request data, FIXME non portable?
+ zmq::message_t request(sizeof(int));
+ memcpy ((void *) request.data (), &noutput_items, sizeof(int));
+ d_socket->send(request);
+ }
+
+ zmq::pollitem_t itemsin[] = { { *d_socket, 0, ZMQ_POLLIN, 0 } };
+ zmq::poll (&itemsin[0], 1, d_timeout);
+
+ // If we got a reply, process
+ if (itemsin[0].revents & ZMQ_POLLIN) {
+ // Receive data
+ zmq::message_t reply;
+ d_socket->recv(&reply);
+
+ // Deserialize header data / tags
+ std::string buf(static_cast<char*>(reply.data()), reply.size());
+
+ if(d_pass_tags){
+ uint64_t rcv_offset;
+ std::vector<gr::tag_t> tags;
+ buf = parse_tag_header(buf, rcv_offset, tags);
+ for(size_t i=0; i<tags.size(); i++){
+ tags[i].offset -= rcv_offset - nitems_written(0);
+ add_item_tag(0, tags[i]);
+ }
+ }
+
+
+ // Copy to ouput buffer and return
+ memcpy(out, (void *)&buf[0], buf.size());
+ return buf.size()/(d_itemsize*d_vlen);
+ }
+
+ return 0;
+ */
+ }
+
+ } /* namespace zeromq */
+} /* namespace gr */
diff --git a/gr-zeromq/lib/sub_msg_source_impl.h
b/gr-zeromq/lib/req_msg_source_impl.h
similarity index 61%
copy from gr-zeromq/lib/sub_msg_source_impl.h
copy to gr-zeromq/lib/req_msg_source_impl.h
index 0db5440..635fa45 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.h
+++ b/gr-zeromq/lib/req_msg_source_impl.h
@@ -20,38 +20,32 @@
* Boston, MA 02110-1301, USA.
*/
-#ifndef INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
-#define INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H
+#ifndef INCLUDED_ZEROMQ_REQ_MSG_SOURCE_IMPL_H
+#define INCLUDED_ZEROMQ_REQ_MSG_SOURCE_IMPL_H
-#include <gnuradio/zeromq/sub_msg_source.h>
-#include "zmq.hpp"
+#include <gnuradio/zeromq/req_msg_source.h>
+#include <zmq.hpp>
namespace gr {
namespace zeromq {
- class sub_msg_source_impl : public sub_msg_source
+ class req_msg_source_impl : public req_msg_source
{
- private:
- int d_timeout; // microseconds, -1 is blocking
+ private:
+ int d_timeout;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
- void readloop();
- boost::thread *d_thread;
- public:
- sub_msg_source_impl(char *address, int timeout);
- ~sub_msg_source_impl();
+ public:
+ req_msg_source_impl(char *address, int timeout);
+ ~req_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);
+ gr_vector_const_void_star &input_items,
+ gr_vector_void_star &output_items);
};
} // namespace zeromq
} // namespace gr
-#endif /* INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H */
-
+#endif /* INCLUDED_ZEROMQ_REQ_MSG_SOURCE_IMPL_H */
diff --git a/gr-zeromq/lib/sub_msg_source_impl.cc
b/gr-zeromq/lib/sub_msg_source_impl.cc
index 1c63f2d..f7a9bc9 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.cc
+++ b/gr-zeromq/lib/sub_msg_source_impl.cc
@@ -68,11 +68,13 @@ namespace gr {
bool sub_msg_source_impl::start(){
d_finished = false;
d_thread = new boost::thread( boost::bind(
&sub_msg_source_impl::readloop , this ) );
+ return true;
}
bool sub_msg_source_impl::stop(){
d_finished = true;
d_thread->join();
+ return true;
}
void sub_msg_source_impl::readloop(){
diff --git a/gr-zeromq/lib/sub_msg_source_impl.h
b/gr-zeromq/lib/sub_msg_source_impl.h
index 0db5440..424eb47 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.h
+++ b/gr-zeromq/lib/sub_msg_source_impl.h
@@ -54,4 +54,3 @@ namespace gr {
} // namespace gr
#endif /* INCLUDED_ZEROMQ_SUB_MSG_SOURCE_IMPL_H */
-
diff --git a/gr-zeromq/swig/zeromq_swig.i b/gr-zeromq/swig/zeromq_swig.i
index c9a891d..78669bd 100644
--- a/gr-zeromq/swig/zeromq_swig.i
+++ b/gr-zeromq/swig/zeromq_swig.i
@@ -31,27 +31,39 @@
#include "gnuradio/zeromq/pub_sink.h"
#include "gnuradio/zeromq/pub_msg_sink.h"
#include "gnuradio/zeromq/push_sink.h"
+#include "gnuradio/zeromq/push_msg_sink.h"
#include "gnuradio/zeromq/rep_sink.h"
+#include "gnuradio/zeromq/rep_msg_sink.h"
#include "gnuradio/zeromq/sub_source.h"
#include "gnuradio/zeromq/sub_msg_source.h"
#include "gnuradio/zeromq/pull_source.h"
+#include "gnuradio/zeromq/pull_msg_source.h"
#include "gnuradio/zeromq/req_source.h"
+#include "gnuradio/zeromq/req_msg_source.h"
%}
%include "gnuradio/zeromq/pub_sink.h"
%include "gnuradio/zeromq/pub_msg_sink.h"
%include "gnuradio/zeromq/push_sink.h"
+%include "gnuradio/zeromq/push_msg_sink.h"
%include "gnuradio/zeromq/rep_sink.h"
+%include "gnuradio/zeromq/rep_msg_sink.h"
%include "gnuradio/zeromq/sub_source.h"
%include "gnuradio/zeromq/sub_msg_source.h"
%include "gnuradio/zeromq/pull_source.h"
+%include "gnuradio/zeromq/pull_msg_source.h"
%include "gnuradio/zeromq/req_source.h"
+%include "gnuradio/zeromq/req_msg_source.h"
GR_SWIG_BLOCK_MAGIC2(zeromq, pub_sink);
GR_SWIG_BLOCK_MAGIC2(zeromq, pub_msg_sink);
GR_SWIG_BLOCK_MAGIC2(zeromq, push_sink);
+GR_SWIG_BLOCK_MAGIC2(zeromq, push_msg_sink);
GR_SWIG_BLOCK_MAGIC2(zeromq, rep_sink);
+GR_SWIG_BLOCK_MAGIC2(zeromq, rep_msg_sink);
GR_SWIG_BLOCK_MAGIC2(zeromq, sub_source);
GR_SWIG_BLOCK_MAGIC2(zeromq, sub_msg_source);
GR_SWIG_BLOCK_MAGIC2(zeromq, pull_source);
+GR_SWIG_BLOCK_MAGIC2(zeromq, pull_msg_source);
GR_SWIG_BLOCK_MAGIC2(zeromq, req_source);
+GR_SWIG_BLOCK_MAGIC2(zeromq, req_msg_source);
- [Commit-gnuradio] [gnuradio] 29/29: Merge branch 'maint', (continued)
- [Commit-gnuradio] [gnuradio] 29/29: Merge branch 'maint', git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 26/29: zeromq: added msg passing example, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 03/29: zmq: default to not pass tags (compatible wire format), git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 02/29: zmq: tags should now be serializing and deserializing correctly for pub_sink/sub_source, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 27/29: zeromq: minor cleanup, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 25/29: zeromq: cleanup and made req_msg_source derive from gr::block, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 09/29: zmq: include pass_tags option default to false in all blocks' grc definitions, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 18/29: zmq: rep/req msg blocks now working, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 05/29: zmq: hoisting header encoding to helper function, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 04/29: zmq: adding pass tags option to zmq pub/sub block grc xml defs, git, 2015/01/12
- [Commit-gnuradio] [gnuradio] 15/29: zmq: building working versions of additional zmq message blocks in place (push/pull + rep/req),
git <=
- [Commit-gnuradio] [gnuradio] 28/29: zeromq: added stream tag passing example, git, 2015/01/12