Use threading.Event

Signed-off-by: christian.lutnik <christian.lutnik@dynatrace.com>

Author: chrfwow

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/grpc.py

@@ -1,7 +1,6 @@
 import json
 import logging
 import threading
-import time
 import typing
 
 import grpc
@@ -66,6 +65,7 @@ def __init__(
         self.streamline_deadline_seconds = config.stream_deadline_ms * 0.001
         self.deadline = config.deadline_ms * 0.001
         self.connected = False
+        self.connected_event = threading.Event()
         self.channel = self._generate_channel(config)
         self.stub = evaluation_pb2_grpc.ServiceStub(self.channel)
 
@@ -161,12 +161,10 @@ def connect(self) -> None:
         )
         self.monitor_thread.start()
         ## block until ready or deadline reached
-        timeout = self.deadline + time.monotonic()
-        while not self.connected and time.monotonic() < timeout:
-            time.sleep(0.05)
-        logger.debug("Finished blocking gRPC state initialization")
 
-        if not self.connected:
+        if self.connected_event.wait(timeout=self.deadline) and self.connected:
+            logger.debug("Finished blocking gRPC state initialization")
+        else:
             raise ProviderNotReadyError(
                 "Blocking init finished before data synced. Consider increasing startup deadline to avoid inconsistent evaluations."
             )
@@ -204,6 +202,7 @@ def _state_change_callback(self, new_state: ChannelConnectivity) -> None:
 
             logger.debug("gRPC error timer started")
             self.timer.start()
+            self.connected_event.clear()
             self.connected = False
 
     def emit_error(self) -> None:
@@ -236,6 +235,7 @@ def listen(self) -> None:
                             )
                         )
                         self.connected = True
+                        self.connected_event.set()
                     elif message.type == "configuration_change":
                         data = MessageToDict(message)["data"]
                         self.handle_changed_flags(data)

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py

@@ -1,7 +1,6 @@
 import json
 import logging
 import threading
-import time
 import typing
 
 import grpc
@@ -50,6 +49,7 @@ def __init__(
         self.emit_provider_stale = emit_provider_stale
 
         self.connected = False
+        self.connected_event = threading.Event()
         self.thread: typing.Optional[threading.Thread] = None
         self.timer: typing.Optional[threading.Timer] = None
 
@@ -142,15 +142,11 @@ def connect(self) -> None:
         )
         self.monitor_thread.start()
         ## block until ready or deadline reached
-        timeout = self.deadline + time.monotonic()
-        while not self.connected and time.monotonic() < timeout:
-            time.sleep(0.05)
-        logger.debug("Finished blocking gRPC state initialization")
-
-        if not self.connected:
-            raise ProviderNotReadyError(
-                "Blocking init finished before data synced. Consider increasing startup deadline to avoid inconsistent evaluations."
-            )
+        if self.connected_event.wait(timeout=self.deadline) and self.connected:
+            logger.debug("Finished blocking gRPC state initialization")
+        raise ProviderNotReadyError(
+            "Blocking init finished before data synced. Consider increasing startup deadline to avoid inconsistent evaluations."
+        )
 
     def monitor(self) -> None:
         self.channel.subscribe(self._state_change_callback, try_to_connect=True)
@@ -186,6 +182,7 @@ def _state_change_callback(self, new_state: grpc.ChannelConnectivity) -> None:
             logger.debug("gRPC error timer started")
             self.timer.start()
             self.connected = False
+            self.connected_event.clear()
 
     def emit_error(self) -> None:
         logger.debug("gRPC error emitted")
@@ -275,6 +272,7 @@ def listen(self) -> None:
                             context_values,
                         )
                         self.connected = True
+                        self.connected_event.set()
 
                     if not self.active:
                         logger.debug("Terminating gRPC sync thread")