Source code for hangups.event
"""Simple event observer system supporting asyncio.
Observers must be removed to avoid memory leaks.
"""
import asyncio
import logging
logger = logging.getLogger(__name__)
[docs]class Event:
"""An event that can notify subscribers with arguments when fired.
Args:
name (str): Name of the new event.
"""
def __init__(self, name):
self._name = str(name)
self._observers = []
[docs] def add_observer(self, callback):
"""Add an observer to this event.
Args:
callback: A function or coroutine callback to call when the event
is fired.
Raises:
ValueError: If the callback has already been added.
"""
if callback in self._observers:
raise ValueError('{} is already an observer of {}'
.format(callback, self))
self._observers.append(callback)
[docs] def remove_observer(self, callback):
"""Remove an observer from this event.
Args:
callback: A function or coroutine callback to remove from this
event.
Raises:
ValueError: If the callback is not an observer of this event.
"""
if callback not in self._observers:
raise ValueError('{} is not an observer of {}'
.format(callback, self))
self._observers.remove(callback)
[docs] async def fire(self, *args, **kwargs):
"""Fire this event, calling all observers with the same arguments."""
logger.debug('Fired {}'.format(self))
for observer in self._observers:
gen = observer(*args, **kwargs)
if asyncio.iscoroutinefunction(observer):
await gen
def __repr__(self):
return 'Event(\'{}\')'.format(self._name)