diff options
| author | Tim O'Shea <tim.oshea753@gmail.com> | 2014-12-30 17:48:09 +0100 |
|---|---|---|
| committer | Tim O'Shea <tim.oshea753@gmail.com> | 2014-12-30 17:48:09 +0100 |
| commit | 1ee94434c6d81d04cc1d5435cd5bd8c2a5875f89 (patch) | |
| tree | 0a6349a845ebe7e50c680afee0037dae42971e00 | |
| parent | cd7e604d527f6324138f778dd15253d78df49350 (diff) | |
zmq: building working versions of additional zmq message blocks in place (push/pull + rep/req)
22 files changed, 1064 insertions, 1 deletions
diff --git a/gr-zeromq/grc/CMakeLists.txt b/gr-zeromq/grc/CMakeLists.txt index 95d7d6c3e9..7807bcfe9d 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 0000000000..6b4fddd6a3 --- /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 0000000000..28544f0084 --- /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 0000000000..0eb6c06729 --- /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 0000000000..1d40bf5e10 --- /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 c8d155f1d8..d365532003 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/include/gnuradio/zeromq/pull_msg_source.h b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h new file mode 100644 index 0000000000..7d87ba1574 --- /dev/null +++ b/gr-zeromq/include/gnuradio/zeromq/pull_msg_source.h @@ -0,0 +1,58 @@ +/* -*- 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_PULL_MSG_SOURCE_H +#define INCLUDED_ZEROMQ_PULL_MSG_SOURCE_H + +#include <gnuradio/zeromq/api.h> +#include <gnuradio/sync_block.h> + +namespace gr { + namespace zeromq { + + /*! + * \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 + { + 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_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 0000000000..27a19b210b --- /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 0000000000..b0fd23c892 --- /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 0000000000..cf9183375f --- /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 0964d819c2..941e5ff12b 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 0000000000..5b207ce677 --- /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/pull_msg_source_impl.h b/gr-zeromq/lib/pull_msg_source_impl.h new file mode 100644 index 0000000000..fb9237c408 --- /dev/null +++ b/gr-zeromq/lib/pull_msg_source_impl.h @@ -0,0 +1,51 @@ +/* -*- 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_PULL_MSG_SOURCE_IMPL_H +#define INCLUDED_ZEROMQ_PULL_MSG_SOURCE_IMPL_H + +#include <gnuradio/zeromq/pull_msg_source.h> +#include <zmq.hpp> + +namespace gr { + namespace zeromq { + + class pull_msg_source_impl : public pull_msg_source + { + private: + int d_timeout; // microseconds, -1 is blocking + zmq::context_t *d_context; + zmq::socket_t *d_socket; + + public: + pull_msg_source_impl(char *address, int timeout); + ~pull_msg_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_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 0000000000..6266cd6a57 --- /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/push_msg_sink_impl.h b/gr-zeromq/lib/push_msg_sink_impl.h new file mode 100644 index 0000000000..b77c998506 --- /dev/null +++ b/gr-zeromq/lib/push_msg_sink_impl.h @@ -0,0 +1,52 @@ +/* -*- 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_IMPL_H +#define INCLUDED_ZEROMQ_PUSH_MSG_SINK_IMPL_H + +#include <gnuradio/zeromq/push_msg_sink.h> +#include <zmq.hpp> + +namespace gr { + namespace zeromq { + + class push_msg_sink_impl : public push_msg_sink + { + private: + float d_timeout; + zmq::context_t *d_context; + zmq::socket_t *d_socket; + + public: + push_msg_sink_impl(char *address, int timeout); + ~push_msg_sink_impl(); + + int work(int noutput_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_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 0000000000..72dd5fb71b --- /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/rep_msg_sink_impl.h b/gr-zeromq/lib/rep_msg_sink_impl.h new file mode 100644 index 0000000000..40d7969472 --- /dev/null +++ b/gr-zeromq/lib/rep_msg_sink_impl.h @@ -0,0 +1,52 @@ +/* -*- 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_IMPL_H +#define INCLUDED_ZEROMQ_REP_MSG_SINK_IMPL_H + +#include <gnuradio/zeromq/rep_msg_sink.h> +#include <zmq.hpp> + +namespace gr { + namespace zeromq { + + class rep_msg_sink_impl : public rep_msg_sink + { + private: + int d_timeout; + zmq::context_t *d_context; + zmq::socket_t *d_socket; + + public: + rep_msg_sink_impl(char *address, int timeout); + ~rep_msg_sink_impl(); + + int work(int noutput_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_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 0000000000..92e26e919c --- /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/req_msg_source_impl.h b/gr-zeromq/lib/req_msg_source_impl.h new file mode 100644 index 0000000000..635fa45b62 --- /dev/null +++ b/gr-zeromq/lib/req_msg_source_impl.h @@ -0,0 +1,51 @@ +/* -*- 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_REQ_MSG_SOURCE_IMPL_H +#define INCLUDED_ZEROMQ_REQ_MSG_SOURCE_IMPL_H + +#include <gnuradio/zeromq/req_msg_source.h> +#include <zmq.hpp> + +namespace gr { + namespace zeromq { + + class req_msg_source_impl : public req_msg_source + { + private: + int d_timeout; + zmq::context_t *d_context; + zmq::socket_t *d_socket; + + public: + req_msg_source_impl(char *address, int timeout); + ~req_msg_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_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 1c63f2d97c..f7a9bc9c26 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 0db544038a..424eb470ad 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 c9a891d9a3..78669bdc69 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); |