diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2017-04-13 09:39:32 -0700 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2017-04-13 09:39:32 -0700 |
| commit | 3a11ee086e712fb4c5077340a682aa2774610e8a (patch) | |
| tree | 1c1c5b27b6dd1863e992a70070c0d4ee46612690 /gr-zeromq | |
| parent | ea748f66010376ee46c6cbe659d9b162e1f51cc8 (diff) | |
| parent | cb4ab189ce58333f6e5262a9972788864f003807 (diff) | |
Merge branch 'next' into python3
Diffstat (limited to 'gr-zeromq')
| -rw-r--r-- | gr-zeromq/lib/base_impl.cc | 17 | ||||
| -rwxr-xr-x | gr-zeromq/python/zeromq/qa_zeromq_sub.py | 81 |
2 files changed, 96 insertions, 2 deletions
diff --git a/gr-zeromq/lib/base_impl.cc b/gr-zeromq/lib/base_impl.cc index f33315dd40..76baeafff2 100644 --- a/gr-zeromq/lib/base_impl.cc +++ b/gr-zeromq/lib/base_impl.cc @@ -165,6 +165,11 @@ namespace gr { if (!(items[0].revents & ZMQ_POLLIN)) return false; + /* Is this the start or continuation of a multi-part message? */ + int64_t more = 0; + size_t more_len = sizeof(more); + d_socket->getsockopt(ZMQ_RCVMORE, &more, &more_len); + /* Reset */ d_msg.rebuild(); d_tags.clear(); @@ -174,8 +179,8 @@ namespace gr { /* Get the message */ d_socket->recv(&d_msg); - /* Parse header */ - if (d_pass_tags) + /* Parse header from the first (or only) message of a multi-part message */ + if (d_pass_tags && !more) { uint64_t rcv_offset; @@ -188,6 +193,14 @@ namespace gr { } } + /* Each message must contain an integer mutliple of data vectors */ + if ((d_msg.size() - d_consumed_bytes) % d_vsize != 0) + { + throw std::runtime_error( + boost::str(boost::format("Incompatible vector sizes: " + "need a multiple of %1% bytes per message") % d_vsize)); + } + /* We got one ! */ return true; } diff --git a/gr-zeromq/python/zeromq/qa_zeromq_sub.py b/gr-zeromq/python/zeromq/qa_zeromq_sub.py new file mode 100755 index 0000000000..f916901ff7 --- /dev/null +++ b/gr-zeromq/python/zeromq/qa_zeromq_sub.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright 2014 Free Software Foundation, Inc. +# +# This file is part of GNU Radio +# +# GNU Radio 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. +# +# GNU Radio 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 GNU Radio; see the file COPYING. If not, write to +# the Free Software Foundation, Inc., 51 Franklin Street, +# Boston, MA 02110-1301, USA. +# + +from gnuradio import gr, gr_unittest +from gnuradio import blocks, zeromq +from gnuradio import eng_notation + +import numpy +import time +import zmq + +class qa_zeromq_sub (gr_unittest.TestCase): + + def setUp (self): + self.tb = gr.top_block () + self.zmq_context = zmq.Context() + self.pub_socket = self.zmq_context.socket(zmq.PUB) + self.pub_socket.bind("tcp://127.0.0.1:5555") + + def tearDown (self): + self.pub_socket.close() + self.zmq_context.term() + self.tb = None + + def test_001 (self): + vlen = 10 + src_data = numpy.array(range(vlen)*100, 'float32') + zeromq_sub_source = zeromq.sub_source(gr.sizeof_float, vlen, "tcp://127.0.0.1:5555") + sink = blocks.vector_sink_f(vlen) + self.tb.connect(zeromq_sub_source, sink) + + self.tb.start() + self.pub_socket.send(src_data.tostring()) + time.sleep(0.25) + self.tb.stop() + self.tb.wait() + self.assertFloatTuplesAlmostEqual(sink.data(), src_data) + + def test_002 (self): + vlen = 10 + + # Construct multipart source data to publish + raw_data = [numpy.array(range(vlen)*100, 'float32'), numpy.array(range(vlen, 2*vlen)*100, 'float32')] + src_data = [a.tostring() for a in raw_data] + zeromq_sub_source = zeromq.sub_source(gr.sizeof_float, vlen, "tcp://127.0.0.1:5555") + sink = blocks.vector_sink_f(vlen) + self.tb.connect(zeromq_sub_source, sink) + + self.tb.start() + self.pub_socket.send_multipart(src_data) + time.sleep(0.25) + self.tb.stop() + self.tb.wait() + + # Source block will concatenate everything together + expected_data = numpy.concatenate(raw_data) + self.assertFloatTuplesAlmostEqual(sink.data(), expected_data) + + +if __name__ == '__main__': + gr_unittest.run(qa_zeromq_sub) |