feat: Support to get the initial state of the Client

Author: dengfeige

demo.py

@@ -10,11 +10,17 @@
 
     config = fp.Config(remote_uri=FEATURE_PROBE_SERVER_URL,  # FeatureProbe server URL
                        sync_mode='pooling',
-                       refresh_interval=3)
+                       refresh_interval=2,
+                       start_wait=5)
 
     # Server Side SDK Key for your project and environment
     SDK_KEY = 'server-8ed48815ef044428826787e9a238b9c6a479f98c'
     with fp.Client(SDK_KEY, config) as client:
+        if client.initialized():
+            print("SDK successfully initialized!")
+        else:
+            print("SDK failed to initialize!")
+            exit()
         # Create one user
         # "userId" is used in rules, should be filled in.
         user = fp.User().with_attr('userId', '00001')

featureprobe/client.py

@@ -14,6 +14,7 @@
 
 import logging
 import time
+from threading import Event
 from typing import Any
 
 from featureprobe.config import Config
@@ -44,9 +45,21 @@ def __init__(self, server_sdk_key: str, config: Config = Config()):
         context = Context(server_sdk_key, config)
         self._event_processor = config.event_processor_creator(context)
         self._data_repo = config.data_repository_creator(context)
+
+        synchronize_process_ready = Event()
         self._synchronizer = config.synchronizer_creator(
-            context, self._data_repo)
+            context, self._data_repo, synchronize_process_ready)
         self._synchronizer.sync()
+        if config.start_wait > 0:
+            Client.__logger.info("Waiting up to " +
+                                 str(config.start_wait) +
+                                 " seconds for FeatureProbe client to initialize...")
+            synchronize_process_ready.wait(config.start_wait)
+        if self._synchronizer.initialized() is True:
+            Client.__logger.info("Started FeatureProbe Client: Successfully")
+        else:
+            Client.__logger.warning(
+                "Initialization timeout exceeded for FeatureProbe Client or an error occurred")
 
     def __enter__(self):
         return self
@@ -63,6 +76,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         """
         self.close()
 
+    def initialized(self):
+        """Tests whether the FeatureProbe client is ready to be used"""
+        return self._synchronizer.initialized()
+
     def flush(self):
         """Manually push events"""
         self._event_processor.flush()

featureprobe/config.py

@@ -50,7 +50,8 @@ def __init__(self,
                  event_url: str = None,
                  remote_uri: str = 'http://127.0.0.1:4007',
                  http_config: HttpConfig = HttpConfig(),
-                 refresh_interval: Union[timedelta, float] = timedelta(seconds=5),
+                 refresh_interval: Union[timedelta, float] = timedelta(seconds=2),
+                 start_wait: float = 5,
                  ):
         self._location = location
         self._synchronizer_creator = SyncMode(sync_mode).synchronizer_creator
@@ -59,6 +60,7 @@ def __init__(self,
         self._synchronizer_url = synchronizer_url
         self._event_url = event_url
         self._remote_uri = remote_uri
+        self._start_wait = start_wait
         self._http_config = http_config or HttpConfig()
         self._refresh_interval = refresh_interval \
             if isinstance(refresh_interval, timedelta) \
@@ -99,3 +101,7 @@ def http_config(self):
     @property
     def refresh_interval(self):
         return self._refresh_interval
+
+    @property
+    def start_wait(self):
+        return self._start_wait

featureprobe/file_synchronizer.py

@@ -14,6 +14,7 @@
 
 import json
 import logging
+from threading import Event
 from typing import TYPE_CHECKING
 
 from featureprobe.model.repository import Repository
@@ -29,27 +30,35 @@ class FileSynchronizer(Synchronizer):
 
     def __init__(self,
                  data_repository: "DataRepository",
-                 location: str):
+                 location: str,
+                 ready: "Event"):
         self._data_repository = data_repository
         self._location = location
+        self._ready = ready
 
     @classmethod
     def from_context(
             cls,
             context: "Context",
-            data_repo: "DataRepository") -> "Synchronizer":
-        return cls(data_repo, context.location)
+            data_repo: "DataRepository",
+            ready: "Event") -> "Synchronizer":
+        return cls(data_repo, context.location, ready)
 
     def sync(self):
         try:
             with open(self._location, 'r', encoding='utf-8') as f:
                 repo = Repository.from_json(json.load(f))
                 self._data_repository.refresh(repo)
+            self._ready.set()
         except FileNotFoundError:
             # sourcery skip: replace-interpolation-with-fstring
             self._logger.error(
                 'repository file resource not found in path: %s' %
                 self._location)
 
+    def initialized(self):
+        return self._ready.is_set()
+
     def close(self):
+        self._ready.clear()
         return

featureprobe/pooling_synchronizer.py

@@ -32,7 +32,11 @@
 class PoolingSynchronizer(Synchronizer):
     __logger = logging.getLogger('FeatureProbe-Synchronizer')
 
-    def __init__(self, context: "Context", data_repo: "DataRepository"):
+    def __init__(
+            self,
+            context: "Context",
+            data_repo: "DataRepository",
+            ready: "threading.Event"):
         self._refresh_interval = context.refresh_interval
         self._api_url = context.synchronizer_url
         self._data_repo = data_repo
@@ -47,13 +51,15 @@ def __init__(self, context: "Context", data_repo: "DataRepository"):
 
         self._scheduler = None
         self._lock = threading.RLock()
+        self._ready = ready
 
     @classmethod
     def from_context(
             cls,
             context: "Context",
-            data_repo: "DataRepository") -> "Synchronizer":
-        return cls(context, data_repo)
+            data_repo: "DataRepository",
+            ready: "threading.Event") -> "Synchronizer":
+        return cls(context, data_repo, ready)
 
     def sync(self):
         PoolingSynchronizer.__logger.info(
@@ -77,6 +83,7 @@ def close(self):
         with self._lock:
             self._scheduler.shutdown()
             del self._scheduler
+            self._ready.clear()
 
     def _poll(self):
         try:
@@ -88,7 +95,13 @@ def _poll(self):
             self.__logger.debug('Http response body: %s' % body)
             repo = Repository.from_json(body)
             self._data_repo.refresh(repo)
+
+            if not self._ready.is_set():
+                self._ready.set()
         except Exception as e:  # noqa
             self.__logger.error(
                 'Unexpected error from polling processor',
                 exc_info=e)
+
+    def initialized(self):
+        return self._ready.is_set()

featureprobe/synchronizer.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from abc import ABC, abstractmethod
+from threading import Event
 
 from typing import TYPE_CHECKING
 
@@ -28,7 +29,8 @@ class Synchronizer(ABC):
     def from_context(
             cls,
             context: "Context",
-            data_repo: "DataRepository") -> "Synchronizer":
+            data_repo: "DataRepository",
+            ready: "Event") -> "Synchronizer":
         pass
 
     @abstractmethod
@@ -39,5 +41,9 @@ def sync(self):
     def close(self):
         pass
 
+    @abstractmethod
+    def initialized(self):
+        pass
+
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()

tests/polling_synchronizer_test.py

@@ -11,3 +11,46 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
+from featureprobe.pooling_synchronizer import PoolingSynchronizer
+from featureprobe.memory_data_repository import MemoryDataRepository
+from featureprobe.config import Config
+from featureprobe.context import Context
+from threading import Event
+from requests import Session
+from unittest.mock import patch
+
+
+def test_init_synchronizer_failed():
+    realy = Event()
+    context = Context("test-sdk-key", Config())
+    synchroizer = PoolingSynchronizer.from_context(
+        context, MemoryDataRepository.from_context(context), realy)
+    realy.wait(2)
+
+    assert not synchroizer.initialized()
+
+
+@patch.object(Session, 'get')
+def test_init_synchronizer_wait_for_init_success(session_get):
+    realy = Event()
+    context = Context("test-sdk-key", Config())
+    session_get.return_value = MockHttpReponse(200, '{}')
+    synchroizer = PoolingSynchronizer.from_context(
+        context, MemoryDataRepository.from_context(context), realy)
+    synchroizer.sync()
+    realy.wait(5)
+
+    assert synchroizer.initialized()
+
+
+class MockHttpReponse:
+    def __init__(self, status_code, json_response):
+        self.status_code = status_code
+        self.response = json_response
+
+    def raise_for_status(self):
+        pass
+
+    def json(self):
+        return self.response