FIX: Live client error handling on connect

Author: nmacholl

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## 0.27.0 - TBD
+
+#### Breaking changes
+- Renamed `DatatbentoLiveProtocol.started` to `DatatbentoLiveProtocol.is_started` which now returns a bool instead of an `asyncio.Event`
+
+#### Bug fixes
+- Fixed an issue where an error message from the live gateway would not properly raise an exception if the connection closed before `Live.start` was called
+
 ## 0.26.0 - 2024-01-16
 
 This release adds support for transcoding DBN data into Apache parquet.

databento/live/protocol.py

@@ -82,7 +82,7 @@ def __init__(
 
         self._authenticated: asyncio.Future[int] = asyncio.Future()
         self._disconnected: asyncio.Future[None] = asyncio.Future()
-        self._started = asyncio.Event()
+        self._started: bool = False
 
     @property
     def authenticated(self) -> asyncio.Future[int]:
@@ -118,14 +118,14 @@ def disconnected(self) -> asyncio.Future[None]:
         return self._disconnected
 
     @property
-    def started(self) -> asyncio.Event:
+    def is_started(self) -> bool:
         """
-        Event that is set when the session has started streaming. This occurs
-        when the SessionStart message is sent to the gateway.
+        True if the session has started streaming. This occurs when the
+        SessionStart message is sent to the gateway.
 
         Returns
         -------
-        asyncio.Event
+        bool
 
         """
         return self._started
@@ -219,7 +219,7 @@ def buffer_updated(self, nbytes: int) -> None:
         logger.debug("read %d bytes from remote gateway", nbytes)
         data = self.__buffer[:nbytes]
 
-        if self.started.is_set():
+        if self.authenticated.done():
             self._process_dbn(data)
         else:
             self._process_gateway(data)
@@ -303,7 +303,7 @@ def start(
         """
         logger.debug("sending start")
         message = SessionStart()
-        self.started.set()
+        self._started = True
         self.transport.write(bytes(message))
 
     def _process_dbn(self, data: bytes) -> None:

databento/live/session.py

@@ -276,7 +276,8 @@ def is_reading(self) -> bool:
 
     def is_started(self) -> bool:
         """
-        Return true if the session's connection has started streaming.
+        Return true if the session's connection has started streaming, false
+        otherwise.
 
         Returns
         -------
@@ -286,7 +287,7 @@ def is_started(self) -> bool:
         with self._lock:
             if self._protocol is None:
                 return False
-            return self._protocol.started.is_set()
+            return self._protocol.is_started
 
     @property
     def metadata(self) -> databento_dbn.Metadata | None:

tests/test_live_protocol.py

@@ -90,7 +90,6 @@ async def test_protocol_connection_streaming(
     )
 
     protocol.start()
-    await asyncio.wait_for(protocol.started.wait(), timeout=1)
     await asyncio.wait_for(protocol.disconnected, timeout=1)
 
     # Assert