Register queue on streaming handle before sending the subscribe request

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Labone Python API Changelog
 
+## Version 3.2.1
+* Fix bug that caused subscriptions to potentially miss value updates after the subscription was registered but before the subscribe functions returned.
+
 ## Version 3.2.0
 * `subscribe` accepts keyword arguments, which are forwarded to the data-server.
   This allows to configure the subscription to the data-server.

pyproject.toml

@@ -103,12 +103,8 @@ target-version = "py39"
 [tool.ruff.lint]
 select = ["ALL"]
 ignore = [
-  # Type annotation for self
-  "ANN101",
   # Missing docstring in `__init__`
   "D107",
-  # Missing type annotation for `cls` in classmethod
-  "ANN102",
   # Missing docstring in magic method
   "D105",
   # Missing type annotation for `*args`

src/labone/__init__.py

@@ -6,8 +6,8 @@
 from labone.instrument import Instrument
 
 __all__ = [
-    "__version__",
-    "Instrument",
     "DataServer",
+    "Instrument",
     "ListNodesFlags",
+    "__version__",
 ]

src/labone/core/__init__.py

@@ -26,16 +26,16 @@
 
 __all__ = [
     "AnnotatedValue",
-    "ListNodesFlags",
-    "ListNodesInfoFlags",
-    "DataQueue",
     "CircularDataQueue",
+    "DataQueue",
     "DistinctConsecutiveDataQueue",
-    "Session",
+    "KernelInfo",
     "KernelSession",
+    "ListNodesFlags",
+    "ListNodesInfoFlags",
     "ServerInfo",
-    "KernelInfo",
-    "ZIContext",
-    "Value",
+    "Session",
     "ShfGeneratorWaveformVectorData",
+    "Value",
+    "ZIContext",
 ]

src/labone/core/session.py

@@ -853,23 +853,28 @@ async def subscribe(
             },
         }
         if get_initial_value:
-            _, initial_value = await asyncio.gather(
-                self._session.subscribe(subscription=subscription),
-                self.get(path),
-            )
             new_queue_type = queue_type or DataQueue
             queue = new_queue_type(
                 path=path,
                 handle=streaming_handle,
             )
-            queue.put_nowait(initial_value)
+            _, initial_value = await asyncio.gather(
+                self._session.subscribe(subscription=subscription),
+                self.get(path),
+            )
+            # If the queue already has received a update event we do not
+            # need to put the initial value in the queue. As it may break the
+            # order.
+            if queue.empty():
+                queue.put_nowait(initial_value)
             return queue
-        await self._session.subscribe(subscription=subscription)
         new_queue_type = queue_type or DataQueue
-        return new_queue_type(
+        queue = new_queue_type(
             path=path,
             handle=streaming_handle,
         )
+        await self._session.subscribe(subscription=subscription)
+        return queue
 
     async def wait_for_state_change(
         self,

src/labone/server/__init__.py

@@ -8,7 +8,7 @@
 
 from labone.mock.session import LabOneServerBase, MockSession, Subscription
 
-__all__ = ["CapnpServer", "LabOneServerBase", "Subscription", "MockSession"]
+__all__ = ["CapnpServer", "LabOneServerBase", "MockSession", "Subscription"]
 
 import warnings