Streaming support

UnleashClient/__init__.py

@@ -16,6 +16,7 @@
 
 from UnleashClient.api import register_client
 from UnleashClient.constants import (
+    APPLICATION_HEADERS,
     DISABLED_VARIATION,
     ETAG,
     METRIC_LAST_SENT_TIME,
@@ -35,6 +36,7 @@
     aggregate_and_send_metrics,
     fetch_and_load_features,
 )
+from UnleashClient.streaming import StreamingConnector, StreamingEventProcessor
 
 from .cache import BaseCache, FileCache
 from .utils import LOGGER, InstanceAllowType, InstanceCounter
@@ -111,7 +113,7 @@ class UnleashClient:
     :param event_callback: Function to call if impression events are enabled.  WARNING: Depending on your event library, this may have performance implications!
     """
 
-    def __init__(
+    def __init__(  # noqa: PLR0915
         self,
         url: str,
         app_name: str,
@@ -136,6 +138,8 @@ def __init__(
         scheduler_executor: Optional[str] = None,
         multiple_instance_mode: InstanceAllowType = InstanceAllowType.WARN,
         event_callback: Optional[Callable[[BaseEvent], None]] = None,
+        experimental_mode: Optional[dict] = None,
+        sse_client_factory: Optional[Callable] = None,
     ) -> None:
         custom_headers = custom_headers or {}
         custom_options = custom_options or {}
@@ -169,6 +173,9 @@ 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.experimental_mode = experimental_mode
+        self._streaming_connector: Optional[StreamingConnector] = None
+        self._sse_client_factory = sse_client_factory
 
         self._do_instance_check(multiple_instance_mode)
 
@@ -267,8 +274,10 @@ def initialize_client(self, fetch_toggles: bool = True) -> None:
             try:
                 base_headers = {
                     **self.unleash_custom_headers,
+                    **APPLICATION_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}",
                 }
 
@@ -277,7 +286,6 @@ def initialize_client(self, fetch_toggles: bool = True) -> None:
                     "unleash-interval": self.unleash_metrics_interval_str_millis,
                 }
 
-                # Setup
                 metrics_args = {
                     "url": self.unleash_url,
                     "app_name": self.unleash_app_name,
@@ -289,6 +297,12 @@ def initialize_client(self, fetch_toggles: bool = True) -> None:
                     "engine": self.engine,
                 }
 
+                base_job_args = {
+                    "cache": self.cache,
+                    "engine": self.engine,
+                    "ready_callback": self._ready_callback,
+                }
+
                 # Register app
                 if not self.unleash_disable_registration:
                     register_client(
@@ -303,47 +317,68 @@ def initialize_client(self, fetch_toggles: bool = True) -> None:
                         self.unleash_request_timeout,
                     )
 
-                if fetch_toggles:
-                    fetch_headers = {
-                        **base_headers,
-                        "unleash-interval": self.unleash_refresh_interval_str_millis,
-                    }
-
-                    job_args = {
-                        "url": self.unleash_url,
-                        "app_name": self.unleash_app_name,
-                        "instance_id": self.unleash_instance_id,
-                        "headers": fetch_headers,
-                        "custom_options": self.unleash_custom_options,
-                        "cache": self.cache,
-                        "engine": self.engine,
-                        "request_timeout": self.unleash_request_timeout,
-                        "request_retries": self.unleash_request_retries,
-                        "project": self.unleash_project_name,
-                        "event_callback": self.unleash_event_callback,
-                        "ready_callback": self._ready_callback,
-                    }
-                    job_func: Callable = fetch_and_load_features
-                else:
-                    job_args = {
-                        "cache": self.cache,
-                        "engine": self.engine,
-                        "ready_callback": self._ready_callback,
-                    }
-                    job_func = load_features
-
-                job_func(**job_args)  # type: ignore
                 # Start periodic jobs
                 self.unleash_scheduler.start()
-                self.fl_job = self.unleash_scheduler.add_job(
-                    job_func,
-                    trigger=IntervalTrigger(
-                        seconds=int(self.unleash_refresh_interval),
-                        jitter=self.unleash_refresh_jitter,
-                    ),
-                    executor=self.unleash_executor_name,
-                    kwargs=job_args,
+
+                # Decide upstream connection mode
+                mode = (
+                    (self.experimental_mode or {}).get("type")
+                    if self.experimental_mode
+                    else None
                 )
+
+                if mode != "streaming":
+                    if fetch_toggles:
+                        # MODE: polling
+
+                        job_args = {
+                            **base_job_args,
+                            "url": self.unleash_url,
+                            "app_name": self.unleash_app_name,
+                            "instance_id": self.unleash_instance_id,
+                            "headers": {
+                                **base_headers,
+                                "unleash-interval": self.unleash_refresh_interval_str_millis,
+                            },
+                            "custom_options": self.unleash_custom_options,
+                            "request_timeout": self.unleash_request_timeout,
+                            "request_retries": self.unleash_request_retries,
+                            "event_callback": self.unleash_event_callback,
+                            "project": self.unleash_project_name,
+                        }
+
+                        job_func: Callable
+                        job_func = fetch_and_load_features
+                    else:
+                        # MODE: offline
+
+                        job_args = base_job_args
+                        job_func = load_features
+
+                    job_func(**job_args)  # type: ignore
+                    self.fl_job = self.unleash_scheduler.add_job(
+                        job_func,
+                        trigger=IntervalTrigger(
+                            seconds=int(self.unleash_refresh_interval),
+                            jitter=self.unleash_refresh_jitter,
+                        ),
+                        executor=self.unleash_executor_name,
+                        kwargs=job_args,
+                    )
+                else:
+                    # MODE: streaming
+                    processor = StreamingEventProcessor(self.engine)
+                    self._streaming_connector = StreamingConnector(
+                        url=self.unleash_url,
+                        headers=base_headers,
+                        request_timeout=self.unleash_request_timeout,
+                        event_processor=processor,
+                        on_ready=self._ready_callback,
+                        sse_client_factory=self._sse_client_factory,
+                        custom_options=self.unleash_custom_options,
+                    )
+                    self._streaming_connector.start()
+
                 if not self.unleash_disable_metrics:
                     self.metric_job = self.unleash_scheduler.add_job(
                         aggregate_and_send_metrics,
@@ -396,7 +431,11 @@ def destroy(self) -> None:
 
         You shouldn't need this too much!
         """
-        self.fl_job.remove()
+        try:
+            if self.fl_job:
+                self.fl_job.remove()
+        except Exception:  # best-effort
+            pass
         if self.metric_job:
             self.metric_job.remove()
 
@@ -411,7 +450,11 @@ def destroy(self) -> None:
                 request_timeout=self.unleash_request_timeout,
                 engine=self.engine,
             )
-
+        if self._streaming_connector:
+            try:
+                self._streaming_connector.stop()
+            except Exception:
+                pass
         self.unleash_scheduler.shutdown()
         self.cache.destroy()
 
@@ -513,9 +556,9 @@ def get_variant(self, feature_name: str, context: Optional[dict] = None) -> dict
                     event_type=UnleashEventType.VARIANT,
                     event_id=uuid.uuid4(),
                     context=context,
-                    enabled=variant["enabled"],
+                    enabled=bool(variant["enabled"]),
                     feature_name=feature_name,
-                    variant=variant["name"],
+                    variant=str(variant["name"]),
                 )
 
                 self.unleash_event_callback(event)

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"

UnleashClient/streaming/__init__.py

@@ -0,0 +1,5 @@
+# ruff: noqa: F401
+# Streaming package
+
+from .connector import StreamingConnector
+from .event_processor import StreamingEventProcessor

UnleashClient/streaming/connector.py

@@ -0,0 +1,123 @@
+import threading
+from typing import Callable, Optional
+
+from ld_eventsource import SSEClient
+from ld_eventsource.config import ConnectStrategy, ErrorStrategy, RetryDelayStrategy
+
+from UnleashClient.constants import STREAMING_URL
+from UnleashClient.utils import LOGGER
+
+from .event_processor import StreamingEventProcessor
+
+
+class StreamingConnector:
+    """
+    Manages the SSE connection lifecycle with reconnect/backoff and delegates
+    event handling to an injected StreamingEventProcessor.
+    """
+
+    def __init__(
+        self,
+        url: str,
+        headers: dict,
+        request_timeout: int,
+        event_processor: StreamingEventProcessor,
+        on_ready: Optional[Callable[[], None]] = None,
+        sse_client_factory: Optional[Callable[[str, dict, int], SSEClient]] = 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:
+        self._base_url = url.rstrip("/") + STREAMING_URL
+        self._headers = {**headers, "Accept": "text/event-stream"}
+        self._timeout = request_timeout
+        self._on_ready = on_ready
+        self._sse_factory = sse_client_factory
+        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._lock = threading.Lock()
+        self._processor = event_processor
+        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()
+        if self._thread:
+            self._thread.join(timeout=5)
+
+    def _run(self):  # noqa: PLR0912
+        """
+        Main streaming loop. Creates SSEClient once and lets it handle retries internally.
+        Only recreates client if there's a catastrophic failure.
+        """
+        client: Optional[SSEClient] = None
+
+        try:
+            LOGGER.info("Connecting to Unleash streaming endpoint: %s", self._base_url)
+
+            if self._sse_factory:
+                client = self._sse_factory(self._base_url, self._headers, self._timeout)
+            else:
+                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,
+                )
+
+                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,
+                )
+
+            for event in client.events:
+                if self._stop.is_set():
+                    break
+                if not event.event:
+                    continue
+
+                self._processor.process(event)
+                if event.event == "unleash-connected" and self._processor.hydrated:
+                    if self._on_ready:
+                        try:
+                            self._on_ready()
+                        except Exception as cb_exc:  # noqa: BLE001
+                            LOGGER.debug("Ready callback error: %s", cb_exc)
+
+            LOGGER.debug("SSE stream ended")
+
+        except Exception as exc:  # noqa: BLE001
+            LOGGER.warning("Streaming connection failed: %s", exc)
+
+        finally:
+            try:
+                if client is not None:
+                    client.close()
+            except Exception:  # noqa: BLE001
+                pass