+
+
+class WaitableAutomaticStateTracker(AutomaticStateTracker):
+ """This is an AutomaticStateTracker that exposes a wait method which
+ will block the calling thread until the state changes with an
+ optional timeout. The caller should check the return value of
+ wait; it will be true if something changed and false if the wait
+ simply timed out. If the return value is true, the instance
+ should be reset() before wait is called again.
+
+ Example usage::
+
+ detector = waitable_presence.WaitableAutomaticStateSubclass()
+ while True:
+ changed = detector.wait(timeout=60 * 5)
+ if changed:
+ detector.reset()
+ # Figure out what changed and react
+ else:
+ # Just a timeout; no need to reset. Maybe do something
+ # else before looping up into wait again.
+ """
+
+ def __init__(
+ self,
+ update_ids_to_update_secs: Dict[str, float],
+ *,
+ override_sleep_delay: Optional[float] = None,
+ ) -> None:
+ """Construct an WaitableAutomaticStateTracker.
+
+ Args:
+ update_ids_to_update_secs: a dict mapping a user-defined
+ update_id into a period (number of seconds) with which
+ we would like this update performed. e.g.::
+
+ update_ids_to_update_secs = {
+ 'refresh_local_state': 10.0,
+ 'refresh_remote_state': 60.0,
+ }
+
+ This would indicate that every 10s we would like to
+ refresh local state whereas every 60s we'd like to
+ refresh remote state.
+
+ override_sleep_delay: By default, this class determines
+ how long the background thread should sleep between
+ automatic invocations to :meth:`heartbeat` based on the
+ period of each update type in update_ids_to_update_secs.
+ If this argument is non-None, it overrides this computation
+ and uses this period as the sleep in the background thread.
+ """
+ self._something_changed = threading.Event()
+ super().__init__(update_ids_to_update_secs, override_sleep_delay=override_sleep_delay)
+
+ def something_changed(self):
+ """Indicate that something has changed."""
+ self._something_changed.set()
+
+ def did_something_change(self) -> bool:
+ """Indicate whether some state has changed in the background."""
+ return self._something_changed.is_set()
+
+ def reset(self):
+ """Call to clear the 'something changed' bit. See usage above."""
+ self._something_changed.clear()
+
+ def wait(self, *, timeout=None):
+ """Wait for something to change or a timeout to lapse.
+
+ Args:
+ timeout: maximum amount of time to wait. If None, wait
+ forever (until something changes).
+ """
+ return self._something_changed.wait(timeout=timeout)