Skip to content

hololinked.core.events.EventDispatcher

The worker that pushes an event. The separation is necessary between Event and EventDispatcher to allow class level definitions of the Event

Source code in hololinked/hololinked/core/events.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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__ = ["_unique_identifier", "_publisher", "_owner_inst", "_descriptor"]

    def __init__(
        self,
        unique_identifier: str,
        publisher: "EventPublisher",
        owner_inst: ParameterizedMetaclass,
        descriptor: Event,
    ) -> None:
        self._unique_identifier = unique_identifier
        self._owner_inst = owner_inst
        self._descriptor = descriptor
        self.publisher = publisher

    @property
    def publisher(self) -> "EventPublisher":
        """Event publishing PUB socket owning object"""
        return self._publisher

    @publisher.setter
    def publisher(self, value: "EventPublisher") -> None:
        if not hasattr(self, "_publisher"):
            self._publisher = value
        elif not isinstance(value, EventPublisher):
            raise AttributeError("Publisher must be of type EventPublisher. Given type: " + str(type(value)))
        if self._publisher is not None:
            self._publisher.register(self)

    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, data=data)

    def receive_acknowledgement(self, timeout: float | int | None) -> bool:
        """
        Not Implemented.

        Receive acknowledgement for an event that was just pushed.
        """
        raise NotImplementedError("Event acknowledgement is not implemented yet.")
        return self._synchronize_event.wait(timeout=timeout)

    def _set_acknowledgement(self, *args, **kwargs) -> None:
        """
        Not Implemented.

        Once an acknowledgement is received from the client, this function is called to set the event.
        """
        raise NotImplementedError("Event acknowledgement is not implemented yet.")
        self._synchronize_event.set()

Functions

push 

push(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:

Name Type Description Default

data

Any

payload of the event

 required
Source code in hololinked/hololinked/core/events.py 
120
121
122
123
124
125
126
127
128
129
130
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, data=data)

receive_acknowledgement 

receive_acknowledgement(timeout: float | int | None) -> bool

Not Implemented.

Receive acknowledgement for an event that was just pushed.

Source code in hololinked/hololinked/core/events.py 
132
133
134
135
136
137
138
139
def receive_acknowledgement(self, timeout: float | int | None) -> bool:
    """
    Not Implemented.

    Receive acknowledgement for an event that was just pushed.
    """
    raise NotImplementedError("Event acknowledgement is not implemented yet.")
    return self._synchronize_event.wait(timeout=timeout)