Fix read forever loop

.github/workflows/tests.yml

@@ -0,0 +1,25 @@
+name: Tests
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+          python-version: "3.14"
+
+      - name: Install dependencies
+        run: uv sync --locked --group dev
+
+      - name: Run tests
+        run: uv run pytest --cov pseud

.gitignore

@@ -3,10 +3,11 @@ dist/
 build/
 docs/build/
 .coverage
-.tox/
 .cache/
 __pycache__/
 *.pyc
 .idea/
 .python-version
+.venv/
+.pytest_cache/
 

pseud/__init__.py

@@ -1,10 +1,5 @@
-import zmq.asyncio
-
 from .auth import *  # noqa
 from .client import Client, SyncClient  # noqa
 from .heartbeat import *  # noqa
 from .predicate import *  # noqa
 from .server import Server  # noqa
-
-
-zmq.asyncio.install()

pseud/auth.py

@@ -49,6 +49,9 @@ def configure(self):
     async def handle_hello(self, *args):
         pass
 
+    async def handle_authentication(self, user_id, routing_id, message_uuid):
+        pass
+
     async def handle_authenticated(self, message):
         pass
 
@@ -99,6 +102,9 @@ async def stop(self):
     async def handle_hello(self, *args):
         pass
 
+    async def handle_authentication(self, user_id, routing_id, message_uuid):
+        pass
+
     async def handle_authenticated(self, message):
         pass
 
@@ -168,6 +174,9 @@ async def _zap_handler(self, message):
     async def handle_hello(self, *args):
         pass
 
+    async def handle_authentication(self, user_id, routing_id, message_uuid):
+        pass
+
     async def handle_authenticated(self, message):
         pass
 
@@ -217,6 +226,9 @@ async def stop(self):
     async def handle_hello(self, *args):
         pass
 
+    async def handle_authentication(self, user_id, routing_id, message_uuid):
+        pass
+
     async def handle_authenticated(self, message):
         pass
 
@@ -287,6 +299,9 @@ async def _zap_handler(self, message):
     async def handle_hello(self, *args):
         pass
 
+    async def handle_authentication(self, user_id, routing_id, message_uuid):
+        pass
+
     async def handle_authenticated(self, message):
         pass
 

pseud/common.py

@@ -74,7 +74,15 @@ def handle_result(future):
 async def read_forever(socket, callback, copy=False):
     while True:
         result = await socket.recv_multipart(copy=copy)
-        await callback(result)
+        try:
+            await callback(result)
+        # asyncio.CancelledError is subclass of BaseException since 3.8
+        except Exception:
+            # This task is the only consumer of the socket. If an exception
+            # escaped here it would kill the task and cut off every peer until
+            # a manual restart.
+            # Drop the bad message, log it, and keep reading.
+            logger.exception('Error handling received message')
 
 
 class AttributeWrapper(object):
@@ -135,7 +143,17 @@ def __init__(self, user_id=None, routing_id=None, peer_routing_id=None,
             name=heartbeat_plugin)
         self.proxy_to = proxy_to
         self.reader = None
-        self.loop = loop or asyncio.get_event_loop()
+        if loop is not None:
+            self.loop = loop
+        else:
+            try:
+                self.loop = asyncio.get_event_loop()
+            except RuntimeError:
+                # No loop is running and none is set for this thread.
+                # Recreate the fallback that asyncio.get_event_loop() itself
+                # used to provide, before it started raising instead.
+                self.loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(self.loop)
         self.reader = None
         self.registry = (registry if registry is not None
                          else create_local_registry(user_id or ''))

pseud/interfaces.py

@@ -57,7 +57,7 @@ async def handle_hello(user_id, routing_id, message_uuid, message):
                                    'Welcome {!r}'.format(user_id)])
         """
 
-    async def handle_authentication(self, user_id, routing_id, message_uuid):
+    async def handle_authentication(user_id, routing_id, message_uuid):
         """
         Called when rpc received acknowledgement of failed authentication.
         """

pseud/packer.py

@@ -64,7 +64,7 @@ def __init__(self, translation_table=None):
 
     def packb(self, data):
         try:
-            return msgpack.packb(data, encoding='utf-8', use_bin_type=True,
+            return msgpack.packb(data, use_bin_type=True,
                                  default=self.ext_type_pack_hook)
         except:
             logger.exception('Packing failed')
@@ -74,7 +74,7 @@ def unpackb(self, packed):
         try:
             if isinstance(packed, zmq.sugar.Frame):
                 packed = packed.bytes
-            return msgpack.unpackb(packed, use_list=False, encoding='utf-8',
+            return msgpack.unpackb(packed, use_list=False,
                                    ext_hook=self.ext_type_unpack_hook)
         except:
             logger.exception('Unpacking failed')

pyproject.toml

@@ -0,0 +1,60 @@
+[build-system]
+requires = ["uv_build"]
+build-backend = "uv_build"
+
+[project]
+name = "pseud"
+version = "1.0.0+qb.13"
+description = "Bidirectionnal RPC Api on top of pyzmq"
+readme = "README.rst"
+license = "Apache-2.0"
+authors = [
+    { name = "Nicolas Delaby", email = "ticosax@free.fr" },
+]
+requires-python = ">=3.10"
+keywords = ["rpc", "zeromq", "pyzmq", "curve", "bidirectional", "asyncio"]
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Intended Audience :: Developers",
+    "Intended Audience :: System Administrators",
+    "Operating System :: OS Independent",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
+]
+dependencies = [
+    "pyzmq>=14.4",
+    "msgpack>=1.0",
+    "zope.component",
+    "zope.interface",
+]
+
+[project.urls]
+Homepage = "https://github.com/ticosax/pseud"
+Documentation = "https://pseud.readthedocs.io/en/latest/index.html"
+
+[project.optional-dependencies]
+doc = ["sphinx", "repoze.sphinx.autointerface"]
+
+[dependency-groups]
+dev = [
+    "pytest",
+    "pytest-cov",
+    "pytest-asyncio",
+]
+
+[tool.uv.build-backend]
+module-root = ""
+
+[[tool.uv.index]]
+name = "pypi"
+url = "https://pypi.org/simple"
+default = true
+
+[tool.coverage.report]
+exclude_lines = [
+    "pragma: no cover",
+    "raise NotImplementedError",
+]

tests/conftest.py

@@ -1,24 +1,41 @@
 import asyncio
+import os
+import stat
 
 import pytest
+import pytest_asyncio
 import zmq.asyncio
 
 
-@pytest.fixture
-def event_loop():
-    loop = zmq.asyncio.ZMQEventLoop()
-    asyncio.set_event_loop(loop)
-    yield loop
-    loop.close()
-
-
-@pytest.fixture
-def loop(event_loop):
-    return event_loop
+@pytest_asyncio.fixture
+async def loop():
+    # pytest-asyncio drives each async test on its own loop and no longer
+    # lets a same-named fixture override it, so grab that loop instead of
+    # creating a separate one that the test wouldn't actually run on.
+    return asyncio.get_running_loop()
 
 
 @pytest.fixture(autouse=True)
 def close_context():
     yield
     context = zmq.asyncio.Context.instance()
     context.destroy(linger=0)
+
+
+@pytest.fixture(autouse=True)
+def cleanup_ipc_sockets():
+    # ipc:// endpoints bind a real unix socket file on disk, and closing
+    # the socket doesn't unlink it. Several tests bind relative, bare
+    # names (e.g. 'ipc://here'), which land in the cwd pytest runs from.
+    # Remove whatever socket files show up there during the test.
+    cwd = os.getcwd()
+    before = set(os.listdir(cwd))
+    yield
+    after = set(os.listdir(cwd))
+    for name in after - before:
+        path = os.path.join(cwd, name)
+        try:
+            if stat.S_ISSOCK(os.stat(path).st_mode):
+                os.unlink(path)
+        except OSError:
+            pass

tests/test_heartbeat.py

@@ -97,7 +97,11 @@ async def collector(sink):
                 sink.extend(await monitoring_socket.recv_multipart())
 
         task = loop.create_task(collector(sink))
-        await asyncio.sleep(1.1)
+        # PLAIN auth handshake + PUB/SUB slow-joiner add up to ~0.5s of
+        # startup latency before the first heartbeat is observed, so the
+        # window needs enough steady-state time on top of that for 10
+        # heartbeats sent every .1s to actually land.
+        await asyncio.sleep(2)
         task.cancel()
         assert len(sink) >= 10
         assert all([b'client' == i for i in sink]), sink