chore: Add python 3.10 support (#10)

Author: borisalekseev

.github/workflows/tests.yaml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 
     services:
       artemis:
@@ -78,7 +78,7 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - run: uv sync --group dev
-      - run: uv run pytest --cov=src/zmqtt --cov-report=xml
+      - run: uv run pytest -vv --cov=src/zmqtt --cov-report=xml
       - uses: codecov/codecov-action@v5
         with:
           files: coverage.xml

.python-version

@@ -1 +1 @@
-3.11
+3.10

pyproject.toml

@@ -6,7 +6,7 @@ readme = "README.md"
 authors = [
     { name = "borisalekseev", email = "i.borisalekseev@gmail.com" }
 ]
-requires-python = ">=3.11"
+requires-python = ">=3.10"
 dependencies = []
 
 [build-system]
@@ -30,7 +30,13 @@ docs = [
 
 [tool.mypy]
 strict = true
-files = ["src"]
+files = ["src", "tests"]
+python_version = "3.10"
+
+[tool.ruff]
+src = ["src"]
+include = ["src/**.py", "tests/**.py"]
+target-version = "py310"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

src/zmqtt/_compat.py

@@ -0,0 +1,31 @@
+"""Python version compatibility helpers."""
+
+import asyncio
+import contextlib
+import sys
+from collections.abc import AsyncGenerator
+
+
+if sys.version_info >= (3, 11):
+
+    @contextlib.asynccontextmanager
+    async def defer_cancellation() -> AsyncGenerator[None, None]:
+        """Temporarily remove pending cancellations, restore them after cleanup."""
+        task = asyncio.current_task()
+        cancels = task.cancelling() if task else 0
+        for _ in range(cancels):
+            if task:
+                task.uncancel()
+        try:
+            yield
+        finally:
+            for _ in range(cancels):
+                if task:
+                    task.cancel()
+
+else:
+
+    @contextlib.asynccontextmanager
+    async def defer_cancellation() -> AsyncGenerator[None, None]:
+        """No-op on Python < 3.11: cancelling()/uncancel() are not available."""
+        yield

src/zmqtt/client.py

@@ -6,6 +6,7 @@
 from dataclasses import dataclass
 from typing import AsyncIterator, Final, Literal, Protocol, overload
 
+from zmqtt._compat import defer_cancellation
 from zmqtt.errors import MQTTDisconnectedError, MQTTTimeoutError
 from zmqtt.log import get_logger
 from zmqtt.packets.auth import Auth
@@ -142,8 +143,7 @@ async def __aenter__(self) -> "Subscription":
     async def __aexit__(self, *exc: object) -> None:
         self._client._subscriptions.remove(self)
         await self._cancel_relays()
-        task = asyncio.current_task()
-        being_cancelled = task is not None and task.cancelling() > 0
+        being_cancelled = isinstance(exc[1], asyncio.CancelledError)
         if (
             not being_cancelled
             and self._registered_filters
@@ -243,21 +243,14 @@ async def __aenter__(self) -> "MQTTClient":
         return self
 
     async def __aexit__(self, *exc: object) -> None:
-        task = asyncio.current_task()
-        cancels = task.cancelling() if task else 0
-        for _ in range(cancels):
-            task.uncancel()  # type: ignore[union-attr]
-        try:
+        async with defer_cancellation():
             if self._run_task is not None:
                 self._run_task.cancel()
                 await asyncio.gather(self._run_task, return_exceptions=True)
                 self._run_task = None
             if self._protocol is not None:
                 await self._protocol.disconnect()
                 self._protocol = None
-        finally:
-            for _ in range(cancels):
-                task.cancel()  # type: ignore[union-attr]
 
     async def publish(
         self,
@@ -356,18 +349,19 @@ async def _run_loop(self) -> None:
 
         while True:
             assert self._protocol is not None
+            protocol_run_task = asyncio.create_task(self._protocol.run())
             try:
                 # Run the protocol as a sub-task so _read_loop is live while we
                 # re-subscribe.  For the first connection subs_to_restore is empty,
                 # so this collapses to the original "await protocol.run()" pattern.
-                async with asyncio.TaskGroup() as tg:
-                    tg.create_task(self._protocol.run())
-                    if subs_to_restore:
-                        await asyncio.sleep(0)  # let _read_loop start
-                        for sub in subs_to_restore:
-                            await sub._reconnect(self._protocol)
-                        subs_to_restore = []
-            except* (MQTTDisconnectedError, MQTTTimeoutError):
+                if subs_to_restore:
+                    await self._protocol.started_event.wait()
+                    for sub in subs_to_restore:
+                        await sub._reconnect(self._protocol)
+                    subs_to_restore = []
+                await protocol_run_task
+
+            except (MQTTDisconnectedError, MQTTTimeoutError):
                 if not self._reconnect.enabled:
                     raise
                 # Close the dead transport to release the file descriptor.

src/zmqtt/protocol.py

@@ -98,9 +98,10 @@ def __init__(
         self._keepalive = keepalive
         self._ping_timeout = ping_timeout
         self._version: Final = version
-        self._buf: PacketBuffer = PacketBuffer(version=version)
+        self._buf = PacketBuffer(version=version)
         self._ping_waiters: list[asyncio.Future[None]] = []
-        self._disconnecting: bool = False
+        self._disconnecting = False
+        self.started_event = asyncio.Event()
 
     async def connect(self, packet: Connect) -> ConnAck:
         """Send CONNECT, read and return CONNACK. Raises on failure."""
@@ -122,11 +123,18 @@ async def connect(self, packet: Connect) -> ConnAck:
 
     async def run(self) -> None:
         """Run read loop and ping loop concurrently until disconnection."""
+        read_task = asyncio.create_task(self._read_loop())
+        ping_task = asyncio.create_task(self._ping_loop())
+        self.started_event.set()
         try:
-            async with asyncio.TaskGroup() as tg:
-                tg.create_task(self._read_loop())
-                tg.create_task(self._ping_loop())
+            await asyncio.gather(read_task, ping_task)
+        except BaseException:
+            read_task.cancel()
+            ping_task.cancel()
+            await asyncio.gather(read_task, ping_task, return_exceptions=True)
+            raise
         finally:
+            self.started_event.clear()
             self._cancel_pending()
 
     async def disconnect(self) -> None: