10 files changed, 55 insertions, 51 deletions
diff --git a/gr-zeromq/lib/pub_sink_impl.cc b/gr-zeromq/lib/pub_sink_impl.cc
index 9cd115da79..3a86819fb3 100644
--- a/gr-zeromq/lib/pub_sink_impl.cc
+++ b/gr-zeromq/lib/pub_sink_impl.cc
@@ -34,7 +34,7 @@ namespace gr {
     pub_sink::make(size_t itemsize, size_t vlen, char *address, bool blocking)
     {
       return gnuradio::get_initial_sptr
-	(new pub_sink_impl(itemsize, vlen, address, blocking));
+        (new pub_sink_impl(itemsize, vlen, address, blocking));
     }
 
     pub_sink_impl::pub_sink_impl(size_t itemsize, size_t vlen, char *address, bool blocking)
@@ -52,6 +52,8 @@ namespace gr {
 
     pub_sink_impl::~pub_sink_impl()
     {
+      d_socket->close();
+      d_context->close();
       delete d_socket;
       delete d_context;
     }
diff --git a/gr-zeromq/lib/pub_sink_impl.h b/gr-zeromq/lib/pub_sink_impl.h
index 8e6fac7485..ad0419ad26 100644
--- a/gr-zeromq/lib/pub_sink_impl.h
+++ b/gr-zeromq/lib/pub_sink_impl.h
@@ -43,8 +43,8 @@ namespace gr {
       ~pub_sink_impl();
 
       int 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);
     };
 
   } // namespace zeromq
diff --git a/gr-zeromq/lib/pull_source_impl.cc b/gr-zeromq/lib/pull_source_impl.cc
index 2498eeedc3..2ee6ad08c8 100644
--- a/gr-zeromq/lib/pull_source_impl.cc
+++ b/gr-zeromq/lib/pull_source_impl.cc
@@ -31,19 +31,18 @@ namespace gr {
   namespace zeromq {
 
     pull_source::sptr
-    pull_source::make(size_t itemsize, size_t vlen, char *address, float timeout)
+    pull_source::make(size_t itemsize, size_t vlen, char *address, int timeout)
     {
       return gnuradio::get_initial_sptr
         (new pull_source_impl(itemsize, vlen, address, timeout));
     }
 
-    pull_source_impl::pull_source_impl(size_t itemsize, size_t vlen, char *address, float timeout)
+    pull_source_impl::pull_source_impl(size_t itemsize, size_t vlen, char *address, int timeout)
       : gr::sync_block("pull_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_timeout = timeout >= 0 ? (int)(timeout*1e6) : 0;
       d_context = new zmq::context_t(1);
       d_socket = new zmq::socket_t(*d_context, ZMQ_PULL);
       int time = 0;
@@ -56,6 +55,8 @@ namespace gr {
      */
     pull_source_impl::~pull_source_impl()
     {
+      d_socket->close();
+      d_context->close();
       delete d_socket;
       delete d_context;
     }
@@ -89,7 +90,7 @@ namespace gr {
         }
       }
       else {
-	return 0; // FIXME: someday when the scheduler does all the poll/selects
+        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 5140dc54bb..e69de81dcd 100644
--- a/gr-zeromq/lib/pull_source_impl.h
+++ b/gr-zeromq/lib/pull_source_impl.h
@@ -39,7 +39,7 @@ namespace gr {
       zmq::socket_t   *d_socket;
 
     public:
-      pull_source_impl(size_t itemsize, size_t vlen, char *address, float timeout);
+      pull_source_impl(size_t itemsize, size_t vlen, char *address, int timeout);
       ~pull_source_impl();
 
       int work(int noutput_items,
diff --git a/gr-zeromq/lib/push_sink_impl.cc b/gr-zeromq/lib/push_sink_impl.cc
index 9d92c51c6d..2df3a6f2db 100644
--- a/gr-zeromq/lib/push_sink_impl.cc
+++ b/gr-zeromq/lib/push_sink_impl.cc
@@ -31,20 +31,18 @@ namespace gr {
   namespace zeromq {
 
     push_sink::sptr
-    push_sink::make(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
+    push_sink::make(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking)
     {
       return gnuradio::get_initial_sptr
         (new push_sink_impl(itemsize, vlen, address, timeout, blocking));
     }
 
-    push_sink_impl::push_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
+    push_sink_impl::push_sink_impl(size_t itemsize, size_t vlen, char *address, int 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_itemsize(itemsize), d_vlen(vlen), d_timeout(timeout), d_blocking(blocking)
     {
-      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);
       int time = 0;
@@ -54,8 +52,10 @@ namespace gr {
 
     push_sink_impl::~push_sink_impl()
     {
-      delete(d_socket);
-      delete(d_context);
+      d_socket->close();
+      d_context->close();
+      delete d_socket;
+      delete d_context;
     }
 
     int
diff --git a/gr-zeromq/lib/push_sink_impl.h b/gr-zeromq/lib/push_sink_impl.h
index e1a9051e21..2ff9bc509a 100644
--- a/gr-zeromq/lib/push_sink_impl.h
+++ b/gr-zeromq/lib/push_sink_impl.h
@@ -40,12 +40,12 @@ namespace gr {
       zmq::socket_t   *d_socket;
 
     public:
-      push_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking);
+      push_sink_impl(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking);
       ~push_sink_impl();
 
       int 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);
     };
 
   } // namespace zeromq
diff --git a/gr-zeromq/lib/rep_sink_impl.cc b/gr-zeromq/lib/rep_sink_impl.cc
index d3d8f81790..35efffa7f4 100644
--- a/gr-zeromq/lib/rep_sink_impl.cc
+++ b/gr-zeromq/lib/rep_sink_impl.cc
@@ -31,19 +31,18 @@ namespace gr {
   namespace zeromq {
 
     rep_sink::sptr
-    rep_sink::make(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
+    rep_sink::make(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking)
     {
       return gnuradio::get_initial_sptr
         (new rep_sink_impl(itemsize, vlen, address, timeout, blocking));
     }
 
-    rep_sink_impl::rep_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
+    rep_sink_impl::rep_sink_impl(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking)
       : gr::sync_block("rep_sink",
-		       gr::io_signature::make(1, 1, itemsize * vlen),
-		       gr::io_signature::make(0, 0, 0)),
-        d_itemsize(itemsize), d_vlen(vlen), d_blocking(blocking)
+                       gr::io_signature::make(1, 1, itemsize * vlen),
+                       gr::io_signature::make(0, 0, 0)),
+        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_REP);
       int time = 0;
@@ -53,6 +52,8 @@ namespace gr {
 
     rep_sink_impl::~rep_sink_impl()
     {
+      d_socket->close();
+      d_context->close();
       delete d_socket;
       delete d_context;
     }
@@ -69,26 +70,26 @@ namespace gr {
 
       //  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()));
+        // receive data request
+        zmq::message_t request;
+        d_socket->recv(&request);
+        int req_output_items = *(static_cast<int*>(request.data()));
 
-	// create message copy and send
-	if (noutput_items < req_output_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);
+        // create message copy and send
+        if (noutput_items < req_output_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*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 noutput_items;
+        }
+        else {
+          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;
-	}
+          return req_output_items;
+        }
       }
 
       return 0;
diff --git a/gr-zeromq/lib/rep_sink_impl.h b/gr-zeromq/lib/rep_sink_impl.h
index 6e10a89b59..500af7a3b3 100644
--- a/gr-zeromq/lib/rep_sink_impl.h
+++ b/gr-zeromq/lib/rep_sink_impl.h
@@ -40,7 +40,7 @@ namespace gr {
       zmq::socket_t   *d_socket;
 
     public:
-      rep_sink_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking);
+      rep_sink_impl(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking);
       ~rep_sink_impl();
 
       int work(int noutput_items,
diff --git a/gr-zeromq/lib/req_source_impl.cc b/gr-zeromq/lib/req_source_impl.cc
index 84cbcc7c8f..121d2e37fb 100644
--- a/gr-zeromq/lib/req_source_impl.cc
+++ b/gr-zeromq/lib/req_source_impl.cc
@@ -31,19 +31,18 @@ namespace gr {
   namespace zeromq {
 
     req_source::sptr
-    req_source::make(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
+    req_source::make(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking)
     {
       return gnuradio::get_initial_sptr
         (new req_source_impl(itemsize, vlen, address, timeout, blocking));
     }
 
-    req_source_impl::req_source_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking)
+    req_source_impl::req_source_impl(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking)
       : gr::sync_block("req_source",
-		       gr::io_signature::make(0, 0, 0),
-		       gr::io_signature::make(1, 1, itemsize * vlen)),
+                       gr::io_signature::make(0, 0, 0),
+                       gr::io_signature::make(1, 1, itemsize * 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;
@@ -53,6 +52,8 @@ namespace gr {
 
     req_source_impl::~req_source_impl()
     {
+      d_socket->close();
+      d_context->close();
       delete d_socket;
       delete d_context;
     }
@@ -65,7 +66,7 @@ namespace gr {
       char *out = (char*)output_items[0];
 
       zmq::pollitem_t itemsout[] = { { *d_socket, 0, ZMQ_POLLOUT, 0 } };
-      zmq::poll (&itemsout[0], 1, 0);
+      zmq::poll (&itemsout[0], 1, d_timeout);
 
       //  If we got a reply, process
       if (itemsout[0].revents & ZMQ_POLLOUT) {
@@ -73,7 +74,6 @@ namespace gr {
         zmq::message_t request(sizeof(int));
         memcpy ((void *) request.data (), &noutput_items, sizeof(int));
         d_socket->send(request, d_blocking ? 0 : ZMQ_NOBLOCK);
-
       }
 
       zmq::pollitem_t itemsin[] = { { *d_socket, 0, ZMQ_POLLIN, 0 } };
diff --git a/gr-zeromq/lib/req_source_impl.h b/gr-zeromq/lib/req_source_impl.h
index 754f2081e8..ab844efa00 100644
--- a/gr-zeromq/lib/req_source_impl.h
+++ b/gr-zeromq/lib/req_source_impl.h
@@ -40,7 +40,7 @@ namespace gr {
       zmq::socket_t   *d_socket;
 
     public:
-      req_source_impl(size_t itemsize, size_t vlen, char *address, float timeout, bool blocking);
+      req_source_impl(size_t itemsize, size_t vlen, char *address, int timeout, bool blocking);
       ~req_source_impl();
 
       int work(int noutput_items,