Skip to content

Publish-Subscribe

hololinked.core.eventloop.pubsub.EventBus

Registry of the events Things can push, and fan-out to whoever is listening.

Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class EventBus:
    """Registry of the events `Thing`s can push, and fan-out to whoever is listening."""

    def __init__(self) -> None:
        self._events = dict()  # type: dict[str, RegisteredEvent]
        self._subscribers = dict()  # type: dict[str, list[Callable[[Any], None]]]
        self._lock = threading.Lock()

    @property
    def event_ids(self) -> set[str]:
        """Unique identifiers of every registered event."""
        return set(self._events)

    def event_for(self, event_id: str) -> RegisteredEvent:
        """
        The event registered under an identifier.

        A subscriber that needs the event itself - to encode its payloads, say - asks for it once,
        when it subscribes, instead of being handed it with every push.

        Parameters
        ----------
        event_id: str
            unique identifier of the event, as `<thing id>/<event name>`

        Returns
        -------
        RegisteredEvent
            the descriptor and the owner registered under that identifier

        Raises
        ------
        KeyError
            if no such event is registered with this bus
        """
        with self._lock:
            if event_id not in self._events:
                raise KeyError(f"event {event_id} is not registered, nothing would ever be delivered")
            return self._events[event_id]

    def register(self, event: Event, owner: Thing) -> None:
        """
        Register an event specific to `Thing` instance.

        Parameters
        ----------
        event: Event
            the descriptor the event was declared as
        owner: Thing
            the instance that pushes the event
        """
        unique_identifier = event.get_unique_identifier(owner)
        with self._lock:
            self._events[unique_identifier] = RegisteredEvent(event, owner, unique_identifier)

    def unregister(self, event: Event, owner: Thing) -> None:
        """
        Unregister an event specific to a `Thing` instance, so that publishing it raises.

        Parameters
        ----------
        event: Event
            the descriptor the event was declared as
        owner: Thing
            the instance that will push the event
        """
        unique_identifier = event.get_unique_identifier(owner)
        with self._lock:
            if self._events.pop(unique_identifier, None) is None:
                warnings.warn(
                    f"event {unique_identifier} not found, did you mean to unregister another event?",
                    UserWarning,
                    stacklevel=2,
                )

    def subscribe(self, callback: Callable[[Any], None], event_id: str) -> None:
        """
        Subscribe to an event with a callback that will be invoked whenever the event is published.

        Parameters
        ----------
        callback: Callable[[Any], None]
            called synchronously, on whichever thread pushed the event - which is a `Thing`'s own
            thread, not the subscriber's. A subscriber that owns loop-bound state (a protocol
            server's connections, say) must hop to its loop itself, with `call_soon_threadsafe`.
        event_id: str
            unique identifier of the event to listen for, as `<thing id>/<event name>`
        """
        with self._lock:
            callbacks = self._subscribers.setdefault(event_id, [])
            if callback not in callbacks:
                callbacks.append(callback)

    def unsubscribe(self, callback: Callable[[Any], None], event_id: str) -> None:
        """
        Unsubscribe from an event.

        Parameters
        ----------
        callback: Callable[[Any], None]
            a callback previously given to `subscribe()`. Unknown callbacks are ignored.
        event_id: str
            the event it was subscribed to. Unknown events are ignored.
        """
        with self._lock:
            callbacks = self._subscribers.get(event_id, None)
            if callbacks is None or callback not in callbacks:
                return
            callbacks.remove(callback)
            if not callbacks:
                del self._subscribers[event_id]  # a long-lived bus must not keep a bucket per past listener

    def publish(self, event_id: str, data: Any) -> None:
        """
        Hand one event's payload to its own subscribers, in subscription order.

        The lock is held for the whole fan-out, which is what serializes concurrent pushes from
        different `Thing` threads onto each subscriber's wire.

        Parameters
        ----------
        event_id: str
            unique identifier of the event being pushed, as `<thing id>/<event name>`
        data: Any
            its payload, unencoded

        Raises
        ------
        AttributeError
            if the event is not registered with this bus
        """
        with self._lock:
            if event_id not in self._events:
                raise AttributeError(f"event name {event_id} not registered")
            for callback in self._subscribers.get(event_id, ()):
                callback(data)

Functions

__init__ 

__init__() -> None
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
34
35
36
37
def __init__(self) -> None:
    self._events = dict()  # type: dict[str, RegisteredEvent]
    self._subscribers = dict()  # type: dict[str, list[Callable[[Any], None]]]
    self._lock = threading.Lock()

register 

register(event: Event, owner: Thing) -> None

Register an event specific to Thing instance.

Parameters:

Name Type Description Default
event
Event

the descriptor the event was declared as

 required
owner
Thing

the instance that pushes the event

 required
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def register(self, event: Event, owner: Thing) -> None:
    """
    Register an event specific to `Thing` instance.

    Parameters
    ----------
    event: Event
        the descriptor the event was declared as
    owner: Thing
        the instance that pushes the event
    """
    unique_identifier = event.get_unique_identifier(owner)
    with self._lock:
        self._events[unique_identifier] = RegisteredEvent(event, owner, unique_identifier)

unregister 

unregister(event: Event, owner: Thing) -> None

Unregister an event specific to a Thing instance, so that publishing it raises.

Parameters:

Name Type Description Default
event
Event

the descriptor the event was declared as

 required
owner
Thing

the instance that will push the event

 required
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def unregister(self, event: Event, owner: Thing) -> None:
    """
    Unregister an event specific to a `Thing` instance, so that publishing it raises.

    Parameters
    ----------
    event: Event
        the descriptor the event was declared as
    owner: Thing
        the instance that will push the event
    """
    unique_identifier = event.get_unique_identifier(owner)
    with self._lock:
        if self._events.pop(unique_identifier, None) is None:
            warnings.warn(
                f"event {unique_identifier} not found, did you mean to unregister another event?",
                UserWarning,
                stacklevel=2,
            )

subscribe 

subscribe(callback: Callable[[Any], None], event_id: str) -> None

Subscribe to an event with a callback that will be invoked whenever the event is published.

Parameters:

Name Type Description Default
callback
Callable[[Any], None]

called synchronously, on whichever thread pushed the event - which is a Thing's own thread, not the subscriber's. A subscriber that owns loop-bound state (a protocol server's connections, say) must hop to its loop itself, with call_soon_threadsafe.

 required
event_id
str

unique identifier of the event to listen for, as <thing id>/<event name>

 required
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def subscribe(self, callback: Callable[[Any], None], event_id: str) -> None:
    """
    Subscribe to an event with a callback that will be invoked whenever the event is published.

    Parameters
    ----------
    callback: Callable[[Any], None]
        called synchronously, on whichever thread pushed the event - which is a `Thing`'s own
        thread, not the subscriber's. A subscriber that owns loop-bound state (a protocol
        server's connections, say) must hop to its loop itself, with `call_soon_threadsafe`.
    event_id: str
        unique identifier of the event to listen for, as `<thing id>/<event name>`
    """
    with self._lock:
        callbacks = self._subscribers.setdefault(event_id, [])
        if callback not in callbacks:
            callbacks.append(callback)

unsubscribe 

unsubscribe(callback: Callable[[Any], None], event_id: str) -> None

Unsubscribe from an event.

Parameters:

Name Type Description Default
callback
Callable[[Any], None]

a callback previously given to subscribe(). Unknown callbacks are ignored.

 required
event_id
str

the event it was subscribed to. Unknown events are ignored.

 required
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def unsubscribe(self, callback: Callable[[Any], None], event_id: str) -> None:
    """
    Unsubscribe from an event.

    Parameters
    ----------
    callback: Callable[[Any], None]
        a callback previously given to `subscribe()`. Unknown callbacks are ignored.
    event_id: str
        the event it was subscribed to. Unknown events are ignored.
    """
    with self._lock:
        callbacks = self._subscribers.get(event_id, None)
        if callbacks is None or callback not in callbacks:
            return
        callbacks.remove(callback)
        if not callbacks:
            del self._subscribers[event_id]  # a long-lived bus must not keep a bucket per past listener

publish 

publish(event_id: str, data: Any) -> None

Hand one event's payload to its own subscribers, in subscription order.

The lock is held for the whole fan-out, which is what serializes concurrent pushes from different Thing threads onto each subscriber's wire.

Parameters:

Name Type Description Default
event_id
str

unique identifier of the event being pushed, as <thing id>/<event name>

 required
data
Any

its payload, unencoded

 required

Raises:

Type Description
AttributeError

if the event is not registered with this bus
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def publish(self, event_id: str, data: Any) -> None:
    """
    Hand one event's payload to its own subscribers, in subscription order.

    The lock is held for the whole fan-out, which is what serializes concurrent pushes from
    different `Thing` threads onto each subscriber's wire.

    Parameters
    ----------
    event_id: str
        unique identifier of the event being pushed, as `<thing id>/<event name>`
    data: Any
        its payload, unencoded

    Raises
    ------
    AttributeError
        if the event is not registered with this bus
    """
    with self._lock:
        if event_id not in self._events:
            raise AttributeError(f"event name {event_id} not registered")
        for callback in self._subscribers.get(event_id, ()):
            callback(data)

event_for 

event_for(event_id: str) -> RegisteredEvent

The event registered under an identifier.

A subscriber that needs the event itself - to encode its payloads, say - asks for it once, when it subscribes, instead of being handed it with every push.

Parameters:

Name Type Description Default
event_id
str

unique identifier of the event, as <thing id>/<event name>

 required

Returns:

Type Description
RegisteredEvent

the descriptor and the owner registered under that identifier

Raises:

Type Description
KeyError

if no such event is registered with this bus
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def event_for(self, event_id: str) -> RegisteredEvent:
    """
    The event registered under an identifier.

    A subscriber that needs the event itself - to encode its payloads, say - asks for it once,
    when it subscribes, instead of being handed it with every push.

    Parameters
    ----------
    event_id: str
        unique identifier of the event, as `<thing id>/<event name>`

    Returns
    -------
    RegisteredEvent
        the descriptor and the owner registered under that identifier

    Raises
    ------
    KeyError
        if no such event is registered with this bus
    """
    with self._lock:
        if event_id not in self._events:
            raise KeyError(f"event {event_id} is not registered, nothing would ever be delivered")
        return self._events[event_id]

event_ids 

event_ids() -> set[str]

Unique identifiers of every registered event.

Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
39
40
41
42
@property
def event_ids(self) -> set[str]:
    """Unique identifiers of every registered event."""
    return set(self._events)

hololinked.core.eventloop.pubsub.EventSubscription

One event's payloads, delivered onto the subscriber's own loop.

Fans out synchronously on whichever thread pushed the event.

Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class EventSubscription:
    """
    One event's payloads, delivered onto the subscriber's own loop.

    Fans out synchronously on whichever thread pushed the event.
    """

    def __init__(
        self,
        bus: EventBus,
        unique_identifier: str,
        loop: asyncio.AbstractEventLoop | None = None,
        maxsize: int = 3,
    ) -> None:
        """
        Subscribe to one event.

        Parameters
        ----------
        bus: EventBus
            the bus to subscribe to
        unique_identifier: str
            the event to listen for, as `<thing id>/<event name>`
        loop: asyncio.AbstractEventLoop, optional
            the loop to deliver on. The running one by default.
        maxsize: int
            how many payloads to hold before dropping the oldest

        Raises
        ------
        KeyError
            if no such event is registered with the bus
        """
        self.event = bus.event_for(unique_identifier)
        self.bus = bus
        self.unique_identifier = unique_identifier
        self.loop = loop or asyncio.get_running_loop()
        self.queue = asyncio.Queue(maxsize=maxsize)  # type: asyncio.Queue
        self.dropped = 0
        bus.subscribe(self.event_callback_threadsafe, unique_identifier)

    def event_callback_threadsafe(self, data: Any) -> None:
        """Called on the pushing thread - hand the payload over and get out of the way."""
        try:
            self.loop.call_soon_threadsafe(self._handle_event_publish, data)
        except RuntimeError:
            pass  # the subscriber's loop is gone, so there is nobody left to deliver to

    def _handle_event_publish(self, data: Any) -> None:
        """Called on the subscriber's loop."""
        if self.queue.full():
            try:
                self.queue.get_nowait()
                self.dropped += 1
            except asyncio.QueueEmpty:
                pass
        self.queue.put_nowait(data)

    async def receive(self, timeout: float | None = None) -> Any:
        """
        Wait for the next payload.

        Parameters
        ----------
        timeout: float, optional
            seconds to wait. Waits indefinitely when not given.

        Returns
        -------
        Any
            the payload, unencoded.

        Raises
        ------
        TimeoutError
            if nothing was pushed before the timeout elapsed.
        """
        if timeout is None:
            return await self.queue.get()
        return await asyncio.wait_for(self.queue.get(), timeout)

    def encode(self, data: Any) -> tuple[bytes, str]:
        """
        Encode one of this event's payloads the way its objekt is registered to be encoded.

        Parameters
        ----------
        data: Any
            a payload from `receive()`

        Returns
        -------
        tuple[bytes, str]
            the encoded body and the content type to declare for it
        """
        return encode_event(self.event, data)

    def unsubscribe(self) -> None:
        """Stop receiving. Safe to call more than once."""
        self.bus.unsubscribe(self.event_callback_threadsafe, self.unique_identifier)

Attributes

event instance-attribute 

event = event_for(unique_identifier)

bus instance-attribute 

bus = bus

loop instance-attribute 

loop = loop or get_running_loop()

queue instance-attribute 

queue = Queue(maxsize=maxsize)

dropped instance-attribute 

dropped = 0

unique_identifier instance-attribute 

unique_identifier = unique_identifier

Functions

__init__ 

__init__(bus: EventBus, unique_identifier: str, loop: AbstractEventLoop | None = None, maxsize: int = 3) -> None

Subscribe to one event.

Parameters:

Name Type Description Default
bus
EventBus

the bus to subscribe to

 required
unique_identifier
str

the event to listen for, as <thing id>/<event name>

 required
loop
AbstractEventLoop | None

the loop to deliver on. The running one by default.

 None
maxsize
int

how many payloads to hold before dropping the oldest

 3

Raises:

Type Description
KeyError

if no such event is registered with the bus
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def __init__(
    self,
    bus: EventBus,
    unique_identifier: str,
    loop: asyncio.AbstractEventLoop | None = None,
    maxsize: int = 3,
) -> None:
    """
    Subscribe to one event.

    Parameters
    ----------
    bus: EventBus
        the bus to subscribe to
    unique_identifier: str
        the event to listen for, as `<thing id>/<event name>`
    loop: asyncio.AbstractEventLoop, optional
        the loop to deliver on. The running one by default.
    maxsize: int
        how many payloads to hold before dropping the oldest

    Raises
    ------
    KeyError
        if no such event is registered with the bus
    """
    self.event = bus.event_for(unique_identifier)
    self.bus = bus
    self.unique_identifier = unique_identifier
    self.loop = loop or asyncio.get_running_loop()
    self.queue = asyncio.Queue(maxsize=maxsize)  # type: asyncio.Queue
    self.dropped = 0
    bus.subscribe(self.event_callback_threadsafe, unique_identifier)

receive async 

receive(timeout: float | None = None) -> Any

Wait for the next payload.

Parameters:

Name Type Description Default
timeout
float | None

seconds to wait. Waits indefinitely when not given.

 None

Returns:

Type Description
Any

the payload, unencoded.

Raises:

Type Description
TimeoutError

if nothing was pushed before the timeout elapsed.
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
async def receive(self, timeout: float | None = None) -> Any:
    """
    Wait for the next payload.

    Parameters
    ----------
    timeout: float, optional
        seconds to wait. Waits indefinitely when not given.

    Returns
    -------
    Any
        the payload, unencoded.

    Raises
    ------
    TimeoutError
        if nothing was pushed before the timeout elapsed.
    """
    if timeout is None:
        return await self.queue.get()
    return await asyncio.wait_for(self.queue.get(), timeout)

encode 

encode(data: Any) -> tuple[bytes, str]

Encode one of this event's payloads the way its objekt is registered to be encoded.

Parameters:

Name Type Description Default
data
Any

a payload from receive()

 required

Returns:

Type Description
tuple[bytes, str]

the encoded body and the content type to declare for it
Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def encode(self, data: Any) -> tuple[bytes, str]:
    """
    Encode one of this event's payloads the way its objekt is registered to be encoded.

    Parameters
    ----------
    data: Any
        a payload from `receive()`

    Returns
    -------
    tuple[bytes, str]
        the encoded body and the content type to declare for it
    """
    return encode_event(self.event, data)

event_callback_threadsafe 

event_callback_threadsafe(data: Any) -> None

Called on the pushing thread - hand the payload over and get out of the way.

Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
210
211
212
213
214
215
def event_callback_threadsafe(self, data: Any) -> None:
    """Called on the pushing thread - hand the payload over and get out of the way."""
    try:
        self.loop.call_soon_threadsafe(self._handle_event_publish, data)
    except RuntimeError:
        pass  # the subscriber's loop is gone, so there is nobody left to deliver to

unsubscribe 

unsubscribe() -> None

Stop receiving. Safe to call more than once.

Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
266
267
268
def unsubscribe(self) -> None:
    """Stop receiving. Safe to call more than once."""
    self.bus.unsubscribe(self.event_callback_threadsafe, self.unique_identifier)

hololinked.core.eventloop.pubsub.RegisteredEvent

Bases: NamedTuple

A registered event, consisting of its descriptor, owner, and unique identifier.

Source code in repo/hololinked/hololinked/core/eventloop/pubsub.py 
20
21
22
23
24
25
26
27
28
class RegisteredEvent(NamedTuple):
    """A registered event, consisting of its descriptor, owner, and unique identifier."""

    descriptor: Event
    """the class level descriptor the event was declared as."""
    owner: Thing
    """the instance that pushes it."""
    unique_identifier: str
    """what the two of them are known by outside the `Thing`, as `<thing id>/<event name>`."""

Attributes

descriptor instance-attribute 

descriptor: Event

the class level descriptor the event was declared as.

owner instance-attribute 

owner: Thing

the instance that pushes it.

unique_identifier instance-attribute 

unique_identifier: str

what the two of them are known by outside the Thing, as <thing id>/<event name>.