diff options
| author | Camilo Solano <solano@ti.rwth-aachen.de> | 2014-05-13 12:53:23 +0200 |
|---|---|---|
| committer | Camilo Solano <solano@ti.rwth-aachen.de> | 2014-05-13 12:53:23 +0200 |
| commit | 9970854f8938fcc015608dc163442467427cb9a6 (patch) | |
| tree | 31390248ae3252291b15fcb1a3a2819eaa5ce2c6 | |
| parent | 746737b4a9fc74bf4e86debdf35e67b8238da8b9 (diff) | |
Add sub_source block and fix pub_sink
| -rwxr-xr-x | gr-zeromq/examples/python/client.py | 3 | ||||
| -rwxr-xr-x | gr-zeromq/examples/python/server.py | 3 | ||||
| -rw-r--r-- | gr-zeromq/examples/zeromq_pubsub.grc | 481 | ||||
| -rw-r--r-- | gr-zeromq/grc/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | gr-zeromq/grc/zeromq_pub_sink.xml | 16 | ||||
| -rw-r--r-- | gr-zeromq/grc/zeromq_sub_source.xml | 65 | ||||
| -rw-r--r-- | gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | gr-zeromq/include/gnuradio/zeromq/pub_sink.h | 2 | ||||
| -rw-r--r-- | gr-zeromq/include/gnuradio/zeromq/pull_source.h | 2 | ||||
| -rw-r--r-- | gr-zeromq/include/gnuradio/zeromq/sub_source.h | 60 | ||||
| -rw-r--r-- | gr-zeromq/lib/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | gr-zeromq/lib/pub_sink_impl.cc | 13 | ||||
| -rw-r--r-- | gr-zeromq/lib/pub_sink_impl.h | 3 | ||||
| -rw-r--r-- | gr-zeromq/lib/sub_source_impl.cc | 102 | ||||
| -rw-r--r-- | gr-zeromq/lib/sub_source_impl.h | 54 | ||||
| -rw-r--r-- | gr-zeromq/swig/zeromq_swig.i | 3 |
16 files changed, 800 insertions, 10 deletions
diff --git a/gr-zeromq/examples/python/client.py b/gr-zeromq/examples/python/client.py index 3d55cc9a98..7663f9a056 100755 --- a/gr-zeromq/examples/python/client.py +++ b/gr-zeromq/examples/python/client.py @@ -50,8 +50,9 @@ class top_block(gr.top_block): source_adr = "tcp://"+self.options.servername+":5555" # blocks - self.zmq_source = zeromq.req_source(gr.sizeof_float, 1, source_adr) + #self.zmq_source = zeromq.req_source(gr.sizeof_float, 1, source_adr) #self.zmq_source = zeromq.pull_source(gr.sizeof_float, 1, source_adr) + self.zmq_source = zeromq.sub_source(gr.sizeof_float, 1, source_adr) #self.zmq_probe = zeromq.push_sink(gr.sizeof_float,probe_adr) self.zmq_probe = zeromq.pub_sink(gr.sizeof_float,probe_adr) diff --git a/gr-zeromq/examples/python/server.py b/gr-zeromq/examples/python/server.py index 28b732ed3e..358b66c568 100755 --- a/gr-zeromq/examples/python/server.py +++ b/gr-zeromq/examples/python/server.py @@ -56,7 +56,8 @@ class top_block(gr.top_block): self.gr_sig_source = analog.sig_source_f(samp_rate, analog.GR_SIN_WAVE , 1000, 1, 0) self.throttle = blocks.throttle(gr.sizeof_float, samp_rate) self.mult = blocks.multiply_const_ff(1) - self.zmq_sink = zeromq.rep_sink(gr.sizeof_float, 1, sink_adr) + #self.zmq_sink = zeromq.rep_sink(gr.sizeof_float, 1, sink_adr) + self.zmq_sink = zeromq.pub_sink(gr.sizeof_float, 1, sink_adr) #self.zmq_sink = zeromq.push_sink(gr.sizeof_float, 1, sink_adr) #self.zmq_probe = zeromq.push_sink(gr.sizeof_float, probe_adr) self.zmq_probe = zeromq.pub_sink(gr.sizeof_float, probe_adr) diff --git a/gr-zeromq/examples/zeromq_pubsub.grc b/gr-zeromq/examples/zeromq_pubsub.grc new file mode 100644 index 0000000000..89e98024c6 --- /dev/null +++ b/gr-zeromq/examples/zeromq_pubsub.grc @@ -0,0 +1,481 @@ +<?xml version='1.0' encoding='ASCII'?> +<flow_graph> + <timestamp>Tue May 13 12:47:08 2014</timestamp> + <block> + <key>options</key> + <param> + <key>id</key> + <value>top_block</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>title</key> + <value></value> + </param> + <param> + <key>author</key> + <value></value> + </param> + <param> + <key>description</key> + <value></value> + </param> + <param> + <key>window_size</key> + <value>1280, 1024</value> + </param> + <param> + <key>generate_options</key> + <value>wx_gui</value> + </param> + <param> + <key>category</key> + <value>Custom</value> + </param> + <param> + <key>run_options</key> + <value>prompt</value> + </param> + <param> + <key>run</key> + <value>True</value> + </param> + <param> + <key>max_nouts</key> + <value>0</value> + </param> + <param> + <key>realtime_scheduling</key> + <value></value> + </param> + <param> + <key>_coordinate</key> + <value>(10, 10)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>variable</key> + <param> + <key>id</key> + <value>samp_rate</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>value</key> + <value>32000</value> + </param> + <param> + <key>_coordinate</key> + <value>(10, 170)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>analog_sig_source_x</key> + <param> + <key>id</key> + <value>analog_sig_source_x_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>samp_rate</key> + <value>samp_rate</value> + </param> + <param> + <key>waveform</key> + <value>analog.GR_TRI_WAVE</value> + </param> + <param> + <key>freq</key> + <value>2000</value> + </param> + <param> + <key>amp</key> + <value>1</value> + </param> + <param> + <key>offset</key> + <value>0</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>minoutbuf</key> + <value>0</value> + </param> + <param> + <key>maxoutbuf</key> + <value>0</value> + </param> + <param> + <key>_coordinate</key> + <value>(170, 78)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>analog_fastnoise_source_x</key> + <param> + <key>id</key> + <value>analog_fastnoise_source_x_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>noise_type</key> + <value>analog.GR_GAUSSIAN</value> + </param> + <param> + <key>amp</key> + <value>1</value> + </param> + <param> + <key>seed</key> + <value>0</value> + </param> + <param> + <key>samples</key> + <value>8192</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>minoutbuf</key> + <value>0</value> + </param> + <param> + <key>maxoutbuf</key> + <value>0</value> + </param> + <param> + <key>_coordinate</key> + <value>(169, 196)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>blocks_add_xx</key> + <param> + <key>id</key> + <value>blocks_add_xx_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>num_inputs</key> + <value>2</value> + </param> + <param> + <key>vlen</key> + <value>1</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>minoutbuf</key> + <value>0</value> + </param> + <param> + <key>maxoutbuf</key> + <value>0</value> + </param> + <param> + <key>_coordinate</key> + <value>(403, 116)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>blocks_throttle</key> + <param> + <key>id</key> + <value>blocks_throttle_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>samples_per_second</key> + <value>samp_rate</value> + </param> + <param> + <key>vlen</key> + <value>1</value> + </param> + <param> + <key>ignoretag</key> + <value>True</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>minoutbuf</key> + <value>0</value> + </param> + <param> + <key>maxoutbuf</key> + <value>0</value> + </param> + <param> + <key>_coordinate</key> + <value>(537, 114)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>zeromq_sub_source</key> + <param> + <key>id</key> + <value>zeromq_sub_source_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>vlen</key> + <value>1</value> + </param> + <param> + <key>address</key> + <value>tcp://127.0.0.1:5555</value> + </param> + <param> + <key>timeout</key> + <value>100</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>minoutbuf</key> + <value>0</value> + </param> + <param> + <key>maxoutbuf</key> + <value>0</value> + </param> + <param> + <key>_coordinate</key> + <value>(146, 415)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>zeromq_pub_sink</key> + <param> + <key>id</key> + <value>zeromq_pub_sink_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>vlen</key> + <value>1</value> + </param> + <param> + <key>address</key> + <value>tcp://127.0.0.1:5555</value> + </param> + <param> + <key>timeout</key> + <value>100</value> + </param> + <param> + <key>blocking</key> + <value>True</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>_coordinate</key> + <value>(753, 111)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <block> + <key>wxgui_scopesink2</key> + <param> + <key>id</key> + <value>wxgui_scopesink2_0</value> + </param> + <param> + <key>_enabled</key> + <value>True</value> + </param> + <param> + <key>type</key> + <value>complex</value> + </param> + <param> + <key>title</key> + <value>Scope Plot</value> + </param> + <param> + <key>samp_rate</key> + <value>samp_rate</value> + </param> + <param> + <key>v_scale</key> + <value>0</value> + </param> + <param> + <key>v_offset</key> + <value>0</value> + </param> + <param> + <key>t_scale</key> + <value>0</value> + </param> + <param> + <key>ac_couple</key> + <value>False</value> + </param> + <param> + <key>xy_mode</key> + <value>False</value> + </param> + <param> + <key>num_inputs</key> + <value>1</value> + </param> + <param> + <key>win_size</key> + <value></value> + </param> + <param> + <key>grid_pos</key> + <value></value> + </param> + <param> + <key>notebook</key> + <value></value> + </param> + <param> + <key>trig_mode</key> + <value>wxgui.TRIG_MODE_AUTO</value> + </param> + <param> + <key>y_axis_label</key> + <value>Counts</value> + </param> + <param> + <key>affinity</key> + <value></value> + </param> + <param> + <key>_coordinate</key> + <value>(696, 407)</value> + </param> + <param> + <key>_rotation</key> + <value>0</value> + </param> + </block> + <connection> + <source_block_id>analog_sig_source_x_0</source_block_id> + <sink_block_id>blocks_add_xx_0</sink_block_id> + <source_key>0</source_key> + <sink_key>0</sink_key> + </connection> + <connection> + <source_block_id>analog_fastnoise_source_x_0</source_block_id> + <sink_block_id>blocks_add_xx_0</sink_block_id> + <source_key>0</source_key> + <sink_key>1</sink_key> + </connection> + <connection> + <source_block_id>blocks_add_xx_0</source_block_id> + <sink_block_id>blocks_throttle_0</sink_block_id> + <source_key>0</source_key> + <sink_key>0</sink_key> + </connection> + <connection> + <source_block_id>blocks_throttle_0</source_block_id> + <sink_block_id>zeromq_pub_sink_0</sink_block_id> + <source_key>0</source_key> + <sink_key>0</sink_key> + </connection> + <connection> + <source_block_id>zeromq_sub_source_0</source_block_id> + <sink_block_id>wxgui_scopesink2_0</sink_block_id> + <source_key>0</source_key> + <sink_key>0</sink_key> + </connection> +</flow_graph> diff --git a/gr-zeromq/grc/CMakeLists.txt b/gr-zeromq/grc/CMakeLists.txt index ad65f81d8e..548c2f28ed 100644 --- a/gr-zeromq/grc/CMakeLists.txt +++ b/gr-zeromq/grc/CMakeLists.txt @@ -19,6 +19,7 @@ install(FILES zeromq_pub_sink.xml + zeromq_sub_source.xml zeromq_push_sink.xml zeromq_pull_source.xml zeromq_rep_sink.xml diff --git a/gr-zeromq/grc/zeromq_pub_sink.xml b/gr-zeromq/grc/zeromq_pub_sink.xml index 64d26e1823..7f0a88691e 100644 --- a/gr-zeromq/grc/zeromq_pub_sink.xml +++ b/gr-zeromq/grc/zeromq_pub_sink.xml @@ -4,7 +4,7 @@ <key>zeromq_pub_sink</key> <category>ZeroMQ Interfaces</category> <import>from gnuradio import zeromq</import> - <make>zeromq.pub_sink($type.itemsize, $address, $blocking)</make> + <make>zeromq.pub_sink($type.itemsize, $vlen, $address, $blocking)</make> <param> <name>IO Type</name> @@ -38,12 +38,26 @@ </param> <param> + <name>Vec Length</name> + <key>vlen</key> + <value>1</value> + <type>int</type> + </param> + + <param> <name>Address</name> <key>address</key> <type>string</type> </param> <param> + <name>Timeout (msec)</name> + <key>timeout</key> + <value>0.1</value> + <type>float</type> + </param> + + <param> <name>Blocking</name> <key>blocking</key> <value>True</value> diff --git a/gr-zeromq/grc/zeromq_sub_source.xml b/gr-zeromq/grc/zeromq_sub_source.xml new file mode 100644 index 0000000000..307c981d44 --- /dev/null +++ b/gr-zeromq/grc/zeromq_sub_source.xml @@ -0,0 +1,65 @@ +<?xml version="1.0"?> +<block> + <name>ZMQ SUB Source</name> + <key>zeromq_sub_source</key> + <category>ZeroMQ Interfaces</category> + <import>from gnuradio import zeromq</import> + <make>zeromq.sub_source($type.itemsize, $vlen, $address, $timeout)</make> + + <param> + <name>IO Type</name> + <key>type</key> + <type>enum</type> + <option> + <name>Complex</name> + <key>complex</key> + <opt>itemsize:gr.sizeof_gr_complex</opt> + </option> + <option> + <name>Float</name> + <key>float</key> + <opt>itemsize:gr.sizeof_float</opt> + </option> + <option> + <name>Int</name> + <key>int</key> + <opt>itemsize:gr.sizeof_int</opt> + </option> + <option> + <name>Short</name> + <key>short</key> + <opt>itemsize:gr.sizeof_short</opt> + </option> + <option> + <name>Byte</name> + <key>byte</key> + <opt>itemsize:gr.sizeof_char</opt> + </option> + </param> + + <param> + <name>Vec Length</name> + <key>vlen</key> + <value>1</value> + <type>int</type> + </param> + + <param> + <name>Address</name> + <key>address</key> + <type>string</type> + </param> + + <param> + <name>Timeout (msec)</name> + <key>timeout</key> + <value>0.1</value> + <type>float</type> + </param> + + <source> + <name>out</name> + <type>$type</type> + </source> + +</block> diff --git a/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt b/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt index 87e79e021b..970cf1ce9a 100644 --- a/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt +++ b/gr-zeromq/include/gnuradio/zeromq/CMakeLists.txt @@ -23,6 +23,7 @@ install(FILES api.h pub_sink.h + sub_source.h pull_source.h push_sink.h rep_sink.h diff --git a/gr-zeromq/include/gnuradio/zeromq/pub_sink.h b/gr-zeromq/include/gnuradio/zeromq/pub_sink.h index 483461de5f..31a53648fa 100644 --- a/gr-zeromq/include/gnuradio/zeromq/pub_sink.h +++ b/gr-zeromq/include/gnuradio/zeromq/pub_sink.h @@ -53,7 +53,7 @@ namespace gr { * \param address ZMQ socket address specifier * \param blocking Indicate whether blocking sends should be used, default true. */ - static sptr make(size_t itemsize, size_t vlen, char *address); + static sptr make(size_t itemsize, size_t vlen, char *address, int timeout=100); }; } // namespace zeromq diff --git a/gr-zeromq/include/gnuradio/zeromq/pull_source.h b/gr-zeromq/include/gnuradio/zeromq/pull_source.h index 3a0c9432ff..5c1d37d353 100644 --- a/gr-zeromq/include/gnuradio/zeromq/pull_source.h +++ b/gr-zeromq/include/gnuradio/zeromq/pull_source.h @@ -34,7 +34,7 @@ namespace gr { * \ingroup zeromq * * \details - * This block will connect to a ZMQ PULL socket, then produce all + * This block will connect to a ZMQ PUSH socket, then produce all * incoming messages as streaming output. */ class ZEROMQ_API pull_source : virtual public gr::sync_block diff --git a/gr-zeromq/include/gnuradio/zeromq/sub_source.h b/gr-zeromq/include/gnuradio/zeromq/sub_source.h new file mode 100644 index 0000000000..9deaa7f3ff --- /dev/null +++ b/gr-zeromq/include/gnuradio/zeromq/sub_source.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_SUB_SOURCE_H +#define INCLUDED_ZEROMQ_SUB_SOURCE_H + +#include <gnuradio/zeromq/api.h> +#include <gnuradio/sync_block.h> + +namespace gr { + namespace zeromq { + + /*! + * \brief Receive messages on ZMQ SUB socket and source stream + * \ingroup zeromq + * + * \details + * This block will connect to a ZMQ PUB socket, then produce all + * incoming messages as streaming output. + */ + class ZEROMQ_API sub_source : virtual public gr::sync_block + { + public: + typedef boost::shared_ptr<sub_source> sptr; + + /*! + * \brief Return a shared_ptr to a new instance of gr::zeromq::sub_source. + * + * \param itemsize Size of a stream item in bytes + * \param vlen Vector length of the input items. Note that one vector is one item. + * \param address ZMQ socket address specifier + * \param timeout Receive timeout in seconds, default is 100ms, 1us increments + * + */ + static sptr make(size_t itemsize, size_t vlen, char *address, int timeout=100); + }; + + } // namespace zeromq +} // namespace gr + +#endif /* INCLUDED_ZEROMQ_SUB_SOURCE_H */ diff --git a/gr-zeromq/lib/CMakeLists.txt b/gr-zeromq/lib/CMakeLists.txt index 57a4c62230..cc62c79f35 100644 --- a/gr-zeromq/lib/CMakeLists.txt +++ b/gr-zeromq/lib/CMakeLists.txt @@ -39,6 +39,7 @@ endif(ENABLE_GR_CTRLPORT) ######################################################################## list(APPEND zeromq_sources pub_sink_impl.cc + sub_source_impl.cc pull_source_impl.cc push_sink_impl.cc rep_sink_impl.cc diff --git a/gr-zeromq/lib/pub_sink_impl.cc b/gr-zeromq/lib/pub_sink_impl.cc index 9d387f9a86..13f86045d7 100644 --- a/gr-zeromq/lib/pub_sink_impl.cc +++ b/gr-zeromq/lib/pub_sink_impl.cc @@ -31,18 +31,23 @@ namespace gr { namespace zeromq { pub_sink::sptr - pub_sink::make(size_t itemsize, size_t vlen, char *address) + pub_sink::make(size_t itemsize, size_t vlen, char *address, int timeout) { return gnuradio::get_initial_sptr - (new pub_sink_impl(itemsize, vlen, address)); + (new pub_sink_impl(itemsize, vlen, address, timeout)); } - pub_sink_impl::pub_sink_impl(size_t itemsize, size_t vlen, char *address) + pub_sink_impl::pub_sink_impl(size_t itemsize, size_t vlen, char *address, int timeout) : gr::sync_block("pub_sink", gr::io_signature::make(1, 1, itemsize * vlen), gr::io_signature::make(0, 0, 0)), - d_itemsize(itemsize), d_vlen(vlen) + d_itemsize(itemsize), d_vlen(vlen), 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_PUB); int time = 0; diff --git a/gr-zeromq/lib/pub_sink_impl.h b/gr-zeromq/lib/pub_sink_impl.h index bf5abdb8fe..9c956ef2fa 100644 --- a/gr-zeromq/lib/pub_sink_impl.h +++ b/gr-zeromq/lib/pub_sink_impl.h @@ -34,11 +34,12 @@ namespace gr { private: size_t d_itemsize; size_t d_vlen; + float d_timeout; zmq::context_t *d_context; zmq::socket_t *d_socket; public: - pub_sink_impl(size_t itemsize, size_t vlen, char *address); + pub_sink_impl(size_t itemsize, size_t vlen, char *address, int timeout); ~pub_sink_impl(); int work(int noutput_items, diff --git a/gr-zeromq/lib/sub_source_impl.cc b/gr-zeromq/lib/sub_source_impl.cc new file mode 100644 index 0000000000..38ddc78e59 --- /dev/null +++ b/gr-zeromq/lib/sub_source_impl.cc @@ -0,0 +1,102 @@ +/* -*- 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 "sub_source_impl.h" + +namespace gr { + namespace zeromq { + + sub_source::sptr + sub_source::make(size_t itemsize, size_t vlen, char *address, int timeout) + { + return gnuradio::get_initial_sptr + (new sub_source_impl(itemsize, vlen, address, timeout)); + } + + sub_source_impl::sub_source_impl(size_t itemsize, size_t vlen, char *address, int timeout) + : gr::sync_block("sub_source", + gr::io_signature::make(0, 0, 0), + gr::io_signature::make(1, 1, itemsize * vlen)), + d_itemsize(itemsize), d_vlen(vlen), 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_SUB); + //int time = 0; + d_socket->setsockopt(ZMQ_SUBSCRIBE, "", 0); + d_socket->connect (address); + } + + /* + * Our virtual destructor. + */ + sub_source_impl::~sub_source_impl() + { + d_socket->close(); + delete d_socket; + delete d_context; + } + + int + sub_source_impl::work(int noutput_items, + gr_vector_const_void_star &input_items, + gr_vector_void_star &output_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); + // Copy to ouput buffer and return + if (msg.size() >= d_itemsize*d_vlen*noutput_items) { + memcpy(out, (void *)msg.data(), d_itemsize*d_vlen*noutput_items); + + return noutput_items; + } + else { + memcpy(out, (void *)msg.data(), msg.size()); + + return msg.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_source_impl.h b/gr-zeromq/lib/sub_source_impl.h new file mode 100644 index 0000000000..44647527b1 --- /dev/null +++ b/gr-zeromq/lib/sub_source_impl.h @@ -0,0 +1,54 @@ +/* -*- 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_SUB_SOURCE_IMPL_H +#define INCLUDED_ZEROMQ_SUB_SOURCE_IMPL_H + +#include <gnuradio/zeromq/sub_source.h> +#include "zmq.hpp" + +namespace gr { + namespace zeromq { + + class sub_source_impl : public sub_source + { + private: + size_t d_itemsize; + size_t d_vlen; + int d_timeout; // microseconds, -1 is blocking + zmq::context_t *d_context; + zmq::socket_t *d_socket; + + public: + sub_source_impl(size_t itemsize, size_t vlen, char *address, int timeout); + ~sub_source_impl(); + + int work(int noutput_items, + gr_vector_const_void_star &input_items, + gr_vector_void_star &output_items); + }; + + } // namespace zeromq +} // namespace gr + +#endif /* INCLUDED_ZEROMQ_SUB_SOURCE_IMPL_H */ + diff --git a/gr-zeromq/swig/zeromq_swig.i b/gr-zeromq/swig/zeromq_swig.i index 36f8b2e279..bcd90550d4 100644 --- a/gr-zeromq/swig/zeromq_swig.i +++ b/gr-zeromq/swig/zeromq_swig.i @@ -31,6 +31,7 @@ #include "gnuradio/zeromq/pub_sink.h" #include "gnuradio/zeromq/push_sink.h" #include "gnuradio/zeromq/rep_sink.h" +#include "gnuradio/zeromq/sub_source.h" #include "gnuradio/zeromq/pull_source.h" #include "gnuradio/zeromq/req_source.h" %} @@ -38,11 +39,13 @@ %include "gnuradio/zeromq/pub_sink.h" %include "gnuradio/zeromq/push_sink.h" %include "gnuradio/zeromq/rep_sink.h" +%include "gnuradio/zeromq/sub_source.h" %include "gnuradio/zeromq/pull_source.h" %include "gnuradio/zeromq/req_source.h" GR_SWIG_BLOCK_MAGIC2(zeromq, pub_sink); GR_SWIG_BLOCK_MAGIC2(zeromq, push_sink); GR_SWIG_BLOCK_MAGIC2(zeromq, rep_sink); +GR_SWIG_BLOCK_MAGIC2(zeromq, sub_source); GR_SWIG_BLOCK_MAGIC2(zeromq, pull_source); GR_SWIG_BLOCK_MAGIC2(zeromq, req_source); |