Updates

- fix client timeout
- add client timeout test
- add socket cleanup on garbage collection

Author: paskozdilar

tests/test_rpc.py

@@ -4,8 +4,11 @@
 import sys
 import tempfile
 import time
+
+import pytest
+
 from zrpc.server import Server, rpc_method
-from zrpc.client import Client
+from zrpc.client import Client, RPCTimeoutError
 
 
 def test_basic_rpc():
@@ -175,3 +178,23 @@ def run_client():
         multiprocessing.Process(target=run_client, daemon=True).start()
 
         assert rpc_event.wait(1)
+
+
+def test_client_timeout():
+    """
+    Test whether client raises RPCTimeoutError on `call()` timeout.
+    """
+
+    client_event = multiprocessing.Event()
+
+    def run_client():
+        with tempfile.TemporaryDirectory() as tempdir:
+            client = Client(socket_dir=tempdir)
+            try:
+                client.call('non_existant', 'non_existant', timeout=0.1)
+            except RPCTimeoutError:
+                client_event.set()
+
+    multiprocessing.Process(target=run_client, daemon=True).start()
+
+    assert client_event.wait(1)

zrpc/client.py

@@ -39,6 +39,13 @@ def __init__(self, socket_dir=None):
         except (OSError, zmq.ZMQError) as exc:
             raise ConnectError('Cannot connect client') from exc
 
+    def __del__(self):
+        try:
+            for socket in self._sockets:
+                socket.close(linger=0)
+        except AttributeError:
+            pass
+
     def __connect(self, socket_name):
         context = self._context
         sockets = self._sockets
@@ -89,12 +96,13 @@ def call(self, server, method, args=(), kwargs={}, timeout=None):
         start_time = time.monotonic()
         events = {}
 
-        current_time = time.monotonic()
+        current_time = start_time
+        elapsed_time = current_time - start_time
 
         iter_timeout = 1
-        while (current_time - start_time) < timeout:
+        while elapsed_time <= timeout:
             socket.send(request)
-            timeout_ms = 1000 * max(0, min(iter_timeout, timeout - (current_time - start_time)))
+            timeout_ms = 1000 * max(0, min(iter_timeout, timeout - elapsed_time))
             logging.debug('Polling sockets with {}ms timeout'.format(timeout_ms))
             events = dict(self._poller.poll(timeout=timeout_ms))
 
@@ -106,6 +114,9 @@ def call(self, server, method, args=(), kwargs={}, timeout=None):
             self.__connect(server)
             socket = sockets[server]
 
+            current_time = time.monotonic()
+            elapsed_time = current_time - start_time
+
         if socket not in events:
             raise RPCTimeoutError('Service "%s" not responding' % server)
 

zrpc/server.py

@@ -136,7 +136,7 @@ def stop(self):
         try:
             if not self._started:
                 return
-            self._context.term()
+            self._socket.close(linger=0)
             os.unlink(self._socket_path)
         except (OSError, AttributeError):
             pass