FFM-6549 Add wait_for_initialization functionality (#55)

* FFM-6549 Add threading.Event as CfClient instance variable and pass to PollingProcessor

* FFM-6549 Set thread when flags and groups are cached

* FFM-6549 Move debug log to after client run has loaded and add an example

* FFM-6549 Linter

* FFM-6549 Edit comment

* FFM-6549 1.1.7 release prep

* FFM-6549 Debug log

* FFM-6549 Move initialized complete log to poll processor

* FFM-6549 Refactor initial polling

* FFM-6549 Refactor initial polling

Author: erdirowlands

examples/wait_for_initialization_example/wait_for_initialization.py

@@ -0,0 +1,28 @@
+import time
+
+from featureflags.config import with_base_url, with_events_url
+from featureflags.evaluations.auth_target import Target
+from featureflags.client import CfClient
+from featureflags.util import log
+
+
+def main():
+    log.debug("Starting example")
+    api_key = "Your API Key"
+    client = CfClient(api_key)
+    # Don't continue until all flags and groups have been loaded into the cache.
+    log.debug("Waiting to load all flags and groups before proceeding")
+    client.wait_for_initialization()
+
+    #  If required you can check the initialized status of the Client at anytime
+    log.debug("Client is_initialized status is %s", client.is_initialized())
+    target = Target(identifier='harness')
+
+    while True:
+        result = client.bool_variation('identifier_of_your_bool_flag', target, False)
+        log.debug("Result %s", result)
+        time.sleep(10)
+
+
+if __name__ == "__main__":
+    main()

featureflags/client.py

@@ -28,6 +28,9 @@ def __init__(
             config: Optional[Config] = None
     ):
         self._client: Optional[Client] = None
+        #  The Client is considered initialized when flags and groups
+        #  are loaded into cache.
+        self._initialized = threading.Event()
         self._auth_token: Optional[str] = None
         self._environment_id: Optional[str] = None
         self._sdk_key: Optional[str] = sdk_key
@@ -47,9 +50,9 @@ def __init__(
         self._repository = Repository(self._config.cache)
         self._evaluator = Evaluator(self._repository)
 
-        log.debug("CfClient initialized")
         self.run()
 
+
     def run(self):
         self.authenticate()
 
@@ -60,6 +63,10 @@ def run(self):
             client=self._client,
             config=self._config,
             environment_id=self._environment_id,
+            #  PollingProcessor is responsible for doing the initial
+            #  flag/group fetch and cache. So we allocate it the responsibility
+            #  for setting the Client is_initialized variable.
+            wait_for_initialization=self._initialized,
             ready=polling_event,
             stream_ready=streaming_event,
             repository=self._repository
@@ -87,6 +94,13 @@ def run(self):
                 environment=self._environment_id
             )
 
+    def wait_for_initialization(self):
+        log.debug("Waiting for initialization to finish")
+        self._initialized.wait()
+
+    def is_initialized(self):
+        return self._initialized.is_set()
+
     def get_environment_id(self):
         return self._environment_id
 

featureflags/polling.py

@@ -13,15 +13,16 @@
 class PollingProcessor(Thread):
 
     def __init__(self, client: AuthenticatedClient, config: Config,
-                 environment_id: str, ready: Event,
-                 stream_ready: Event,
+                 environment_id: str, wait_for_initialization: Event,
+                 ready: Event, stream_ready: Event,
                  repository: DataProviderInterface) -> None:
         Thread.__init__(self)
         self.daemon = True
         self.__environment_id = environment_id
         self.__client = client
         self.__config = config
         self.__running = False
+        self.__wait_for_initialization = wait_for_initialization
         self.__ready = ready
         self.__stream_ready = stream_ready
         self.__repository = repository
@@ -34,19 +35,32 @@ def run(self):
                             str(self.__config.pull_interval) +
                             " setting to 60")
                 self.__config.pull_interval = 60
+
+            self.__running = True
+            #  Get initial flags and groups
+            try:
+                log.info("Fetching initial target segments and flags")
+                self.retrieve_flags_and_segments()
+                log.info("Initial target segments and flags fetched. "
+                         "PollingProcessor will start in: " +
+                         str(self.__config.pull_interval) + " seconds")
+                #  Segments and flags have been cached so
+                #  mark the Client as initialised.
+                self.__wait_for_initialization.set()
+                log.debug("CfClient initialized")
+            except Exception as ex:
+                log.exception(
+                    'Error: Exception encountered when '
+                    'getting initial flags and segments. %s',
+                    ex
+                )
+            #  Sleep for an interval before going into the polling loop.
+            time.sleep(self.__config.pull_interval)
             log.info("Starting PollingProcessor with request interval: " +
                      str(self.__config.pull_interval))
-            self.__running = True
             while self.__running:
                 start_time = time.time()
                 try:
-                    t1 = Thread(target=self.__retrieve_segments)
-                    t2 = Thread(target=self.__retrieve_flags)
-                    t1.start()
-                    t2.start()
-                    t1.join()
-                    t2.join()
-
                     if self.__config.enable_stream and \
                             self.__stream_ready.is_set():
                         log.debug('Poller will be paused because' +
@@ -55,6 +69,7 @@ def run(self):
                         self.__ready.wait()
                         log.debug('Poller resuming ')
                     else:
+                        self.retrieve_flags_and_segments()
                         self.__ready.set()
                 except Exception as e:
                     log.exception(
@@ -73,6 +88,14 @@ def stop(self):
         log.info("Stopping PollingProcessor")
         self.__running = False
 
+    def retrieve_flags_and_segments(self):
+        t1 = Thread(target=self.__retrieve_segments)
+        t2 = Thread(target=self.__retrieve_flags)
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+
     def __retrieve_flags(self):
         log.debug("Loading feature flags")
         flags = retrieve_flags(