Synchro

def event(self, event_id: str, event_data: Any) -> None:
    """Add an entry to the table of triggered event

    Args:
        event_id: unique ID of the event
        event_data: arbitrary data to transmit to the event receiver
    """

    with self._triggered_events_lock:
        # Remove obsolete triggered events. We could additionally call at other times.
        self._clean_triggered_events()

        # In case a event already exists for this ID, overwrite it with the newer one
        self._triggered_events[event_id] = {
            'start_time': time.time(),
            'data': event_data,
        }

    # check if added event completes some listeners
    completed_listeners = self._all_events()
    with self._pending_listeners_lock:
        for completed_listener in completed_listeners:
            # check: listener may have been removed since tested `_all_events` as we didn't keep the lock
            if completed_listener in self._pending_listeners:
                # wake up waiting listener
                self._pending_listeners[completed_listener]['event'].set()

def wait(self, event_ids: list[str], timeout: float) -> Tuple[bool, List[Any]]:
    """Wait for a registered listener to complete.

    Blocks until all events for this listener are triggered or until timeout is reached.

    Args:
        event_ids: list of the unique IDs for all events to wait
        timeout: maximum time to wait for replies in seconds

    Returns:
        A tuple consisting of
            - a bool set to `True` if all events for this listener were delivered to this
                request, `False` if not
            - a list of the arbitrary data associated with each delivered event

    Raises:
           FedbiomedNodeToNodeError: timeout has incorrect type or value
    """

    # Check value for timeout, as bad value may cause hard to detect problems
    if not isinstance(timeout, (float, int)) or timeout < 0 or timeout > MAX_TRIGGERED_EVENT_TIMEOUT:
        raise FedbiomedSynchroError(f"{ErrorNumbers.FB324}: Cannot wait {timeout} seconds. "
                                    f"Should be int or float between 0 and {MAX_TRIGGERED_EVENT_TIMEOUT}")

    time_initial = time.time()

    with self._pending_listeners_lock:
        self._id_counter += 1
        listener_id = self._id_counter
        event = threading.Event()
        self._pending_listeners[listener_id] = {
            'start_time': time_initial,
            'event': event,
            'event_ids': event_ids,
        }

    # wait until all events are triggered or timeout is reached
    while not self._all_events(listener_id) and (time.time() < time_initial + timeout):
        # be sure not to hold any lock when waiting !
        event.wait(time_initial + timeout - time.time())
        event.clear()

    with self._pending_listeners_lock:
        with self._triggered_events_lock:
            # check if all events were received (and are still available for delivery)
            all_received = set(self._pending_listeners[listener_id]['event_ids']).\
                issubset(set(self._triggered_events.keys()))

            # create list of delivered data
            events_data = [self._triggered_events[reqid]['data']
                           for reqid in self._pending_listeners[listener_id]['event_ids']
                           if reqid in self._triggered_events]

            if self._remove_delivered:
                # remove all events for this request from the triggered events available
                # for delivery
                for reqid in self._pending_listeners[listener_id]['event_ids']:
                    if reqid in self._triggered_events:
                        self._triggered_events.pop(reqid)

            # remove expired listener
            del self._pending_listeners[listener_id]

    return all_received, events_data