refactor: Require the data source update sink in FDv1 data sources

ldclient/client.py

@@ -22,9 +22,6 @@
     get_plugin_hooks
 )
 from ldclient.impl.client_common import secure_mode_hash as _secure_mode_hash
-from ldclient.impl.datasource.feature_requester import FeatureRequesterImpl
-from ldclient.impl.datasource.polling import PollingUpdateProcessor
-from ldclient.impl.datasource.streaming import StreamingUpdateProcessor
 from ldclient.impl.datasystem import DataAvailability, DataSystem
 from ldclient.impl.datasystem.fdv2 import FDv2
 from ldclient.impl.evaluator import Evaluator, error_reason
@@ -37,7 +34,7 @@
 from ldclient.impl.flag_tracker import FlagTrackerImpl
 from ldclient.impl.model.feature_flag import FeatureFlag
 from ldclient.impl.rwlock import ReadWriteLock
-from ldclient.impl.stubs import NullEventProcessor, NullUpdateProcessor
+from ldclient.impl.stubs import NullEventProcessor
 from ldclient.impl.util import check_uwsgi, log
 from ldclient.interfaces import (
     BigSegmentStoreStatusProvider,
@@ -245,27 +242,6 @@ def _set_event_processor(self, config):
         self._event_processor = config.event_processor_class(config)
         return None
 
-    def _make_update_processor(self, config, store, ready, diagnostic_accumulator):
-        if config.update_processor_class:
-            log.info("Using user-specified update processor: " + str(config.update_processor_class))
-            return config.update_processor_class(config, store, ready)
-
-        if config.offline or config.use_ldd:
-            return NullUpdateProcessor(config, store, ready)
-
-        if config.stream:
-            return StreamingUpdateProcessor(config, store, ready, diagnostic_accumulator)
-
-        log.info("Disabling streaming API")
-        log.warning("You should only disable the streaming API if instructed to do so by LaunchDarkly support")
-
-        if config.feature_requester_class:
-            feature_requester = config.feature_requester_class(config)
-        else:
-            feature_requester = FeatureRequesterImpl(config)  # type: FeatureRequester
-
-        return PollingUpdateProcessor(config, feature_requester, store, ready)
-
     def get_sdk_key(self) -> Optional[str]:
         """Returns the configured SDK key."""
         return self._config.sdk_key

ldclient/impl/datasource/async_polling.py

@@ -9,7 +9,7 @@
 
 from ldclient.async_config import AsyncConfig
 from ldclient.impl.aio.concurrency import AsyncEvent, AsyncRepeatingTask
-from ldclient.impl.datasource.datasource_common import async_sink_or_store
+from ldclient.impl.datasource.datasource_common import require_update_sink
 from ldclient.impl.retry import (
     FailureKind,
     RetryState,
@@ -22,6 +22,7 @@
     log
 )
 from ldclient.interfaces import (
+    AsyncDataSourceUpdateSink,
     AsyncFeatureRequester,
     AsyncFeatureStore,
     AsyncUpdateProcessor,
@@ -41,7 +42,7 @@ class AsyncPollingUpdateProcessor(AsyncUpdateProcessor):
 
     def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store: AsyncFeatureStore, ready: AsyncEvent, retry_state: Optional[RetryState] = None):
         self._config = config
-        self._data_source_update_sink = config.data_source_update_sink
+        self._data_source_update_sink: AsyncDataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._requester = requester
         self._store = store
         self._ready = ready
@@ -60,8 +61,7 @@ async def stop(self):
         log.info("Stopping AsyncPollingUpdateProcessor")
         self._task.stop()
 
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
+        self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
         # Do not close the transport while an in-flight request still uses it.
         try:
@@ -73,10 +73,8 @@ async def _fetch_and_store(self) -> None:
         """Makes one poll request and records the outcome on the retry state."""
         try:
             all_data = await self._requester.get_all_data()
-            await async_sink_or_store(self._data_source_update_sink, self._store).init(all_data)
-
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+            await self._data_source_update_sink.init(all_data)
+            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
 
             # Report the status before signaling readiness, so a caller that
             # wakes on readiness cannot still read INITIALIZING.
@@ -105,5 +103,4 @@ async def _fetch_and_store(self) -> None:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace)
 
-        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)

ldclient/impl/datasource/async_streaming.py

@@ -18,8 +18,8 @@
 from ldclient.impl.datasource.datasource_common import (
     STREAM_ALL_PATH,
     StreamClosedError,
-    async_sink_or_store,
-    parse_path
+    parse_path,
+    require_update_sink
 )
 from ldclient.impl.retry import (
     FailureKind,
@@ -29,6 +29,7 @@
 )
 from ldclient.impl.util import http_error_description, log, monotonic_seconds
 from ldclient.interfaces import (
+    AsyncDataSourceUpdateSink,
     AsyncUpdateProcessor,
     DataSourceErrorInfo,
     DataSourceErrorKind,
@@ -49,7 +50,7 @@ def __init__(self, config, store, ready, diagnostic_accumulator, sse_factory: Op
         if config.payload_filter_key is not None:
             self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
         self._config = config
-        self._data_source_update_sink = config.data_source_update_sink
+        self._data_source_update_sink: AsyncDataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._store = store
         self._running = False
         self._ready = ready
@@ -123,8 +124,7 @@ async def _run(self):
                         self._record_stream_init(False)
                         self._connection_attempt_started_monotonic = 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("AsyncStreamingUpdateProcessor initialized ok.")
@@ -172,8 +172,7 @@ async def stop(self):
         # OFF means an explicit shutdown. No stream failure produces it. It is
         # reported before teardown, so a slow close cannot hold back the status
         # that tells a waiter to give up. The sink drops anything after OFF.
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
+        self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
         # Cancel the run task before tearing down the rest, preventing _run from
         # starting a fresh SSE connection against the session we are closing.
@@ -196,7 +195,7 @@ def initialized(self):
     # Returns True if we initialized the feature store
     async def _process_message(self, msg: Event) -> bool:
         """Process a single SSE event.  Returns True on a successful ``put``."""
-        target = async_sink_or_store(self._data_source_update_sink, self._store)
+        target = self._data_source_update_sink
         if msg.event == 'put':
             all_data = json.loads(msg.data)
             init_data: Mapping[VersionedDataKind, Mapping[str, dict]] = {FEATURES: all_data['data']['flags'], SEGMENTS: all_data['data']['segments']}
@@ -270,8 +269,7 @@ async def _handle_error(self, error: Exception) -> bool:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay))
 
-        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)
 
         if delay > 0:
             await asyncio.sleep(delay)

ldclient/impl/datasource/datasource_common.py

@@ -5,20 +5,16 @@
 # currently excluded from documentation - see docs/README.md
 
 from collections import namedtuple
-from typing import Mapping, Optional, Protocol, Union, runtime_checkable
+from typing import Mapping, Optional, Protocol, TypeVar, runtime_checkable
 
 from ldclient.impl.util import _LD_ENVID_HEADER
-from ldclient.interfaces import (
-    AsyncDataSourceUpdateSink,
-    AsyncFeatureStore,
-    DataSourceUpdateSink,
-    FeatureStore
-)
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
 STREAM_ALL_PATH = '/all'
 FDV1_POLLING_ENDPOINT = '/sdk/latest-all'
 
+T = TypeVar('T')
+
 ParsedPath = namedtuple('ParsedPath', ['kind', 'key'])
 
 
@@ -39,32 +35,18 @@ def __init__(self):
         super().__init__("the server closed the stream connection")
 
 
-def sink_or_store(sink: Optional[DataSourceUpdateSink], store: FeatureStore) -> Union[DataSourceUpdateSink, FeatureStore]:
-    """
-    The original implementation of the data sources relied on the feature store
-    directly, which we are trying to move away from. Customers who might have
-    instantiated one directly for some reason wouldn't know they have to set
-    the config's sink manually, so we have to fall back to the store if the
-    sink isn't present.
-
-    The next major release should be able to simplify this structure and
-    remove the need for fall back to the data store because the update sink
-    should always be present.
+def require_update_sink(sink: Optional[T]) -> T:
     """
-    if sink is None:
-        return store
-
-    return sink
+    Returns the data source update sink, or raises if it is missing.
 
+    The data system puts the sink on the config before it builds a data
+    source. A data source writes all its updates to the sink, so it cannot
+    operate without one.
 
-def async_sink_or_store(sink: Optional[AsyncDataSourceUpdateSink], store: AsyncFeatureStore) -> Union[AsyncDataSourceUpdateSink, AsyncFeatureStore]:
-    """
-    The async counterpart of :func:`sink_or_store`, kept separate so each side
-    names its own sink and store types. One shared generic function accepted a
-    sync sink beside an async store without complaint.
+    :raises ValueError: if the sink is None
     """
     if sink is None:
-        return store
+        raise ValueError("config.data_source_update_sink is not set. The data system must set it before it builds a data source.")
 
     return sink
 

ldclient/impl/datasource/polling.py

@@ -11,7 +11,7 @@
 from ldclient.config import Config
 from ldclient.impl.datasource.datasource_common import (
     record_environment_id,
-    sink_or_store
+    require_update_sink
 )
 from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.retry import (
@@ -52,7 +52,7 @@ class PollingUpdateProcessor(UpdateProcessor):
 
     def __init__(self, config: Config, requester: FeatureRequester, store: FeatureStore, ready: Event, retry_state: Optional[RetryState] = None):
         self._config = config
-        self._data_source_update_sink: Optional[DataSourceUpdateSink] = config.data_source_update_sink
+        self._data_source_update_sink: DataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._requester = requester
         self._store = store
         self._ready = ready
@@ -69,21 +69,15 @@ def initialized(self):
     def stop(self):
         log.info("Stopping PollingUpdateProcessor")
         self._task.stop()
-
-        if self._data_source_update_sink is None:
-            return
-
         self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
     def _poll(self) -> None:
         """Makes one poll request and records the outcome on the retry state."""
         try:
             (all_data, headers) = self._get_all_data_with_headers()
             record_environment_id(self._data_source_update_sink, headers)
-            sink_or_store(self._data_source_update_sink, self._store).init(all_data)
-
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+            self._data_source_update_sink.init(all_data)
+            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
 
             # Report the status before signaling readiness, so a caller that
             # wakes on readiness cannot still read INITIALIZING.
@@ -112,8 +106,7 @@ def _poll(self) -> None:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace)
 
-        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)
 
     def _get_all_data_with_headers(self) -> Tuple[Any, Optional[Mapping[str, str]]]:
         """

ldclient/impl/datasource/streaming.py

@@ -20,7 +20,7 @@
     StreamClosedError,
     parse_path,
     record_environment_id,
-    sink_or_store
+    require_update_sink
 )
 from ldclient.impl.http import HTTPFactory, _http_factory
 from ldclient.impl.retry import (
@@ -34,6 +34,7 @@
     DataSourceErrorInfo,
     DataSourceErrorKind,
     DataSourceState,
+    DataSourceUpdateSink,
     UpdateProcessor
 )
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
@@ -57,7 +58,7 @@ def __init__(self, config, store, ready, diagnostic_accumulator, retry_state: Op
         if config.payload_filter_key is not None:
             self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
         self._config = config
-        self._data_source_update_sink = config.data_source_update_sink
+        self._data_source_update_sink: DataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._store = store
         self._running = False
         self._ready = ready
@@ -91,7 +92,7 @@ def run(self):
                     message_ok = False
                     message_handled = False
                     try:
-                        message_ok = self._process_message(sink_or_store(self._data_source_update_sink, self._store), action)
+                        message_ok = self._process_message(self._data_source_update_sink, action)
                         message_handled = True
                     except json.decoder.JSONDecodeError as e:
                         log.info("Error while handling stream event; will restart stream: %s" % e)
@@ -113,8 +114,7 @@ def run(self):
                         self._record_stream_init(False)
                         self._connection_attempt_started_monotonic = 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.")
@@ -167,8 +167,7 @@ def stop(self):
         # OFF means an explicit shutdown. No stream failure produces it. It is
         # reported before teardown, so a slow close cannot hold back the status
         # that tells a waiter to give up. The sink drops anything after OFF.
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
+        self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
         if self._sse:
             self._sse.close()
@@ -256,8 +255,7 @@ def _handle_error(self, error: Exception) -> bool:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay))
 
-        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)
 
         interrupted = self._stop_event.wait(min(delay, TIMEOUT_MAX))
 

ldclient/impl/datasystem/fdv1.py

@@ -277,7 +277,6 @@ def target_availability(self) -> DataAvailability:
         return DataAvailability.REFRESHED
 
     def _make_update_processor(self, config: Config, store: FeatureStore, ready: Event):
-        # Mirrors LDClient._make_update_processor but scoped for FDv1
         if config.update_processor_class:
             return config.update_processor_class(config, store, ready)