feat: Add async FDv1 polling data source and feature requester

ldclient/impl/aio/concurrency.py

@@ -223,6 +223,16 @@ def stop(self):
         if task is not None and task is not asyncio.current_task():
             task.cancel()
 
+    async def wait_stopped(self):
+        """Waits for the task to finish unwinding after ``stop()``. A no-op if
+        the task never started or is the current task."""
+        task = self.__task
+        if task is not None and task is not asyncio.current_task():
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+
     async def _run(self):
         try:
             if self.__initial_delay > 0:

ldclient/impl/datasource/async_feature_requester.py

@@ -0,0 +1,56 @@
+"""
+Default implementation of feature flag polling requests.
+"""
+
+import json
+from collections import namedtuple
+from typing import Optional
+from urllib import parse
+
+from ldclient.impl.aio.transport import AsyncHTTPTransport
+from ldclient.impl.datasource.datasource_common import FDV1_POLLING_ENDPOINT
+from ldclient.impl.util import _headers, log, throw_if_unsuccessful_response
+from ldclient.interfaces import FeatureRequester
+from ldclient.versioned_data_kind import FEATURES, SEGMENTS
+
+CacheEntry = namedtuple('CacheEntry', ['data', 'etag'])
+
+
+class AsyncFeatureRequesterImpl(FeatureRequester):
+    def __init__(self, config, transport: Optional[AsyncHTTPTransport] = None):
+        self._cache: dict = dict()
+        # Only close the transport on shutdown if we created it; an injected
+        # transport is owned by the caller.
+        self._owns_transport = transport is None
+        self._transport = transport if transport is not None else AsyncHTTPTransport(config)
+        self._config = config
+        self._poll_uri = config.base_uri + FDV1_POLLING_ENDPOINT
+        if config.payload_filter_key is not None:
+            self._poll_uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
+
+    async def get_all_data(self):
+        uri = self._poll_uri
+        hdrs = _headers(self._config)
+        cache_entry = self._cache.get(uri)
+        hdrs['Accept-Encoding'] = 'gzip'
+        if cache_entry is not None:
+            hdrs['If-None-Match'] = cache_entry.etag
+        r = await self._transport.request('GET', uri, headers=hdrs)
+        throw_if_unsuccessful_response(r)
+        if r.status == 304 and cache_entry is not None:
+            data = cache_entry.data
+            etag = cache_entry.etag
+            from_cache = True
+        else:
+            data = json.loads(r.body)
+            etag = r.headers.get('ETag')
+            from_cache = False
+            if etag is not None:
+                self._cache[uri] = CacheEntry(data=data, etag=etag)
+        log.debug("%s response status:[%d] From cache? [%s] ETag:[%s]", uri, r.status, from_cache, etag)
+
+        return {FEATURES: data['flags'], SEGMENTS: data['segments']}
+
+    async def close(self):
+        if self._owns_transport:
+            await self._transport.close()

ldclient/impl/datasource/async_polling.py

@@ -0,0 +1,91 @@
+"""
+Default implementation of the polling component.
+"""
+
+# currently excluded from documentation - see docs/README.md
+
+import time
+from typing import Optional
+
+from ldclient.async_config import AsyncConfig
+from ldclient.impl.aio.concurrency import AsyncEvent, AsyncRepeatingTask
+from ldclient.impl.datasource.async_feature_requester import (
+    AsyncFeatureRequesterImpl
+)
+from ldclient.impl.datasource.datasource_common import sink_or_store
+from ldclient.impl.util import (
+    UnsuccessfulResponseException,
+    http_error_message,
+    is_http_error_recoverable,
+    log
+)
+from ldclient.interfaces import (
+    AsyncFeatureStore,
+    AsyncUpdateProcessor,
+    DataSourceErrorInfo,
+    DataSourceErrorKind,
+    DataSourceState
+)
+
+
+class AsyncPollingUpdateProcessor(AsyncUpdateProcessor):
+    def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequesterImpl, store: AsyncFeatureStore, ready: AsyncEvent):
+        self._config = config
+        self._data_source_update_sink = config.data_source_update_sink
+        self._requester = requester
+        self._store = store
+        self._ready = ready
+        self._task = AsyncRepeatingTask("ldclient.datasource.polling", config.poll_interval, 0, self._fetch_and_store)
+
+    def start(self):
+        log.info("Starting AsyncPollingUpdateProcessor with request interval: " + str(self._config.poll_interval))
+        self._task.start()
+
+    def initialized(self):
+        return self._ready.is_set() is True and self._store.initialized is True
+
+    async def stop(self):
+        self.__stop_with_error_info(None)
+        # Wait for the current poll to finish before closing the transport. This way
+        # stop() does not return until background work has stopped, so we never close
+        # the transport while a request is still using it.
+        await self._task.wait_stopped()
+        await self._requester.close()
+
+    def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]):
+        log.info("Stopping AsyncPollingUpdateProcessor")
+        self._task.stop()
+
+        if self._data_source_update_sink is None:
+            return
+
+        self._data_source_update_sink.update_status(DataSourceState.OFF, error)
+
+    async def _fetch_and_store(self):
+        try:
+            all_data = await self._requester.get_all_data()
+            await sink_or_store(self._data_source_update_sink, self._store).init(all_data)
+            if not self._ready.is_set() and self._store.initialized:
+                log.info("AsyncPollingUpdateProcessor initialized ok")
+                self._ready.set()
+
+            # Signal VALID once the store is populated.
+            if self._store.initialized and self._data_source_update_sink is not None:
+                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+        except UnsuccessfulResponseException as e:
+            error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, e.status, time.time(), str(e))
+
+            http_error_message_result = http_error_message(e.status, "polling request")
+            if not is_http_error_recoverable(e.status):
+                log.error(http_error_message_result)
+                self._ready.set()  # if client is initializing, make it stop waiting; has no effect if already inited
+                self.__stop_with_error_info(error_info)
+            else:
+                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)
+        except Exception as e:
+            log.exception('Error: Exception encountered when updating flags. %s' % e)
+            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(e)))