chore: Introducing Synchronizer protocol & streaming implementation

Author: keelerm84

ldclient/impl/datasource/streaming.py

@@ -14,6 +14,7 @@
 )
 from ld_eventsource.errors import HTTPStatusError
 
+from ldclient.impl.datasystem.protocolv2 import ChangeSet
 from ldclient.impl.http import HTTPFactory, _http_factory
 from ldclient.impl.util import (
     http_error_message,
@@ -36,9 +37,9 @@
 BACKOFF_RESET_INTERVAL = 60
 JITTER_RATIO = 0.5
 
-STREAM_ALL_PATH = '/all'
+STREAM_ALL_PATH = "/all"
 
-ParsedPath = namedtuple('ParsedPath', ['kind', 'key'])
+ParsedPath = namedtuple("ParsedPath", ["kind", "key"])
 
 
 class StreamingUpdateProcessor(Thread, UpdateProcessor):
@@ -47,7 +48,7 @@ def __init__(self, config, store, ready, diagnostic_accumulator):
         self.daemon = True
         self._uri = config.stream_base_uri + STREAM_ALL_PATH
         if config.payload_filter_key is not None:
-            self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
+            self._uri += "?%s" % parse.urlencode({"filter": config.payload_filter_key})
         self._config = config
         self._data_source_update_sink = config.data_source_update_sink
         self._store = store
@@ -61,31 +62,42 @@ def run(self):
         self._running = True
         self._sse = self._create_sse_client()
         self._connection_attempt_start_time = time.time()
+
         for action in self._sse.all:
             if isinstance(action, Event):
-                message_ok = False
+                change_set: Optional[ChangeSet] = None
                 try:
-                    message_ok = self._process_message(self._sink_or_store(), action)
+                    change_set = self._process_message(self._sink_or_store(), action)
                 except json.decoder.JSONDecodeError as e:
-                    log.info("Error while handling stream event; will restart stream: %s" % e)
+                    log.info(
+                        "Error while handling stream event; will restart stream: %s" % e
+                    )
                     self._sse.interrupt()
 
                     self._handle_error(e)
                 except Exception as e:
-                    log.info("Error while handling stream event; will restart stream: %s" % e)
+                    log.info(
+                        "Error while handling stream event; will restart stream: %s" % e
+                    )
                     self._sse.interrupt()
 
                     if self._data_source_update_sink is not None:
-                        error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e))
+                        error_info = DataSourceErrorInfo(
+                            DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)
+                        )
 
-                        self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+                        self._data_source_update_sink.update_status(
+                            DataSourceState.INTERRUPTED, error_info
+                        )
 
-                if message_ok:
+                if change_set is not None:
                     self._record_stream_init(False)
                     self._connection_attempt_start_time = None
 
                     if self._data_source_update_sink is not None:
-                        self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+                        self._data_source_update_sink.update_status(
+                            DataSourceState.VALID, None
+                        )
 
                     if not self._ready.is_set():
                         log.info("StreamingUpdateProcessor initialized ok.")
@@ -104,19 +116,32 @@ def _record_stream_init(self, failed: bool):
         if self._diagnostic_accumulator and self._connection_attempt_start_time:
             current_time = int(time.time() * 1000)
             elapsed = current_time - int(self._connection_attempt_start_time * 1000)
-            self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed)
+            self._diagnostic_accumulator.record_stream_init(
+                current_time, elapsed if elapsed >= 0 else 0, failed
+            )
 
     def _create_sse_client(self) -> SSEClient:
         # We don't want the stream to use the same read timeout as the rest of the SDK.
         http_factory = _http_factory(self._config)
-        stream_http_factory = HTTPFactory(http_factory.base_headers, http_factory.http_config, override_read_timeout=stream_read_timeout)
+        stream_http_factory = HTTPFactory(
+            http_factory.base_headers,
+            http_factory.http_config,
+            override_read_timeout=stream_read_timeout,
+        )
         return SSEClient(
             connect=ConnectStrategy.http(
-                url=self._uri, headers=http_factory.base_headers, pool=stream_http_factory.create_pool_manager(1, self._uri), urllib3_request_options={"timeout": stream_http_factory.timeout}
+                url=self._uri,
+                headers=http_factory.base_headers,
+                pool=stream_http_factory.create_pool_manager(1, self._uri),
+                urllib3_request_options={"timeout": stream_http_factory.timeout},
             ),
             error_strategy=ErrorStrategy.always_continue(),  # we'll make error-handling decisions when we see a Fault
             initial_retry_delay=self._config.initial_reconnect_delay,
-            retry_delay_strategy=RetryDelayStrategy.default(max_delay=MAX_RETRY_DELAY, backoff_multiplier=2, jitter_multiplier=JITTER_RATIO),
+            retry_delay_strategy=RetryDelayStrategy.default(
+                max_delay=MAX_RETRY_DELAY,
+                backoff_multiplier=2,
+                jitter_multiplier=JITTER_RATIO,
+            ),
             retry_delay_reset_threshold=BACKOFF_RESET_INTERVAL,
             logger=log,
         )
@@ -142,39 +167,54 @@ def _sink_or_store(self):
         return self._data_source_update_sink
 
     def initialized(self):
-        return self._running and self._ready.is_set() is True and self._store.initialized is True
+        return (
+            self._running
+            and self._ready.is_set() is True
+            and self._store.initialized is True
+        )
 
     # Returns True if we initialized the feature store
     def _process_message(self, store, msg: Event) -> bool:
-        if msg.event == 'put':
+        if msg.event == "put":
             all_data = json.loads(msg.data)
-            init_data = {FEATURES: all_data['data']['flags'], SEGMENTS: all_data['data']['segments']}
-            log.debug("Received put event with %d flags and %d segments", len(init_data[FEATURES]), len(init_data[SEGMENTS]))
+            init_data = {
+                FEATURES: all_data["data"]["flags"],
+                SEGMENTS: all_data["data"]["segments"],
+            }
+            log.debug(
+                "Received put event with %d flags and %d segments",
+                len(init_data[FEATURES]),
+                len(init_data[SEGMENTS]),
+            )
             store.init(init_data)
             return True
-        elif msg.event == 'patch':
+        elif msg.event == "patch":
             payload = json.loads(msg.data)
-            path = payload['path']
-            obj = payload['data']
-            log.debug("Received patch event for %s, New version: [%d]", path, obj.get("version"))
+            path = payload["path"]
+            obj = payload["data"]
+            log.debug(
+                "Received patch event for %s, New version: [%d]",
+                path,
+                obj.get("version"),
+            )
             target = StreamingUpdateProcessor._parse_path(path)
             if target is not None:
                 store.upsert(target.kind, obj)
             else:
                 log.warning("Patch for unknown path: %s", path)
-        elif msg.event == 'delete':
+        elif msg.event == "delete":
             payload = json.loads(msg.data)
-            path = payload['path']
+            path = payload["path"]
             # noinspection PyShadowingNames
-            version = payload['version']
+            version = payload["version"]
             log.debug("Received delete event for %s, New version: [%d]", path, version)
             target = StreamingUpdateProcessor._parse_path(path)
             if target is not None:
                 store.delete(target.kind, target.key, version)
             else:
                 log.warning("Delete for unknown path: %s", path)
         else:
-            log.warning('Unhandled event in stream processor: ' + msg.event)
+            log.warning("Unhandled event in stream processor: " + msg.event)
         return False
 
     # Returns true to continue, false to stop
@@ -183,21 +223,32 @@ def _handle_error(self, error: Exception) -> bool:
             return False  # don't retry if we've been deliberately stopped
 
         if isinstance(error, json.decoder.JSONDecodeError):
-            error_info = DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error))
+            error_info = DataSourceErrorInfo(
+                DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)
+            )
 
             log.error("Unexpected error on stream connection: %s, will retry" % error)
             self._record_stream_init(True)
             self._connection_attempt_start_time = None
 
             if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+                self._data_source_update_sink.update_status(
+                    DataSourceState.INTERRUPTED, error_info
+                )
         elif isinstance(error, HTTPStatusError):
             self._record_stream_init(True)
             self._connection_attempt_start_time = None
 
-            error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, error.status, time.time(), str(error))
+            error_info = DataSourceErrorInfo(
+                DataSourceErrorKind.ERROR_RESPONSE,
+                error.status,
+                time.time(),
+                str(error),
+            )
 
-            http_error_message_result = http_error_message(error.status, "stream connection")
+            http_error_message_result = http_error_message(
+                error.status, "stream connection"
+            )
             if not is_http_error_recoverable(error.status):
                 log.error(http_error_message_result)
                 self._ready.set()  # if client is initializing, make it stop waiting; has no effect if already inited
@@ -208,14 +259,21 @@ def _handle_error(self, error: Exception) -> bool:
                 log.warning(http_error_message_result)
 
                 if self._data_source_update_sink is not None:
-                    self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+                    self._data_source_update_sink.update_status(
+                        DataSourceState.INTERRUPTED, error_info
+                    )
         else:
             log.warning("Unexpected error on stream connection: %s, will retry" % error)
             self._record_stream_init(True)
             self._connection_attempt_start_time = None
 
             if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error)))
+                self._data_source_update_sink.update_status(
+                    DataSourceState.INTERRUPTED,
+                    DataSourceErrorInfo(
+                        DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error)
+                    ),
+                )
             # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of urllib3 internals
         self._connection_attempt_start_time = time.time() + self._sse.next_retry_delay
         return True

ldclient/impl/datasourcev2/__init__.py

@@ -1,6 +1,79 @@
 """
 This module houses FDv2 types and implementations of synchronizers and
 initializers for the datasystem.
+
+All types and implementations in this module are considered internal
+and are not part of the public API of the LaunchDarkly Python SDK.
+They are subject to change without notice and should not be used directly
+by users of the SDK.
+
+You have been warned.
 """
 
+from abc import abstractmethod
+from dataclasses import dataclass
+from typing import Iterable, Mapping, Optional, Protocol, Tuple
+
+from ldclient.impl.datasystem.protocolv2 import ChangeSet, Selector
+from ldclient.impl.util import _Result
+from ldclient.interfaces import DataSourceErrorInfo, DataSourceState
+
+PollingResult = _Result[Tuple[ChangeSet, Mapping], str]
+
+
+class PollingRequester(Protocol):  # pylint: disable=too-few-public-methods
+    """
+    PollingRequester allows PollingDataSource to delegate fetching data to
+    another component.
+
+    This is useful for testing the PollingDataSource without needing to set up
+    a test HTTP server.
+    """
+
+    @abstractmethod
+    def fetch(self, selector: Optional[Selector]) -> PollingResult:
+        """
+        Fetches the data for the given selector.
+        Returns a Result containing a tuple of ChangeSet and any request headers,
+        or an error if the data could not be retrieved.
+        """
+        raise NotImplementedError
+
+
+@dataclass(frozen=True)
+class Update:
+    """
+    Update represents the results of a synchronizer's ongoing sync
+    method.
+    """
+
+    state: DataSourceState
+    change_set: Optional[ChangeSet] = None
+    error: Optional[DataSourceErrorInfo] = None
+    revert_to_fdv1: bool = False
+    # NOTE: I hate this. Let's make this some sort of metadata object so we can
+    # retrieve all sorts of information.
+    environment_id: Optional[str] = None
+
+
+class Synchronizer(Protocol):  # pylint: disable=too-few-public-methods
+    """
+    Synchronizer represents a component capable of synchronizing data from an external
+    data source, such as a streaming or polling API.
+
+    It is responsible for yielding Update objects that represent the current state
+    of the data source, including any changes that have occurred since the last
+    synchronization.
+    """
+
+    @abstractmethod
+    def sync(self) -> Iterable[Update]:
+        """
+        sync should begin the synchronization process for the data source, yielding
+        Update objects until the connection is closed or an unrecoverable error
+        occurs.
+        """
+        raise NotImplementedError
+
+
 __all__: list[str] = []

ldclient/impl/datasourcev2/polling.py

@@ -1,16 +1,16 @@
 """
-Default implementation of the polling synchronizer and initializer.
+This module contains the implementations of a polling synchronizer and
+initializer, along with any required supporting classes and protocols.
 """
 
 import json
-from abc import abstractmethod
 from collections import namedtuple
-from collections.abc import Mapping
-from typing import Optional, Protocol, Tuple
+from typing import Iterable, Optional
 from urllib import parse
 
 import urllib3
 
+from ldclient.impl.datasourcev2 import PollingRequester, PollingResult, Update
 from ldclient.impl.datasystem.protocolv2 import (
     Basis,
     ChangeSet,
@@ -38,27 +38,6 @@
 
 POLLING_ENDPOINT = "/sdk/poll"
 
-PollingResult = _Result[Tuple[ChangeSet, Mapping], str]
-
-
-class PollingRequester(Protocol):  # pylint: disable=too-few-public-methods
-    """
-    PollingRequester allows PollingDataSource to delegate fetching data to
-    another component.
-
-    This is useful for testing the PollingDataSource without needing to set up
-    a test HTTP server.
-    """
-
-    @abstractmethod
-    def fetch(self, selector: Optional[Selector]) -> PollingResult:
-        """
-        Fetches the data for the given selector.
-        Returns a Result containing a tuple of ChangeSet and any request headers,
-        or an error if the data could not be retrieved.
-        """
-        raise NotImplementedError
-
 
 CacheEntry = namedtuple("CacheEntry", ["data", "etag"])