factor out HTTP logic with ConnectStrategy abstraction (#6)

Author: eli-darkly

README.md

@@ -14,7 +14,7 @@ This package's primary purpose is to support the [LaunchDarkly SDK for Python](h
 * Setting read timeouts, custom headers, and other HTTP request properties.
 * Specifying that connections should be retried under circumstances where the standard EventSource behavior would not retry them, such as if the server returns an HTTP error status.
 
-This is a synchronous implementation which blocks the caller's thread when reading events or reconnecting.
+This is a synchronous implementation which blocks the caller's thread when reading events or reconnecting. By default, it uses `urllib3` to make HTTP requests, but it can be configured to read any input stream.
 
 ## Supported Python versions
 

contract-tests/stream_entity.py

@@ -35,15 +35,15 @@ def run(self):
         stream_url = self.options["streamUrl"]
         try:
             self.log.info('Opening stream from %s', stream_url)
-            request = RequestParams(
+            connect = ConnectStrategy.http(
                 url=stream_url,
                 headers=self.options.get("headers"),
                 urllib3_request_options=None if self.options.get("readTimeoutMs") is None else {
                     "timeout": urllib3.Timeout(read=millis_to_seconds(self.options.get("readTimeoutMs")))
                 }
             )                    
             sse = SSEClient(
-                request,
+                connect,
                 initial_retry_delay=millis_to_seconds(self.options.get("initialDelayMs")),
                 last_event_id=self.options.get("lastEventId"),
                 error_strategy=ErrorStrategy.from_lambda(lambda _:

ld_eventsource/__init__.py

@@ -1,2 +1 @@
-from ld_eventsource.request_params import RequestParams
 from ld_eventsource.sse_client import *

ld_eventsource/config/__init__.py

@@ -1,2 +1,3 @@
+from .connect_strategy import ConnectStrategy, ConnectionClient, ConnectionResult
 from .error_strategy import ErrorStrategy
 from .retry_delay_strategy import RetryDelayStrategy

ld_eventsource/config/connect_strategy.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+from logging import Logger
+from typing import Callable, Iterator, Optional, Union
+from urllib3 import PoolManager
+
+from ld_eventsource.http import _HttpClientImpl, _HttpConnectParams
+
+
+class ConnectStrategy:
+    """
+    An abstraction for how :class:`.SSEClient` should obtain an input stream.
+    
+    The default implementation is :meth:`http()`, which makes HTTP requests with ``urllib3``.
+    Or, if you want to consume an input stream from some other source, you can create your own
+    subclass of :class:`ConnectStrategy`.
+
+    Instances of this class should be immutable and should not contain any state that is specific
+    to one active stream. The :class:`ConnectionClient` that they produce is stateful and belongs
+    to a single :class:`.SSEClient`.
+    """
+
+    def create_client(self, logger: Logger) -> ConnectionClient:
+        """
+        Creates a client instance.
+
+        This is called once when an :class:`.SSEClient` is created. The SSEClient returns the
+        returned :class:`ConnectionClient` and uses it to perform all subsequent connection attempts.
+
+        :param logger: the logger being used by the SSEClient
+        """
+        raise NotImplementedError("ConnectStrategy base class cannot be used by itself")
+    
+    @staticmethod
+    def http(
+        url: str,
+        headers: Optional[dict]=None,
+        pool: Optional[PoolManager]=None,
+        urllib3_request_options: Optional[dict]=None
+    ) -> ConnectStrategy:
+        """
+        Creates the default HTTP implementation, specifying request parameters.
+
+        :param url: the stream URL
+        :param headers: optional HTTP headers to add to the request
+        :param pool: optional urllib3 ``PoolManager`` to provide an HTTP client
+        :param urllib3_request_options: optional ``kwargs`` to add to the ``request`` call; these
+            can include any parameters supported by ``urllib3``, such as ``timeout``
+        """
+        return _HttpConnectStrategy(_HttpConnectParams(url, headers, pool, urllib3_request_options))
+
+
+class ConnectionClient:
+    """
+    An object provided by :class:`.ConnectStrategy` that is retained by a single
+    :class:`.SSEClient` to perform all connection attempts by that instance.
+
+    For the default HTTP implementation, this represents an HTTP connection pool.
+    """
+
+    def connect(self, last_event_id: Optional[str]) -> ConnectionResult:
+        """
+        Attempts to connect to a stream. Raises an exception if unsuccessful.
+
+        :param last_event_id: the current value of :attr:`SSEClient.last_event_id`
+        (should be sent to the server to support resuming an interrupted stream)
+        :return: a :class:`ConnectionResult` representing the stream
+        """
+        raise NotImplementedError("ConnectionClient base class cannot be used by itself")
+    
+    def close(self):
+        """
+        Does whatever is necessary to release resources when the SSEClient is closed.
+        """
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.close()
+
+
+
+class ConnectionResult:
+    """
+    The return type of :meth:`ConnectionClient.connect()`.
+    """
+    def __init__(
+        self,
+        stream: Iterator[bytes],
+        closer: Optional[Callable]
+    ):
+        self.__stream = stream
+        self.__closer = closer
+    
+    @property
+    def stream(self) -> Iterator[bytes]:
+        """
+        An iterator that returns chunks of data.
+        """
+        return self.__stream
+
+    def close(self):
+        """
+        Does whatever is necessary to release the connection.
+        """
+        if self.__closer:
+            self.__closer()
+            self.__closer = None
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.close()
+
+
+# _HttpConnectStrategy and _HttpConnectionClient are defined here rather than in http.py to avoid
+# a circular module reference.
+
+class _HttpConnectStrategy(ConnectStrategy):
+    def __init__(self, params: _HttpConnectParams):
+        self.__params = params
+    
+    def create_client(self, logger: Logger) -> ConnectionClient:
+        return _HttpConnectionClient(self.__params, logger)
+
+
+class _HttpConnectionClient(ConnectionClient):
+    def __init__(self, params: _HttpConnectParams, logger: Logger):
+        self.__impl = _HttpClientImpl(params, logger)
+
+    def connect(self, last_event_id: Optional[str]) -> ConnectionResult:
+        stream, closer = self.__impl.connect(last_event_id)
+        return ConnectionResult(stream, closer)
+
+    def close(self):
+        self.__impl.close()
+
+
+__all__ = ['ConnectStrategy', 'ConnectionClient', 'ConnectionResult']

ld_eventsource/http.py

@@ -0,0 +1,84 @@
+from logging import Logger
+from typing import Callable, Iterator, Optional, Tuple
+from urllib3 import PoolManager
+from urllib3.exceptions import MaxRetryError
+from urllib3.util import Retry
+
+from ld_eventsource.errors import HTTPContentTypeError, HTTPStatusError
+
+_CHUNK_SIZE = 10000
+
+
+class _HttpConnectParams:
+    def __init__(
+        self,
+        url: str,
+        headers: Optional[dict]=None,
+        pool: Optional[PoolManager]=None,
+        urllib3_request_options: Optional[dict]=None
+    ):
+        self.__url = url
+        self.__headers = headers
+        self.__pool = pool
+        self.__urllib3_request_options = urllib3_request_options
+
+    @property
+    def url(self) -> str:
+        return self.__url
+    
+    @property
+    def headers(self) -> Optional[dict]:
+        return self.__headers
+    
+    @property
+    def pool(self) -> Optional[PoolManager]:
+        return self.__pool
+    
+    @property
+    def urllib3_request_options(self) -> Optional[dict]:
+        return self.__urllib3_request_options
+
+
+class _HttpClientImpl:
+    def __init__(self, params: _HttpConnectParams, logger: Logger):
+        self.__params = params
+        self.__pool = params.pool or PoolManager()
+        self.__should_close_pool = params.pool is not None
+        self.__logger = logger
+    
+    def connect(self, last_event_id: Optional[str]) -> Tuple[Iterator[bytes], Callable]:
+        self.__logger.info("Connecting to stream at %s" % self.__params.url)
+        
+        headers = self.__params.headers.copy() if self.__params.headers else {}
+        headers['Cache-Control'] = 'no-cache'
+        headers['Accept'] = 'text/event-stream'
+
+        if last_event_id:
+            headers['Last-Event-ID'] = last_event_id
+
+        request_options = self.__params.urllib3_request_options.copy() if self.__params.urllib3_request_options else {}
+        request_options['headers'] = headers
+
+        try:
+            resp = self.__pool.request(
+                'GET',
+                self.__params.url,
+                preload_content=False,
+                retries=Retry(total=None, read=0, connect=0, status=0, other=0, redirect=3),
+                **request_options)
+        except MaxRetryError as e:
+            reason: Optional[Exception] = e.reason
+            if reason is not None:
+                raise reason  # e.reason is the underlying I/O error        
+        if resp.status >= 400 or resp.status == 204:
+            raise HTTPStatusError(resp.status)
+        content_type = resp.headers.get('Content-Type', None)
+        if content_type is None or not str(content_type).startswith("text/event-stream"):
+            raise HTTPContentTypeError(content_type or '')
+
+        stream = resp.stream(_CHUNK_SIZE)
+        return stream, resp.release_conn
+
+    def close(self):
+        if self.__should_close_pool:
+            self.__pool.clear()