summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJohannes Schmitz <schmitz@ti.rwth-aachen.de>2014-05-07 17:59:01 +0200
committerJohannes Schmitz <schmitz@ti.rwth-aachen.de>2014-05-07 17:59:01 +0200
commitea2b698dbad69cf5ae62c7cac3a7506d776b0dcd (patch)
treefb7f3d208433e75d4f18a3e13dc630e73af28869
parent35aaec4d0d4b6f995911fd6d38697bc239cd68cf (diff)
zeromq: Fix doxygen and memcpy for vlen
Diffstat
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/pub_sink.h3
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/pull_source.h1
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/push_sink.h1
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/rep_sink.h1
-rw-r--r--gr-zeromq/include/gnuradio/zeromq/req_source.h2
-rw-r--r--gr-zeromq/lib/pub_sink_impl.cc14
-rw-r--r--gr-zeromq/lib/pub_sink_impl.h3
-rw-r--r--gr-zeromq/lib/pull_source_impl.cc29
-rw-r--r--gr-zeromq/lib/pull_source_impl.h1
-rw-r--r--gr-zeromq/lib/push_sink_impl.cc9
-rw-r--r--gr-zeromq/lib/push_sink_impl.h1
-rw-r--r--gr-zeromq/lib/rep_sink_impl.cc10
-rw-r--r--gr-zeromq/lib/rep_sink_impl.h1
-rw-r--r--gr-zeromq/lib/req_source_impl.cc27
-rw-r--r--gr-zeromq/lib/req_source_impl.h1
15 files changed, 57 insertions, 47 deletions
diff --git a/gr-zeromq/include/gnuradio/zeromq/pub_sink.h b/gr-zeromq/include/gnuradio/zeromq/pub_sink.h
index 7c5734ee12..46ad80b0b3 100644
--- a/gr-zeromq/include/gnuradio/zeromq/pub_sink.h
+++ b/gr-zeromq/include/gnuradio/zeromq/pub_sink.h
@@ -49,10 +49,11 @@ namespace gr {
* \brief Return a shared_ptr to a new instance of zeromq::pub_sink.
*
* \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 blocking Indicate whether blocking sends should be used, default true.
*/
- static sptr make(size_t itemsize, char *address, bool blocking=true);
+ static sptr make(size_t itemsize, size_t vlen, char *address, bool blocking=true);
};
} // namespace zeromq
diff --git a/gr-zeromq/include/gnuradio/zeromq/pull_source.h b/gr-zeromq/include/gnuradio/zeromq/pull_source.h
index 12c1005174..4306931e6e 100644
--- a/gr-zeromq/include/gnuradio/zeromq/pull_source.h
+++ b/gr-zeromq/include/gnuradio/zeromq/pull_source.h
@@ -46,6 +46,7 @@ namespace gr {
* \brief Return a shared_ptr to a new instance of gr::zeromq::pull_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
*
diff --git a/gr-zeromq/include/gnuradio/zeromq/push_sink.h b/gr-zeromq/include/gnuradio/zeromq/push_sink.h
index 2f68e44ca0..46ad6a4863 100644
--- a/gr-zeromq/include/gnuradio/zeromq/push_sink.h
+++ b/gr-zeromq/include/gnuradio/zeromq/push_sink.h
@@ -50,6 +50,7 @@ namespace gr {
* \brief Return a shared_ptr to a new instance of gr::zeromq::push_sink
*
* \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 blocking Indicate whether blocking sends should be used, default true.
*
diff --git a/gr-zeromq/include/gnuradio/zeromq/rep_sink.h b/gr-zeromq/include/gnuradio/zeromq/rep_sink.h
index 9c2cafba12..374607e4b6 100644
--- a/gr-zeromq/include/gnuradio/zeromq/rep_sink.h
+++ b/gr-zeromq/include/gnuradio/zeromq/rep_sink.h
@@ -48,6 +48,7 @@ namespace gr {
* \brief Return a shared_ptr to a new instance of zeromq::rep_sink.
*
* \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 Timeout for request poll, in seconds
* \param blocking Indicate whether blocking sends should be used, default true.
diff --git a/gr-zeromq/include/gnuradio/zeromq/req_source.h b/gr-zeromq/include/gnuradio/zeromq/req_source.h
index 5d3f380f8b..5fc3682241 100644
--- a/gr-zeromq/include/gnuradio/zeromq/req_source.h
+++ b/gr-zeromq/include/gnuradio/zeromq/req_source.h
@@ -47,7 +47,7 @@ namespace gr {
*
*
* \param itemsize Size of a stream item in bytes
- * \param vlen of the input items.
+ * \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
*
diff --git a/gr-zeromq/lib/pub_sink_impl.cc b/gr-zeromq/lib/pub_sink_impl.cc
index 086e995b2e..07002395f5 100644
--- a/gr-zeromq/lib/pub_sink_impl.cc
+++ b/gr-zeromq/lib/pub_sink_impl.cc
@@ -31,17 +31,17 @@ namespace gr {
namespace zeromq {
pub_sink::sptr
- pub_sink::make(size_t itemsize, char *address, bool blocking)
+ pub_sink::make(size_t itemsize, size_t vlen, char *address, bool blocking)
{
return gnuradio::get_initial_sptr
- (new pub_sink_impl(itemsize, address, blocking));
+ (new pub_sink_impl(itemsize, vlen, address, blocking));
}
- pub_sink_impl::pub_sink_impl(size_t itemsize, char *address, bool blocking)
+ pub_sink_impl::pub_sink_impl(size_t itemsize, size_t vlen, char *address, bool blocking)
: gr::sync_block("pub_sink",
- gr::io_signature::make(1, 1, itemsize),
+ gr::io_signature::make(1, 1, itemsize * vlen),
gr::io_signature::make(0, 0, 0)),
- d_itemsize(itemsize), d_blocking(blocking)
+ d_itemsize(itemsize), d_vlen(vlen), d_blocking(blocking)
{
d_context = new zmq::context_t(1);
d_socket = new zmq::socket_t(*d_context, ZMQ_PUB);
@@ -62,8 +62,8 @@ namespace gr {
const char *in = (const char *)input_items[0];
// create message copy and send
- zmq::message_t msg(d_itemsize*noutput_items);
- memcpy((void *)msg.data(), in, d_itemsize*noutput_items);
+ zmq::message_t msg(d_itemsize*d_vlen*noutput_items);
+ memcpy((void *)msg.data(), in, d_itemsize*d_vlen*noutput_items);
d_socket->send(msg, d_blocking ? 0 : ZMQ_NOBLOCK);
return noutput_items;
diff --git a/gr-zeromq/lib/pub_sink_impl.h b/gr-zeromq/lib/pub_sink_impl.h
index d1cb02d289..6d36e207cd 100644
--- a/gr-zeromq/lib/pub_sink_impl.h
+++ b/gr-zeromq/lib/pub_sink_impl.h
@@ -33,12 +33,13 @@ namespace gr {
{
private:
size_t d_itemsize;
+ size_t d_vlen;
bool d_blocking;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
public:
- pub_sink_impl(size_t itemsize, char *address, bool blocking);
+ pub_sink_impl(size_t itemsize, size_t vlen, char *address, bool blocking);
~pub_sink_impl();
int work(int noutput_items,
diff --git a/gr-zeromq/lib/pull_source_impl.cc b/gr-zeromq/lib/pull_source_impl.cc
index 9b9e50a38f..b29a056d84 100644
--- a/gr-zeromq/lib/pull_source_impl.cc
+++ b/gr-zeromq/lib/pull_source_impl.cc
@@ -41,7 +41,7 @@ namespace gr {
: gr::sync_block("pull_source",
gr::io_signature::make(0, 0, 0),
gr::io_signature::make(1, 1, itemsize * vlen)),
- d_itemsize(itemsize)
+ d_itemsize(itemsize), d_vlen(vlen)
{
d_timeout = timeout >= 0 ? (int)(timeout*1e6) : 0;
d_context = new zmq::context_t(1);
@@ -71,21 +71,22 @@ namespace gr {
// If we got a reply, process
if (items[0].revents & ZMQ_POLLIN) {
- // Receive data
- zmq::message_t msg;
- d_socket->recv(&msg);
+ // Receive data
+ zmq::message_t msg;
+ std::cout << "pull before" << std::endl;
+ d_socket->recv(&msg);
+ std::cout << "pull after" << std::endl;
+ // 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);
- // Copy to ouput buffer and return
- if (msg.size() >= d_itemsize*noutput_items) {
- memcpy(out, (void *)msg.data(), d_itemsize*noutput_items);
+ return noutput_items;
+ }
+ else {
+ memcpy(out, (void *)msg.data(), msg.size());
- return noutput_items;
- }
- else {
- memcpy(out, (void *)msg.data(), msg.size());
-
- return msg.size()/d_itemsize;
- }
+ return msg.size()/(d_itemsize*d_vlen);
+ }
}
else {
return 0; // FIXME: someday when the scheduler does all the poll/selects
diff --git a/gr-zeromq/lib/pull_source_impl.h b/gr-zeromq/lib/pull_source_impl.h
index 3e56bdc4c5..de2e903a85 100644
--- a/gr-zeromq/lib/pull_source_impl.h
+++ b/gr-zeromq/lib/pull_source_impl.h
@@ -33,6 +33,7 @@ namespace gr {
{
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;
diff --git a/gr-zeromq/lib/push_sink_impl.cc b/gr-zeromq/lib/push_sink_impl.cc
index 43ccd9dbca..1438e524a5 100644
--- a/gr-zeromq/lib/push_sink_impl.cc
+++ b/gr-zeromq/lib/push_sink_impl.cc
@@ -41,7 +41,7 @@ namespace gr {
: gr::sync_block("push_sink",
gr::io_signature::make(1, 1, itemsize * vlen),
gr::io_signature::make(0, 0, 0)),
- d_itemsize(itemsize)
+ d_itemsize(itemsize), d_vlen(vlen)
{
d_blocking = blocking;
d_context = new zmq::context_t(1);
@@ -63,10 +63,11 @@ namespace gr {
const char *in = (const char *) input_items[0];
// create message copy and send
- zmq::message_t msg(d_itemsize*noutput_items);
- memcpy((void *)msg.data(), in, d_itemsize*noutput_items);
+ zmq::message_t msg(d_itemsize*d_vlen*noutput_items);
+ memcpy((void *)msg.data(), in, d_itemsize*d_vlen*noutput_items);
+ std::cout << "before" << std::endl;
d_socket->send(msg, d_blocking ? 0 : ZMQ_NOBLOCK);
-
+ std::cout << "after" << std::endl;
return noutput_items;
}
diff --git a/gr-zeromq/lib/push_sink_impl.h b/gr-zeromq/lib/push_sink_impl.h
index 77f1f60c8d..76ee2c2b39 100644
--- a/gr-zeromq/lib/push_sink_impl.h
+++ b/gr-zeromq/lib/push_sink_impl.h
@@ -33,6 +33,7 @@ namespace gr {
{
private:
size_t d_itemsize;
+ size_t d_vlen;
bool d_blocking;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
diff --git a/gr-zeromq/lib/rep_sink_impl.cc b/gr-zeromq/lib/rep_sink_impl.cc
index 38bb7d779e..7e5f43df4c 100644
--- a/gr-zeromq/lib/rep_sink_impl.cc
+++ b/gr-zeromq/lib/rep_sink_impl.cc
@@ -41,7 +41,7 @@ namespace gr {
: gr::sync_block("rep_sink",
gr::io_signature::make(1, 1, itemsize * vlen),
gr::io_signature::make(0, 0, 0)),
- d_itemsize(itemsize), d_blocking(blocking)
+ d_itemsize(itemsize), d_vlen(vlen), d_blocking(blocking)
{
d_timeout = timeout >= 0 ? (int)(timeout*1e6) : 0;
d_context = new zmq::context_t(1);
@@ -74,15 +74,15 @@ namespace gr {
// create message copy and send
if (noutput_items < req_output_items) {
- zmq::message_t msg(d_itemsize*noutput_items);
- memcpy((void *)msg.data(), in, d_itemsize*noutput_items);
+ zmq::message_t msg(d_itemsize*d_vlen*noutput_items);
+ memcpy((void *)msg.data(), in, d_itemsize*d_vlen*noutput_items);
d_socket->send(msg, d_blocking ? 0 : ZMQ_NOBLOCK);
return noutput_items;
}
else {
- zmq::message_t msg(d_itemsize*req_output_items);
- memcpy((void *)msg.data(), in, d_itemsize*req_output_items);
+ zmq::message_t msg(d_itemsize*d_vlen*req_output_items);
+ memcpy((void *)msg.data(), in, d_itemsize*d_vlen*req_output_items);
d_socket->send(msg, d_blocking ? 0 : ZMQ_NOBLOCK);
return req_output_items;
diff --git a/gr-zeromq/lib/rep_sink_impl.h b/gr-zeromq/lib/rep_sink_impl.h
index 0996d53616..17e1da790d 100644
--- a/gr-zeromq/lib/rep_sink_impl.h
+++ b/gr-zeromq/lib/rep_sink_impl.h
@@ -33,6 +33,7 @@ namespace gr {
{
private:
size_t d_itemsize;
+ size_t d_vlen;
int d_timeout;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
diff --git a/gr-zeromq/lib/req_source_impl.cc b/gr-zeromq/lib/req_source_impl.cc
index ffe31f9953..8dedf6c6af 100644
--- a/gr-zeromq/lib/req_source_impl.cc
+++ b/gr-zeromq/lib/req_source_impl.cc
@@ -41,7 +41,7 @@ namespace gr {
: gr::sync_block("req_source",
gr::io_signature::make(0, 0, 0),
gr::io_signature::make(1, 1, itemsize * vlen)),
- d_itemsize(itemsize)
+ d_itemsize(itemsize), d_vlen(vlen)
{
d_context = new zmq::context_t(1);
d_socket = new zmq::socket_t(*d_context, ZMQ_REQ);
@@ -56,8 +56,8 @@ namespace gr {
int
req_source_impl::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)
{
char *out = (char*)output_items[0];
@@ -66,10 +66,10 @@ namespace gr {
// 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);
+ // 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 } };
@@ -77,14 +77,13 @@ namespace gr {
// If we got a reply, process
if (itemsin[0].revents & ZMQ_POLLIN) {
- // Receive data
- zmq::message_t reply;
- d_socket->recv(&reply);
+ // Receive data
+ zmq::message_t reply;
+ d_socket->recv(&reply);
- // Copy to ouput buffer and return
- memcpy(out, (void *)reply.data(), reply.size());
-
- return reply.size()/d_itemsize;
+ // Copy to ouput buffer and return
+ memcpy(out, (void *)reply.data(), reply.size());
+ return reply.size()/(d_itemsize*d_vlen);
}
return 0;
diff --git a/gr-zeromq/lib/req_source_impl.h b/gr-zeromq/lib/req_source_impl.h
index 6c87b48bca..4d40f80cb2 100644
--- a/gr-zeromq/lib/req_source_impl.h
+++ b/gr-zeromq/lib/req_source_impl.h
@@ -33,6 +33,7 @@ namespace gr {
{
private:
size_t d_itemsize;
+ size_t d_vlen;
zmq::context_t *d_context;
zmq::socket_t *d_socket;
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-24 13:08:39 +0000