fix: Close the connection client on SSEClient.close() (#67)

Author: jsonbailey

README.md

@@ -22,6 +22,11 @@ An async implementation, `AsyncSSEClient`, is also available for use with `async
 pip install launchdarkly-eventsource[async]
 ```
 
+> [!CAUTION]
+> The async implementation (`AsyncSSEClient` and its associated `AsyncConnectStrategy` API) is experimental and should NOT be considered ready for production use. It may change or be removed without notice and is not subject to backwards compatibility guarantees.
+>
+> Pin to a specific minor version and review the changelog before upgrading.
+
 ## Supported Python versions
 
 This version of the package is compatible with Python 3.9 and higher.

ld_eventsource/async_client.py

@@ -16,6 +16,12 @@ class AsyncSSEClient:
     """
     An async client for reading a Server-Sent Events stream.
 
+    .. caution::
+        This feature is experimental and should NOT be considered ready for production
+        use. It may change or be removed without notice and is not subject to backwards
+        compatibility guarantees. Pin to a specific minor version and review the changelog
+        before upgrading.
+
     This is an async/await implementation. The expected usage is to create an ``AsyncSSEClient``
     instance (either as an async context manager or directly), then read from it using the async
     iterator properties :attr:`events` or :attr:`all`.

ld_eventsource/config/async_connect_strategy.py

@@ -10,6 +10,12 @@ class AsyncConnectStrategy:
     """
     An abstraction for how :class:`.AsyncSSEClient` should obtain an input stream.
 
+    .. caution::
+        This feature is experimental and should NOT be considered ready for production
+        use. It may change or be removed without notice and is not subject to backwards
+        compatibility guarantees. Pin to a specific minor version and review the changelog
+        before upgrading.
+
     The default implementation is :meth:`http()`, which makes HTTP requests with ``aiohttp``.
     Or, if you want to consume an input stream from some other source, you can create your own
     subclass of :class:`AsyncConnectStrategy`.
@@ -58,6 +64,12 @@ class AsyncConnectionClient:
     """
     An object provided by :class:`.AsyncConnectStrategy` that is retained by a single
     :class:`.AsyncSSEClient` to perform all connection attempts by that instance.
+
+    .. caution::
+        This feature is experimental and should NOT be considered ready for production
+        use. It may change or be removed without notice and is not subject to backwards
+        compatibility guarantees. Pin to a specific minor version and review the changelog
+        before upgrading.
     """
 
     async def connect(self, last_event_id: Optional[str]) -> AsyncConnectionResult:
@@ -85,6 +97,12 @@ async def __aexit__(self, type, value, traceback):
 class AsyncConnectionResult:
     """
     The return type of :meth:`AsyncConnectionClient.connect()`.
+
+    .. caution::
+        This feature is experimental and should NOT be considered ready for production
+        use. It may change or be removed without notice and is not subject to backwards
+        compatibility guarantees. Pin to a specific minor version and review the changelog
+        before upgrading.
     """
 
     def __init__(

ld_eventsource/reader.py

@@ -66,7 +66,6 @@ def __init__(
     def last_event_id(self):
         return self._last_event_id
 
-    @property
     def events_and_comments(self):
         event_type = ""
         event_data = None

ld_eventsource/sse_client.py

@@ -95,7 +95,7 @@ def __init__(
         if isinstance(connect, str):
             connect = ConnectStrategy.http(connect)
         elif not isinstance(connect, ConnectStrategy):
-            raise TypeError("request must be either a string or ConnectStrategy")
+            raise TypeError("connect must be either a string or ConnectStrategy")
 
         self.__base_retry_delay = initial_retry_delay
         self.__base_retry_delay_strategy = (
@@ -116,7 +116,7 @@ def __init__(
             logger.propagate = False
         self.__logger = logger
 
-        self.__connection_client = connect.create_client(logger)
+        self.__connection_client: ConnectionClient = connect.create_client(logger)
         self.__connection_result: Optional[ConnectionResult] = None
         self._retry_reset_baseline: float = 0
         self.__disconnected_time: float = 0
@@ -151,6 +151,7 @@ def close(self):
         """
         self.__closed = True
         self.interrupt()
+        self.__connection_client.close()
 
     def interrupt(self):
         """
@@ -181,6 +182,22 @@ def all(self) -> Iterable[Action]:
         already active, so you do not need to call :meth:`start()` unless you want to verify that
         the stream is connected before trying to read events.
         """
+        return self._all_generator()
+
+    @property
+    def events(self) -> Iterable[Event]:
+        """
+        An iterable series of :class:`.Event` objects received from the stream.
+
+        Use :attr:`all` instead if you also want to know about other kinds of occurrences.
+
+        Iterating over this property automatically starts or restarts the stream if it is not
+        already active, so you do not need to call :meth:`start()` unless you want to verify that
+        the stream is connected before trying to read events.
+        """
+        return self._events_generator()
+
+    def _all_generator(self):
         while True:
             # Reading implies starting the stream if it isn't already started. We might also
             # be restarting since we could have been interrupted at any time.
@@ -194,11 +211,12 @@ def all(self) -> Iterable[Action]:
             reader = _SSEReader(lines, self.__last_event_id, None)
             error: Optional[Exception] = None
             try:
-                for ec in reader.events_and_comments:
+                for ec in reader.events_and_comments():
+                    self.__last_event_id = reader.last_event_id
                     yield ec
                     if self.__interrupted:
                         break
-                # If we finished iterating all of reader.events_and_comments, it means the stream
+                # If we finished iterating all of reader.events_and_comments(), it means the stream
                 # was closed without an error.
                 self.__connection_result = None
             except Exception as e:
@@ -226,18 +244,8 @@ def all(self) -> Iterable[Action]:
             yield Fault(error)
             continue  # try to connect again
 
-    @property
-    def events(self) -> Iterable[Event]:
-        """
-        An iterable series of :class:`.Event` objects received from the stream.
-
-        Use :attr:`all` instead if you also want to know about other kinds of occurrences.
-
-        Iterating over this property automatically starts or restarts the stream if it is not
-        already active, so you do not need to call :meth:`start()` unless you want to verify that
-        the stream is connected before trying to read events.
-        """
-        for item in self.all:
+    def _events_generator(self):
+        for item in self._all_generator():
             if isinstance(item, Event):
                 yield item
 

ld_eventsource/testing/test_reader.py

@@ -55,7 +55,7 @@ def test_mixed_terminators(self):
 
 class TestSSEReader:
     def expect_output(self, lines, expected):
-        output = list(_SSEReader(lines).events_and_comments)
+        output = list(_SSEReader(lines).events_and_comments())
         assert output == expected
 
     def test_parses_event_with_all_fields(self):
@@ -98,12 +98,12 @@ def store_retry(value):
             got_retry = value
 
         lines = ["retry: 1000"]
-        list(_SSEReader(lines, None, store_retry).events_and_comments)
+        list(_SSEReader(lines, None, store_retry).events_and_comments())
         assert got_retry == 1000
 
     def test_ignores_retry_interval_if_no_callback_given(self):
         lines = ["retry: 1000"]
-        list(_SSEReader(lines, None, None).events_and_comments)
+        list(_SSEReader(lines, None, None).events_and_comments())
 
     def test_remembers_last_event_id(self):
         lines = [
@@ -124,5 +124,5 @@ def test_remembers_last_event_id(self):
             Event("message", "third", None, "b"),
             Event("message", "fourth", "", ""),
         ]
-        output = list(_SSEReader(lines, last_event_id="a").events_and_comments)
+        output = list(_SSEReader(lines, last_event_id="a").events_and_comments())
         assert output == expected

ld_eventsource/testing/test_sync_async_parity.py

@@ -0,0 +1,56 @@
+"""
+Parity guard for the hand-maintained synchronous and asynchronous client pairs.
+
+The sync and async implementations in this package are written and maintained by hand as
+parallel files. This test asserts that each public sync/async pair exposes the same public
+surface, so that a method or property added to one side but forgotten on the other fails CI.
+
+The public surface includes properties as well as methods, so the comparison is over all
+public attribute names (anything not starting with ``_``) rather than callables only.
+"""
+
+import pytest
+
+from ld_eventsource.async_client import AsyncSSEClient
+from ld_eventsource.config.async_connect_strategy import (
+    AsyncConnectionClient, AsyncConnectionResult, AsyncConnectStrategy)
+from ld_eventsource.config.connect_strategy import (ConnectionClient,
+                                                    ConnectionResult,
+                                                    ConnectStrategy)
+from ld_eventsource.sse_client import SSEClient
+
+
+def _public_names(cls):
+    return {name for name in dir(cls) if not name.startswith("_")} - set(dir(object))
+
+
+# Each param pairs a public synchronous class with its asynchronous counterpart, plus
+# allowlists of names intentionally present on only one side. Every allowlisted name must
+# carry a justification; if the test flags a genuinely one-sided member, add it here with a
+# comment rather than weakening the comparison.
+@pytest.mark.parametrize(
+    "sync_cls, async_cls, sync_only, async_only",
+    [
+        pytest.param(SSEClient, AsyncSSEClient, set(), set(), id="SSEClient/AsyncSSEClient"),
+        pytest.param(ConnectStrategy, AsyncConnectStrategy, set(), set(), id="ConnectStrategy/AsyncConnectStrategy"),
+        pytest.param(ConnectionClient, AsyncConnectionClient, set(), set(), id="ConnectionClient/AsyncConnectionClient"),
+        pytest.param(ConnectionResult, AsyncConnectionResult, set(), set(), id="ConnectionResult/AsyncConnectionResult"),
+    ],
+)
+def test_sync_async_public_surface_matches(sync_cls, async_cls, sync_only, async_only):
+    sync_names = _public_names(sync_cls)
+    async_names = _public_names(async_cls)
+
+    sync_only_diff = sync_names - async_names - sync_only
+    async_only_diff = async_names - sync_names - async_only
+
+    assert not sync_only_diff, (
+        f"{sync_cls.__name__} exposes public names with no counterpart on "
+        f"{async_cls.__name__}: {sorted(sync_only_diff)}. Add the missing member to the "
+        f"async class, or allowlist it in 'sync_only' with a justification."
+    )
+    assert not async_only_diff, (
+        f"{async_cls.__name__} exposes public names with no counterpart on "
+        f"{sync_cls.__name__}: {sorted(async_only_diff)}. Add the missing member to the "
+        f"sync class, or allowlist it in 'async_only' with a justification."
+    )