5 files changed, 34 insertions, 18 deletions
diff --git a/gr-zeromq/lib/push_sink_impl.cc b/gr-zeromq/lib/push_sink_impl.cc
index 5acc3c555e..9d92c51c6d 100644
--- a/gr-zeromq/lib/push_sink_impl.cc
+++ b/gr-zeromq/lib/push_sink_impl.cc
@@ -31,18 +31,19 @@ namespace gr {
   namespace zeromq {
 
     push_sink::sptr
-    push_sink::make(size_t itemsize, size_t vlen, char *address, bool blocking)
+    push_sink::make(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
     {
       return gnuradio::get_initial_sptr
-        (new push_sink_impl(itemsize, vlen, address, blocking));
+        (new push_sink_impl(itemsize, vlen, address, timeout, blocking));
     }
 
-    push_sink_impl::push_sink_impl(size_t itemsize, size_t vlen, char *address, bool blocking)
+    push_sink_impl::push_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
       : gr::sync_block("push_sink",
                        gr::io_signature::make(1, 1, itemsize * vlen),
                        gr::io_signature::make(0, 0, 0)),
         d_itemsize(itemsize), d_vlen(vlen)
     {
+      d_timeout = timeout >= 0 ? (int)(timeout*1e6) : 0;
       d_blocking = blocking;
       d_context = new zmq::context_t(1);
       d_socket = new zmq::socket_t(*d_context, ZMQ_PUSH);
@@ -64,11 +65,21 @@ namespace gr {
     {
       const char *in = (const char *) input_items[0];
 
-      // create message copy and send
-      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;
+      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) {
+        // create message copy and send
+        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 {
+        return 0;
+      }
     }
 
   } /* namespace zeromq */
diff --git a/gr-zeromq/lib/push_sink_impl.h b/gr-zeromq/lib/push_sink_impl.h
index a34bb28a70..e1a9051e21 100644
--- a/gr-zeromq/lib/push_sink_impl.h
+++ b/gr-zeromq/lib/push_sink_impl.h
@@ -34,12 +34,13 @@ namespace gr {
     private:
       size_t          d_itemsize;
       size_t          d_vlen;
+      float           d_timeout;
       bool            d_blocking;
       zmq::context_t  *d_context;
       zmq::socket_t   *d_socket;
 
     public:
-      push_sink_impl(size_t itemsize, size_t vlen, char *address, bool blocking);
+      push_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking);
       ~push_sink_impl();
 
       int work(int noutput_items,
diff --git a/gr-zeromq/lib/rep_sink_impl.h b/gr-zeromq/lib/rep_sink_impl.h
index a933d94fa6..6e10a89b59 100644
--- a/gr-zeromq/lib/rep_sink_impl.h
+++ b/gr-zeromq/lib/rep_sink_impl.h
@@ -35,9 +35,9 @@ namespace gr {
       size_t          d_itemsize;
       size_t          d_vlen;
       int             d_timeout;
+      bool            d_blocking;
       zmq::context_t  *d_context;
       zmq::socket_t   *d_socket;
-      bool            d_blocking;
 
     public:
       rep_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking);
diff --git a/gr-zeromq/lib/req_source_impl.cc b/gr-zeromq/lib/req_source_impl.cc
index 937b5942c0..84cbcc7c8f 100644
--- a/gr-zeromq/lib/req_source_impl.cc
+++ b/gr-zeromq/lib/req_source_impl.cc
@@ -31,18 +31,19 @@ namespace gr {
   namespace zeromq {
 
     req_source::sptr
-    req_source::make(size_t itemsize, size_t vlen, char *address)
+    req_source::make(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
     {
       return gnuradio::get_initial_sptr
-        (new req_source_impl(itemsize, vlen, address));
+        (new req_source_impl(itemsize, vlen, address, timeout, blocking));
     }
 
-    req_source_impl::req_source_impl(size_t itemsize, size_t vlen, char *address)
+    req_source_impl::req_source_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
       : gr::sync_block("req_source",
 		       gr::io_signature::make(0, 0, 0),
 		       gr::io_signature::make(1, 1, itemsize * vlen)),
-        d_itemsize(itemsize), d_vlen(vlen)
+        d_itemsize(itemsize), d_vlen(vlen), d_timeout(timeout), d_blocking(blocking)
     {
+      d_timeout = timeout >= 0 ? (int)(timeout*1e6) : 0;
       d_context = new zmq::context_t(1);
       d_socket = new zmq::socket_t(*d_context, ZMQ_REQ);
       int time = 0;
@@ -68,14 +69,15 @@ namespace gr {
 
       //  If we got a reply, process
       if (itemsout[0].revents & ZMQ_POLLOUT) {
-        // Request data, FIXME non portable
+        // Request data, FIXME non portable?
         zmq::message_t request(sizeof(int));
         memcpy ((void *) request.data (), &noutput_items, sizeof(int));
-        d_socket->send(request);
+        d_socket->send(request, d_blocking ? 0 : ZMQ_NOBLOCK);
+
       }
 
       zmq::pollitem_t itemsin[] = { { *d_socket, 0, ZMQ_POLLIN, 0 } };
-      zmq::poll (&itemsin[0], 1, 0);
+      zmq::poll (&itemsin[0], 1, d_timeout);
 
       //  If we got a reply, process
       if (itemsin[0].revents & ZMQ_POLLIN) {
diff --git a/gr-zeromq/lib/req_source_impl.h b/gr-zeromq/lib/req_source_impl.h
index c906e9137d..754f2081e8 100644
--- a/gr-zeromq/lib/req_source_impl.h
+++ b/gr-zeromq/lib/req_source_impl.h
@@ -34,11 +34,13 @@ namespace gr {
     private:
       size_t          d_itemsize;
       size_t          d_vlen;
+      int             d_timeout;
+      bool            d_blocking;
       zmq::context_t  *d_context;
       zmq::socket_t   *d_socket;
 
     public:
-      req_source_impl(size_t itemsize, size_t vlen, char *address);
+      req_source_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking);
       ~req_source_impl();
 
       int work(int noutput_items,