2 QMP Events and EventListeners
4 Asynchronous QMP uses `EventListener` objects to listen for events. An
5 `EventListener` is a FIFO event queue that can be pre-filtered to listen
6 for only specific events. Each `EventListener` instance receives its own
7 copy of events that it hears, so events may be consumed without fear or
8 worry for depriving other listeners of events they need to hear.
11 EventListener Tutorial
12 ----------------------
14 In all of the following examples, we assume that we have a `QMPClient`
15 instantiated named ``qmp`` that is already connected.
18 `listener()` context blocks with one name
19 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
21 The most basic usage is by using the `listener()` context manager to
26 with qmp.listener('STOP') as listener:
27 await qmp.execute('stop')
30 The listener is active only for the duration of the ‘with’ block. This
31 instance listens only for ‘STOP’ events.
34 `listener()` context blocks with two or more names
35 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
37 Multiple events can be selected for by providing any ``Iterable[str]``:
41 with qmp.listener(('STOP', 'RESUME')) as listener:
42 await qmp.execute('stop')
43 event = await listener.get()
44 assert event['event'] == 'STOP'
46 await qmp.execute('cont')
47 event = await listener.get()
48 assert event['event'] == 'RESUME'
51 `listener()` context blocks with no names
52 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
54 By omitting names entirely, you can listen to ALL events.
58 with qmp.listener() as listener:
59 await qmp.execute('stop')
60 event = await listener.get()
61 assert event['event'] == 'STOP'
63 This isn’t a very good use case for this feature: In a non-trivial
64 running system, we may not know what event will arrive next. Grabbing
65 the top of a FIFO queue returning multiple kinds of events may be prone
69 Using async iterators to retrieve events
70 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
72 If you’d like to simply watch what events happen to arrive, you can use
73 the listener as an async iterator:
77 with qmp.listener() as listener:
78 async for event in listener:
79 print(f"Event arrived: {event['event']}")
81 This is analogous to the following code:
85 with qmp.listener() as listener:
87 event = listener.get()
88 print(f"Event arrived: {event['event']}")
90 This event stream will never end, so these blocks will never terminate.
93 Using asyncio.Task to concurrently retrieve events
94 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
96 Since a listener’s event stream will never terminate, it is not likely
97 useful to use that form in a script. For longer-running clients, we can
98 create event handlers by using `asyncio.Task` to create concurrent
103 async def print_events(listener):
105 async for event in listener:
106 print(f"Event arrived: {event['event']}")
107 except asyncio.CancelledError:
110 with qmp.listener() as listener:
111 task = asyncio.Task(print_events(listener))
112 await qmp.execute('stop')
113 await qmp.execute('cont')
117 However, there is no guarantee that these events will be received by the
118 time we leave this context block. Once the context block is exited, the
119 listener will cease to hear any new events, and becomes inert.
121 Be mindful of the timing: the above example will *probably*– but does
122 not *guarantee*– that both STOP/RESUMED events will be printed. The
123 example below outlines how to use listeners outside of a context block.
126 Using `register_listener()` and `remove_listener()`
127 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
129 To create a listener with a longer lifetime, beyond the scope of a
130 single block, create a listener and then call `register_listener()`:
135 def __init__(self, qmp):
137 self.listener = EventListener()
139 async def print_events(self):
141 async for event in self.listener:
142 print(f"Event arrived: {event['event']}")
143 except asyncio.CancelledError:
147 self.task = asyncio.Task(self.print_events)
148 self.qmp.register_listener(self.listener)
149 await qmp.execute('stop')
150 await qmp.execute('cont')
152 async def stop(self):
155 self.qmp.remove_listener(self.listener)
157 The listener can be deactivated by using `remove_listener()`. When it is
158 removed, any possible pending events are cleared and it can be
159 re-registered at a later time.
162 Using the built-in all events listener
163 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
165 The `QMPClient` object creates its own default listener named
166 :py:obj:`~Events.events` that can be used for the same purpose without
167 having to create your own:
171 async def print_events(listener):
173 async for event in listener:
174 print(f"Event arrived: {event['event']}")
175 except asyncio.CancelledError:
178 task = asyncio.Task(print_events(qmp.events))
180 await qmp.execute('stop')
181 await qmp.execute('cont')
187 Using both .get() and async iterators
188 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
190 The async iterator and `get()` methods pull events from the same FIFO
191 queue. If you mix the usage of both, be aware: Events are emitted
192 precisely once per listener.
194 If multiple contexts try to pull events from the same listener instance,
195 events are still emitted only precisely once.
197 This restriction can be lifted by creating additional listeners.
200 Creating multiple listeners
201 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
203 Additional `EventListener` objects can be created at-will. Each one
204 receives its own copy of events, with separate FIFO event queues.
208 my_listener = EventListener()
209 qmp.register_listener(my_listener)
211 await qmp.execute('stop')
212 copy1 = await my_listener.get()
213 copy2 = await qmp.events.get()
215 assert copy1 == copy2
217 In this example, we await an event from both a user-created
218 `EventListener` and the built-in events listener. Both receive the same
225 `EventListener` objects can be cleared, clearing all events seen thus far:
229 await qmp.execute('stop')
231 await qmp.execute('cont')
232 event = await qmp.events.get()
233 assert event['event'] == 'RESUME'
235 `EventListener` objects are FIFO queues. If events are not consumed,
236 they will remain in the queue until they are witnessed or discarded via
237 `clear()`. FIFO queues will be drained automatically upon leaving a
238 context block, or when calling `remove_listener()`.
241 Accessing listener history
242 ~~~~~~~~~~~~~~~~~~~~~~~~~~
244 `EventListener` objects record their history. Even after being cleared,
245 you can obtain a record of all events seen so far:
249 await qmp.execute('stop')
250 await qmp.execute('cont')
253 assert len(qmp.events.history) == 2
254 assert qmp.events.history[0]['event'] == 'STOP'
255 assert qmp.events.history[1]['event'] == 'RESUME'
257 The history is updated immediately and does not require the event to be
264 `EventListener` objects can be given complex filtering criteria if names
269 def job1_filter(event) -> bool:
270 event_data = event.get('data', {})
271 event_job_id = event_data.get('id')
272 return event_job_id == "job1"
274 with qmp.listener('JOB_STATUS_CHANGE', job1_filter) as listener:
275 await qmp.execute('blockdev-backup', arguments={'job-id': 'job1', ...})
276 async for event in listener:
277 if event['data']['status'] == 'concluded':
280 These filters might be most useful when parameterized. `EventListener`
281 objects expect a function that takes only a single argument (the raw
282 event, as a `Message`) and returns a bool; True if the event should be
283 accepted into the stream. You can create a function that adapts this
284 signature to accept configuration parameters:
288 def job_filter(job_id: str) -> EventFilter:
289 def filter(event: Message) -> bool:
290 return event['data']['id'] == job_id
293 with qmp.listener('JOB_STATUS_CHANGE', job_filter('job2')) as listener:
294 await qmp.execute('blockdev-backup', arguments={'job-id': 'job2', ...})
295 async for event in listener:
296 if event['data']['status'] == 'concluded':
300 Activating an existing listener with `listen()`
301 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
303 Listeners with complex, long configurations can also be created manually
304 and activated temporarily by using `listen()` instead of `listener()`:
308 listener = EventListener(('BLOCK_JOB_COMPLETED', 'BLOCK_JOB_CANCELLED',
309 'BLOCK_JOB_ERROR', 'BLOCK_JOB_READY',
310 'BLOCK_JOB_PENDING', 'JOB_STATUS_CHANGE'))
312 with qmp.listen(listener):
313 await qmp.execute('blockdev-backup', arguments={'job-id': 'job3', ...})
314 async for event in listener:
316 if event['event'] == 'BLOCK_JOB_COMPLETED':
319 Any events that are not witnessed by the time the block is left will be
320 cleared from the queue; entering the block is an implicit
321 `register_listener()` and leaving the block is an implicit
325 Activating multiple existing listeners with `listen()`
326 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
328 While `listener()` is only capable of creating a single listener,
329 `listen()` is capable of activating multiple listeners simultaneously:
333 def job_filter(job_id: str) -> EventFilter:
334 def filter(event: Message) -> bool:
335 return event['data']['id'] == job_id
338 jobA = EventListener('JOB_STATUS_CHANGE', job_filter('jobA'))
339 jobB = EventListener('JOB_STATUS_CHANGE', job_filter('jobB'))
341 with qmp.listen(jobA, jobB):
342 qmp.execute('blockdev-create', arguments={'job-id': 'jobA', ...})
343 qmp.execute('blockdev-create', arguments={'job-id': 'jobB', ...})
345 async for event in jobA.get():
346 if event['data']['status'] == 'concluded':
348 async for event in jobB.get():
349 if event['data']['status'] == 'concluded':
353 Extending the `EventListener` class
354 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
356 In the case that a more specialized `EventListener` is desired to
357 provide either more functionality or more compact syntax for specialized
358 cases, it can be extended.
360 One of the key methods to extend or override is
361 :py:meth:`~EventListener.accept()`. The default implementation checks an
362 incoming message for:
364 1. A qualifying name, if any :py:obj:`~EventListener.names` were
365 specified at initialization time
366 2. That :py:obj:`~EventListener.event_filter()` returns True.
368 This can be modified however you see fit to change the criteria for
369 inclusion in the stream.
371 For convenience, a ``JobListener`` class could be created that simply
372 bakes in configuration so it does not need to be repeated:
376 class JobListener(EventListener):
377 def __init__(self, job_id: str):
378 super().__init__(('BLOCK_JOB_COMPLETED', 'BLOCK_JOB_CANCELLED',
379 'BLOCK_JOB_ERROR', 'BLOCK_JOB_READY',
380 'BLOCK_JOB_PENDING', 'JOB_STATUS_CHANGE'))
383 def accept(self, event) -> bool:
384 if not super().accept(event):
386 if event['event'] in ('BLOCK_JOB_PENDING', 'JOB_STATUS_CHANGE'):
387 return event['data']['id'] == job_id
388 return event['data']['device'] == job_id
390 From here on out, you can conjure up a custom-purpose listener that
391 listens only for job-related events for a specific job-id easily:
395 listener = JobListener('job4')
396 with qmp.listener(listener):
397 await qmp.execute('blockdev-backup', arguments={'job-id': 'job4', ...})
398 async for event in listener:
400 if event['event'] == 'BLOCK_JOB_COMPLETED':
404 Experimental Interfaces & Design Issues
405 ---------------------------------------
407 These interfaces are not ones I am sure I will keep or otherwise modify
410 qmp.listener()’s type signature
411 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
413 `listener()` does not return anything, because it was assumed the caller
414 already had a handle to the listener. However, for
415 ``qmp.listener(EventListener())`` forms, the caller will not have saved
416 a handle to the listener.
418 Because this function can accept *many* listeners, I found it hard to
419 accurately type in a way where it could be used in both “one” or “many”
420 forms conveniently and in a statically type-safe manner.
422 Ultimately, I removed the return altogether, but perhaps with more time
423 I can work out a way to re-add it.
432 from contextlib import contextmanager
446 from .error import QMPError
447 from .message import Message
450 EventNames = Union[str, Iterable[str], None]
451 EventFilter = Callable[[Message], bool]
454 class ListenerError(QMPError):
456 Generic error class for `EventListener`-related problems.
462 Selectively listens for events with runtime configurable filtering.
464 This class is designed to be directly usable for the most common cases,
465 but it can be extended to provide more rigorous control.
468 One or more names of events to listen for.
469 When not provided, listen for ALL events.
471 An optional event filtering function.
472 When names are also provided, this acts as a secondary filter.
474 When ``names`` and ``event_filter`` are both provided, the names
475 will be filtered first, and then the filter function will be called
476 second. The event filter function can assume that the format of the
477 event is a known format.
481 names: EventNames = None,
482 event_filter: Optional[EventFilter] = None,
484 # Queue of 'heard' events yet to be witnessed by a caller.
485 self._queue: 'asyncio.Queue[Message]' = asyncio.Queue()
487 # Intended as a historical record, NOT a processing queue or backlog.
488 self._history: List[Message] = []
490 #: Primary event filter, based on one or more event names.
491 self.names: Set[str] = set()
492 if isinstance(names, str):
493 self.names.add(names)
494 elif names is not None:
495 self.names.update(names)
497 #: Optional, secondary event filter.
498 self.event_filter: Optional[EventFilter] = event_filter
501 def history(self) -> Tuple[Message, ...]:
503 A read-only history of all events seen so far.
505 This represents *every* event, including those not yet witnessed
506 via `get()` or ``async for``. It persists between `clear()`
507 calls and is immutable.
509 return tuple(self._history)
511 def accept(self, event: Message) -> bool:
513 Determine if this listener accepts this event.
515 This method determines which events will appear in the stream.
516 The default implementation simply checks the event against the
517 list of names and the event_filter to decide if this
518 `EventListener` accepts a given event. It can be
519 overridden/extended to provide custom listener behavior.
521 User code is not expected to need to invoke this method.
523 :param event: The event under consideration.
524 :return: `True`, if this listener accepts this event.
526 name_ok = (not self.names) or (event['event'] in self.names)
528 (not self.event_filter) or self.event_filter(event)
531 async def put(self, event: Message) -> None:
533 Conditionally put a new event into the FIFO queue.
535 This method is not designed to be invoked from user code, and it
536 should not need to be overridden. It is a public interface so
537 that `QMPClient` has an interface by which it can inform
538 registered listeners of new events.
540 The event will be put into the queue if
541 :py:meth:`~EventListener.accept()` returns `True`.
543 :param event: The new event to put into the FIFO queue.
545 if not self.accept(event):
548 self._history.append(event)
549 await self._queue.put(event)
551 async def get(self) -> Message:
553 Wait for the very next event in this stream.
555 If one is already available, return that one.
557 return await self._queue.get()
559 def empty(self) -> bool:
561 Return `True` if there are no pending events.
563 return self._queue.empty()
565 def clear(self) -> List[Message]:
567 Clear this listener of all pending events.
569 Called when an `EventListener` is being unregistered, this clears the
570 pending FIFO queue synchronously. It can be also be used to
571 manually clear any pending events, if desired.
573 :return: The cleared events, if any.
576 Take care when discarding events. Cleared events will be
577 silently tossed on the floor. All events that were ever
578 accepted by this listener are visible in `history()`.
583 events.append(self._queue.get_nowait())
584 except asyncio.QueueEmpty:
589 def __aiter__(self) -> AsyncIterator[Message]:
592 async def __anext__(self) -> Message:
594 Enables the `EventListener` to function as an async iterator.
596 It may be used like this:
600 async for event in listener:
603 These iterators will never terminate of their own accord; you
604 must provide break conditions or otherwise prepare to run them
605 in an `asyncio.Task` that can be cancelled.
607 return await self.get()
612 Events is a mix-in class that adds event functionality to the QMP class.
614 It's designed specifically as a mix-in for `QMPClient`, and it
615 relies upon the class it is being mixed into having a 'logger'
618 def __init__(self) -> None:
619 self._listeners: List[EventListener] = []
621 #: Default, all-events `EventListener`.
622 self.events: EventListener = EventListener()
623 self.register_listener(self.events)
625 # Parent class needs to have a logger
626 self.logger: logging.Logger
628 async def _event_dispatch(self, msg: Message) -> None:
630 Given a new event, propagate it to all of the active listeners.
632 :param msg: The event to propagate.
634 for listener in self._listeners:
635 await listener.put(msg)
637 def register_listener(self, listener: EventListener) -> None:
639 Register and activate an `EventListener`.
641 :param listener: The listener to activate.
642 :raise ListenerError: If the given listener is already registered.
644 if listener in self._listeners:
645 raise ListenerError("Attempted to re-register existing listener")
646 self.logger.debug("Registering %s.", str(listener))
647 self._listeners.append(listener)
649 def remove_listener(self, listener: EventListener) -> None:
651 Unregister and deactivate an `EventListener`.
653 The removed listener will have its pending events cleared via
654 `clear()`. The listener can be re-registered later when
657 :param listener: The listener to deactivate.
658 :raise ListenerError: If the given listener is not registered.
660 if listener == self.events:
661 raise ListenerError("Cannot remove the default listener.")
662 self.logger.debug("Removing %s.", str(listener))
664 self._listeners.remove(listener)
667 def listen(self, *listeners: EventListener) -> Iterator[None]:
669 Context manager: Temporarily listen with an `EventListener`.
671 Accepts one or more `EventListener` objects and registers them,
672 activating them for the duration of the context block.
674 `EventListener` objects will have any pending events in their
675 FIFO queue cleared upon exiting the context block, when they are
678 :param \*listeners: One or more EventListeners to activate.
679 :raise ListenerError: If the given listener(s) are already active.
684 for listener in listeners:
685 self.register_listener(listener)
686 _added.append(listener)
691 for listener in _added:
692 self.remove_listener(listener)
697 names: EventNames = (),
698 event_filter: Optional[EventFilter] = None
699 ) -> Iterator[EventListener]:
701 Context manager: Temporarily listen with a new `EventListener`.
703 Creates an `EventListener` object and registers it, activating
704 it for the duration of the context block.
707 One or more names of events to listen for.
708 When not provided, listen for ALL events.
710 An optional event filtering function.
711 When names are also provided, this acts as a secondary filter.
713 :return: The newly created and active `EventListener`.
715 listener = EventListener(names, event_filter)
716 with self.listen(listener):