feat: streaming connector (#368)

Author: Tymek

UnleashClient/__init__.py

@@ -20,6 +20,7 @@
     BootstrapConnector,
     OfflineConnector,
     PollingConnector,
+    StreamingConnector,
 )
 from UnleashClient.constants import (
     DISABLED_VARIATION,
@@ -43,6 +44,11 @@
 from .cache import BaseCache, FileCache
 from .utils import LOGGER, InstanceAllowType, InstanceCounter
 
+try:
+    from typing import Literal, TypedDict
+except ImportError:
+    from typing_extensions import Literal, TypedDict  # type: ignore
+
 INSTANCES = InstanceCounter()
 _BASE_CONTEXT_FIELDS = [
     "userId",
@@ -55,6 +61,10 @@
 ]
 
 
+class ExperimentalMode(TypedDict, total=False):
+    type: Literal["streaming", "polling"]
+
+
 def build_ready_callback(
     event_callback: Optional[Callable[[BaseEvent], None]] = None,
 ) -> Optional[Callable]:
@@ -113,6 +123,7 @@ class UnleashClient:
     :param scheduler_executor: Name of APSCheduler executor to use if using a custom scheduler.
     :param multiple_instance_mode: Determines how multiple instances being instantiated is handled by the SDK, when set to InstanceAllowType.BLOCK, the client constructor will fail when more than one instance is detected, when set to InstanceAllowType.WARN, multiple instances will be allowed but log a warning, when set to InstanceAllowType.SILENTLY_ALLOW, no warning or failure will be raised when instantiating multiple instances of the client. Defaults to InstanceAllowType.WARN
     :param event_callback: Function to call if impression events are enabled.  WARNING: Depending on your event library, this may have performance implications!
+    :param experimental_mode: Optional dict to configure mode. Use {"type": "streaming"} to enable streaming or {"type": "polling"} (default).
     """
 
     def __init__(
@@ -140,6 +151,7 @@ def __init__(
         scheduler_executor: Optional[str] = None,
         multiple_instance_mode: InstanceAllowType = InstanceAllowType.WARN,
         event_callback: Optional[Callable[[BaseEvent], None]] = None,
+        experimental_mode: Optional[ExperimentalMode] = None,
     ) -> None:
         custom_headers = custom_headers or {}
         custom_options = custom_options or {}
@@ -173,6 +185,7 @@ def __init__(
         self.unleash_verbose_log_level = verbose_log_level
         self.unleash_event_callback = event_callback
         self._ready_callback = build_ready_callback(event_callback)
+        self.connector_mode: ExperimentalMode = experimental_mode or {"type": "polling"}
 
         self._do_instance_check(multiple_instance_mode)
 
@@ -279,6 +292,7 @@ def initialize_client(self, fetch_toggles: bool = True) -> None:
                     **self.unleash_custom_headers,
                     "unleash-connection-id": self.connection_id,
                     "unleash-appname": self.unleash_app_name,
+                    "unleash-instanceid": self.unleash_instance_id,
                     "unleash-sdk": f"{SDK_NAME}:{SDK_VERSION}",
                 }
 
@@ -295,8 +309,19 @@ def initialize_client(self, fetch_toggles: bool = True) -> None:
                         self.strategy_mapping,
                         self.unleash_request_timeout,
                     )
+                mode = self.connector_mode.get("type", "polling")
 
-                if fetch_toggles:
+                if mode == "streaming" and fetch_toggles:
+                    self.connector = StreamingConnector(
+                        engine=self.engine,
+                        cache=self.cache,
+                        url=self.unleash_url,
+                        headers=base_headers,
+                        request_timeout=self.unleash_request_timeout,
+                        ready_callback=self._ready_callback,
+                        custom_options=self.unleash_custom_options,
+                    )
+                elif fetch_toggles:
                     start_scheduler = True
                     self.connector = PollingConnector(
                         engine=self.engine,
@@ -422,10 +447,16 @@ def destroy(self) -> None:
             )
 
         try:
-            self.unleash_scheduler.shutdown()
+            if hasattr(self, "unleash_scheduler") and self.unleash_scheduler:
+                self.unleash_scheduler.remove_all_jobs()
+                self.unleash_scheduler.shutdown(wait=True)
+        except Exception as exc:
+            LOGGER.warning("Exception during scheduler teardown: %s", exc)
+
+        try:
+            self.cache.destroy()
         except Exception as exc:
-            LOGGER.warning("Exception during scheduler shutdown: %s", exc)
-        self.cache.destroy()
+            LOGGER.warning("Exception during cache teardown: %s", exc)
 
     @staticmethod
     def _get_fallback_value(

UnleashClient/connectors/__init__.py

@@ -2,10 +2,12 @@
 from .bootstrap_connector import BootstrapConnector
 from .offline_connector import OfflineConnector
 from .polling_connector import PollingConnector
+from .streaming_connector import StreamingConnector
 
 __all__ = [
     "BaseConnector",
     "BootstrapConnector",
     "OfflineConnector",
     "PollingConnector",
+    "StreamingConnector",
 ]

UnleashClient/connectors/streaming_connector.py

@@ -0,0 +1,125 @@
+import threading
+from typing import Callable, Optional
+
+from ld_eventsource import SSEClient
+from ld_eventsource.config import ConnectStrategy, ErrorStrategy, RetryDelayStrategy
+from yggdrasil_engine.engine import UnleashEngine
+
+from UnleashClient.cache import BaseCache
+from UnleashClient.connectors.base_connector import BaseConnector
+from UnleashClient.constants import APPLICATION_HEADERS, FEATURES_URL, STREAMING_URL
+from UnleashClient.utils import LOGGER
+
+
+class StreamingConnector(BaseConnector):
+    def __init__(
+        self,
+        engine: UnleashEngine,
+        cache: BaseCache,
+        url: str,
+        headers: dict,
+        request_timeout: int,
+        ready_callback: Optional[Callable] = None,
+        backoff_initial: float = 2.0,
+        backoff_max: float = 30.0,
+        backoff_multiplier: float = 2.0,
+        backoff_jitter: Optional[float] = 0.5,
+        custom_options: Optional[dict] = None,
+    ) -> None:
+        super().__init__(engine=engine, cache=cache, ready_callback=ready_callback)
+        self._base_url = url.rstrip("/") + STREAMING_URL
+        self._headers = {
+            **headers,
+            **APPLICATION_HEADERS,
+            "Accept": "text/event-stream",
+        }
+        self._timeout = request_timeout
+        self._backoff_initial = backoff_initial
+        self._backoff_max = backoff_max
+        self._backoff_multiplier = backoff_multiplier
+        self._backoff_jitter = backoff_jitter
+        self._stop = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+        self._client: Optional[SSEClient] = None
+        base_options = custom_options or {}
+        if self._timeout is not None and "timeout" not in base_options:
+            base_options = {"timeout": self._timeout, **base_options}
+        self._custom_options = base_options
+
+    def start(self):
+        if self._thread and self._thread.is_alive():
+            return
+        self._stop.clear()
+        self._thread = threading.Thread(
+            target=self._run, name="UnleashStreaming", daemon=True
+        )
+        self._thread.start()
+
+    def stop(self):
+        self._stop.set()
+        try:
+            if self._client:
+                self._client.close()
+        except Exception:
+            pass
+        if self._thread:
+            self._thread.join(timeout=5)
+
+    def _run(self):
+        try:
+            LOGGER.info("Connecting to Unleash streaming endpoint: %s", self._base_url)
+
+            connect_strategy = ConnectStrategy.http(
+                self._base_url,
+                headers=self._headers,
+                urllib3_request_options=self._custom_options,
+            )
+
+            retry_strategy = RetryDelayStrategy.default(
+                max_delay=self._backoff_max,
+                backoff_multiplier=self._backoff_multiplier,
+                jitter_multiplier=self._backoff_jitter,
+            )
+
+            self._client = SSEClient(
+                connect=connect_strategy,
+                initial_retry_delay=self._backoff_initial,
+                retry_delay_strategy=retry_strategy,
+                retry_delay_reset_threshold=60.0,
+                error_strategy=ErrorStrategy.always_continue(),
+                logger=LOGGER,
+            )
+
+            # Initial hydration happens in the stream.
+            for event in self._client.events:
+                if self._stop.is_set():
+                    break
+                if not event.event:
+                    continue
+
+                if event.event in ("unleash-connected", "unleash-updated"):
+                    try:
+                        self.engine.take_state(event.data)
+                        self.cache.set(FEATURES_URL, self.engine.get_state())
+
+                        if event.event == "unleash-connected" and self.ready_callback:
+                            try:
+                                self.ready_callback()
+                            except Exception:
+                                LOGGER.debug("Ready callback failed", exc_info=True)
+                    except Exception:
+                        LOGGER.error("Error applying streaming state", exc_info=True)
+                        self.load_features()
+                else:
+                    LOGGER.debug("Ignoring SSE event type: %s", event.event)
+
+            LOGGER.debug("SSE stream ended")
+        except Exception as exc:
+            LOGGER.warning("Streaming connection failed: %s", exc)
+            self.load_features()
+        finally:
+            try:
+                if self._client is not None:
+                    self._client.close()
+            except Exception:
+                pass

UnleashClient/constants.py

@@ -6,7 +6,7 @@
 REQUEST_TIMEOUT = 30
 REQUEST_RETRIES = 3
 METRIC_LAST_SENT_TIME = "mlst"
-CLIENT_SPEC_VERSION = "5.1.9"
+CLIENT_SPEC_VERSION = "5.2.2"
 
 # =Unleash=
 APPLICATION_HEADERS = {
@@ -19,6 +19,7 @@
 REGISTER_URL = "/client/register"
 FEATURES_URL = "/client/features"
 METRICS_URL = "/client/metrics"
+STREAMING_URL = "/client/streaming"
 
 # Cache keys
 FAILED_STRATEGIES = "failed_strategies"

pyproject.toml

@@ -31,13 +31,14 @@ dependencies=[
     "importlib_metadata",
     "python-dateutil",
     "semver < 4.0.0",
-    "yggdrasil-engine",
+    "yggdrasil-engine >= 1.0.0",
+    "launchdarkly-eventsource",
 ]
 
 [project.urls]
 Homepage = "https://github.com/Unleash/unleash-python-sdk"
 Documentation = "https://docs.getunleash.io/unleash-python-sdk"
-Changelog = "https://github.com/Unleash/unleash-python-sdk/blob/main/CHANGELOG.md"
+Changelog = "https://github.com/unleash/unleash-python-sdk/blob/main/CHANGELOG.md"
 Repository = "https://github.com/Unleash/unleash-python-sdk"
 Issues = "https://github.com/Unleash/unleash-python-sdk/issues"
 

requirements.txt

@@ -6,7 +6,8 @@ mmhash3
 python-dateutil
 requests
 semver
-yggdrasil-engine
+yggdrasil-engine>=1.0.0
+launchdarkly-eventsource
 
 # Development packages
 # - Testing