hololinked.core.events.EventDispatcher

class EventDispatcher:
    """
    The worker that pushes an event.

    The separation is necessary between `Event` and `EventDispatcher` to allow class level definitions of the `Event`
    """

    __slots__ = ["_descriptor", "_owner_inst", "_publisher", "unique_identifier"]

    def __init__(
        self,
        unique_identifier: str,
        publisher: EventBus | None,
        owner_inst: Thing,
        descriptor: Event,
    ) -> None:
        self.unique_identifier = unique_identifier
        self._owner_inst = owner_inst
        self._descriptor = descriptor
        self.publisher = publisher

    @property
    def publisher(self) -> EventBus:
        """The `EventBus` this event is registered with, through which it reaches every protocol."""
        return self._publisher  # ty: ignore[invalid-return-type]

    @publisher.setter
    def publisher(self, value: EventBus | None) -> None:
        from hololinked.core.eventloop import EventBus

        if not hasattr(self, "_publisher"):
            self._publisher = value
        elif not isinstance(value, EventBus):
            raise AttributeError("Publisher must be of type EventBus. Given type: " + str(type(value)))

    def push(self, data: Any) -> None:
        """
        Publish the event.

        Multipart payloads are not supported. Supply either a serializable object or a
        bytes object for binary data, not both.

        Parameters
        ----------
        data: Any
            payload of the event
        """
        self.publisher.publish(self.unique_identifier, data)

    def receive_acknowledgement(self, timeout: float | None) -> bool:
        """
        Receive acknowledgement for an event that was just pushed.

        Not Implemented.

        Parameters
        ----------
        timeout: float | int | None
            timeout for receiving the acknowledgement, in seconds. If None, wait indefinitely.

        Returns
        -------
        bool
            True if acknowledgement is received, False if timeout is reached.
        """
        raise NotImplementedError("Event acknowledgement is not implemented yet.")
        return self._synchronize_event.wait(timeout=timeout)

    def _set_acknowledgement(self, *args, **kwargs) -> None:
        """
        Once an acknowledgement is received from the client, this function is called to set the event.

        Not Implemented.
        """
        raise NotImplementedError("Event acknowledgement is not implemented yet.")
        self._synchronize_event.set()