HivenEventHandler and Listeners¶
Important
DispatchEventListener is the base Class for SingleDispatchEventListener and MultiDispatchEventListener, which should not be used
openhivenpy.events.DispatchEventListener
¶
Base Class for all DispatchEventListeners
awaitable: Union[Awaitable, Callable]
property
readonly
¶
client: HivenClient
property
readonly
¶
event_name: str
property
readonly
¶
Methods¶
__call__(self, *args, **kwargs)
special
¶
Returns the dispatch function of the class itself
:param event_data: Data of the received event :param args: Args that will be passed to the coroutine :param kwargs: Kwargs that will be passed to the coroutine :return: The coroutine instance which can be used in an await expression or asyncio functions
Source code in openhivenpy\events.py
def __call__(self, *args, **kwargs) -> Union[Coroutine, Awaitable]:
"""
Returns the dispatch function of the class itself
:param event_data: Data of the received event
:param args: Args that will be passed to the coroutine
:param kwargs: Kwargs that will be passed to the coroutine
:return: The coroutine instance which can be used in an await
expression or asyncio functions
"""
dispatch: Union[Callable, Union[Awaitable, Callable]] = getattr(
self, 'dispatch'
)
return dispatch(*args, **kwargs)
__init__(self, client, event_name, awaitable)
special
¶
Source code in openhivenpy\events.py
def __init__(
self,
client: HivenClient,
event_name: str,
awaitable: Union[Awaitable, Callable]
):
self._client = client
self._event_name = event_name
self._awaitable: Optional[Awaitable] = None
self.set_awaitable(awaitable)
self._client.add_listener(self)
__repr__(self)
special
¶
Source code in openhivenpy\events.py
def __repr__(self):
info = [
('event_name', getattr(self, 'event_name', None)),
('awaitable', getattr(self, 'awaitable', None))
]
return '<MultiDispatchEventListener {}>'.format(
' '.join('%s=%s' % t for t in info)
)
__str__(self)
special
¶
Source code in openhivenpy\events.py
def __str__(self):
return f"<{self.__class__.__name__} for event {self.event_name}>"
dispatch(self)
async
¶
Source code in openhivenpy\events.py
async def dispatch(self) -> None:
...
set_awaitable(self, awaitable)
¶
Sets the coroutine of the event_listener, which will be called when ws-events are received
:param awaitable: Awaitable (Coroutine Function). Can NOT be a
initialised coroutine, since parameters will be passed on Runtime and
coroutines are limited to a one-time execution. Coroutine functions
or standard asyncio awaitable functions are defined using the
async def syntax
Source code in openhivenpy\events.py
def set_awaitable(self, awaitable: Union[Awaitable, Callable]) -> None:
"""
Sets the coroutine of the event_listener, which will be called when
ws-events are received
:param awaitable: Awaitable (Coroutine Function). Can NOT be a
initialised coroutine, since parameters will be passed on Runtime and
coroutines are limited to a one-time execution. Coroutine functions
or standard asyncio awaitable functions are defined using the
`async def` syntax
"""
if inspect.iscoroutine(awaitable):
raise RuntimeError(
"Coroutine are not allowed as event_listener awaitable")
elif inspect.isawaitable(awaitable) or inspect.iscoroutinefunction(
awaitable):
self._awaitable = awaitable
else:
raise RuntimeError(f"Expected awaitable, but got {type(awaitable)}")
Important
Usage classes, which represent DispatchEventListener
openhivenpy.events.MultiDispatchEventListener
¶
EventListener Class that is used primarily for EventListeners that will be called multiple times
Methods¶
dispatch(self, *args, **kwargs)
async
¶
Dispatches the EventListener and calls a coroutine if one was passed. Does not raise exceptions but silences them!
:param args: Args that will be passed to the coroutine :param kwargs: Kwargs that will be passed to the coroutine
Source code in openhivenpy\events.py
async def dispatch(self, *args, **kwargs) -> None:
"""
Dispatches the EventListener and calls a coroutine if one was passed.
Does not raise exceptions but silences them!
:param args: Args that will be passed to the coroutine
:param kwargs: Kwargs that will be passed to the coroutine
"""
try:
await self.awaitable(*args, **kwargs)
except Exception as e:
utils.log_traceback(
brief=f"[EVENTS] Ignoring exception in {repr(self)}:",
exc_info=sys.exc_info()
)
raise RuntimeError(
f"Failed to execute assigned coroutine "
f"'{self.awaitable.__name__}'"
) from e
openhivenpy.events.SingleDispatchEventListener
¶
EventListener Class that will be called only once and will store the events data, args and kwargs
args: tuple
property
readonly
¶
dispatched: bool
property
readonly
¶
kwargs: dict
property
readonly
¶
Methods¶
__init__(self, client, event_name, awaitable)
special
¶
Source code in openhivenpy\events.py
def __init__(self, client: HivenClient, event_name: str,
awaitable: Union[Awaitable, Callable]):
self._dispatched = False
self._event_data = None
self._args = None
self._kwargs = None
super().__init__(client, event_name, awaitable)
__repr__(self)
special
¶
Source code in openhivenpy\events.py
def __repr__(self):
info = [
('event_name', getattr(self, 'event_name', None)),
('dispatched', self.dispatched),
('awaitable', getattr(self, 'awaitable', None))
]
return '<SingleDispatchEventListener {}>'.format(' '.join('%s=%s' % t for t in info))
dispatch(self, *args, **kwargs)
async
¶
Dispatches the EventListener and calls a coroutine if one was passed
:param args: Args that will be passed to the coroutine :param kwargs: Kwargs that will be passed to the coroutine :return: Returns the passed event_data
Source code in openhivenpy\events.py
async def dispatch(self, *args, **kwargs) -> None:
"""
Dispatches the EventListener and calls a coroutine if one was passed
:param args: Args that will be passed to the coroutine
:param kwargs: Kwargs that will be passed to the coroutine
:return: Returns the passed event_data
"""
try:
self._args = args
self._kwargs = kwargs
await self.awaitable(*args, **kwargs)
except Exception as e:
utils.log_traceback(
brief=f"[EVENTS] Ignoring exception in {repr(self)}:",
exc_info=sys.exc_info()
)
raise RuntimeError(f"Failed to execute assigned coroutine '{self.awaitable.__name__}'") from e
self._dispatched = True
self.client.remove_listener(self)
Important
The HivenEventHandler is already inherited in the standard client (HivenClient),
so all properties and methods can be used there as well.
openhivenpy.events.HivenEventHandler
¶
Events class used to register the main event listeners. Is inherited by the HivenClient for easier access.
active_listeners: Dict[str, List[DispatchEventListener]]
property
readonly
¶
available_events: List[str]
property
readonly
¶
non_buffer_events: List[str]
property
readonly
¶
Methods¶
__init__(self, client, parsers)
special
¶
Source code in openhivenpy\events.py
def __init__(self, client: HivenClient, parsers: HivenParsers):
self.parsers = parsers
self._client = client
self._active_listeners = {}
self._available_events = EVENTS
self._non_buffer_events = NON_BUFFER_EVENTS
# Searching through the HivenClient to find all async functions that
# were registered for event_listening Regular functions will NOT be
# registered and only if they are async! This will avoid that errors
# are thrown when trying to call the functions using 'await'
for listener in inspect.getmembers(
self, predicate=inspect.iscoroutinefunction
):
func_name = listener[0].replace('on_', '')
awaitable = listener[1]
if func_name in self._available_events:
self.add_multi_listener(func_name, awaitable)
logger.debug(f"[EVENTS] Event {listener[0]} registered")
add_listener(self, listener)
¶
Adds the listener to the client cache and will create a new list if the event_name does not exist yet!
:param listener: The Listener that will be added
Source code in openhivenpy\events.py
def add_listener(self, listener: DispatchEventListener) -> None:
"""
Adds the listener to the client cache and will create a new list if
the event_name does not exist yet!
:param listener: The Listener that will be added
"""
if self.active_listeners.get(listener.event_name):
self.active_listeners[listener.event_name].append(listener)
else:
self.active_listeners[listener.event_name] = [listener]
add_multi_listener(self, event_name, awaitable)
¶
Adds a new event listener to the list of active listeners
:param event_name: The key/name of the event the EventListener should be listening to :param awaitable: Coroutine that should be called when the EventListener was dispatched :return: The newly created EventListener
Source code in openhivenpy\events.py
def add_multi_listener(
self,
event_name: str,
awaitable: Union[Callable, Awaitable]
) -> MultiDispatchEventListener:
"""
Adds a new event listener to the list of active listeners
:param event_name: The key/name of the event the EventListener should
be listening to
:param awaitable: Coroutine that should be called when the
EventListener was dispatched
:return: The newly created EventListener
"""
event_name = event_name.replace('on_', '')
if event_name not in self.available_events:
raise UnknownEventError(
"The passed event type is invalid/does not exist"
)
if self._active_listeners.get(event_name) is None:
self._active_listeners[event_name] = []
return MultiDispatchEventListener(self._client, event_name, awaitable)
add_single_listener(self, event_name, awaitable)
¶
Adds a new single dispatch event listener to the list of active listeners
:param event_name: The key/name of the event the EventListener should be listening to :param awaitable: Coroutine that should be called when the EventListener was dispatched :return: The newly created EventListener
Source code in openhivenpy\events.py
def add_single_listener(
self,
event_name: str,
awaitable: Union[Callable, Coroutine]
) -> SingleDispatchEventListener:
"""
Adds a new single dispatch event listener to the list of active
listeners
:param event_name: The key/name of the event the EventListener should
be listening to
:param awaitable: Coroutine that should be called when the
EventListener was dispatched
:return: The newly created EventListener
"""
event_name = event_name.replace('on_', '')
if event_name not in self.available_events:
raise UnknownEventError(
"The passed event type is invalid/does not exist"
)
if self._active_listeners.get(event_name) is None:
self._active_listeners[event_name] = []
return SingleDispatchEventListener(self._client, event_name, awaitable)
call_listeners(self, event_name, args, kwargs)
async
¶
Dispatches all active EventListeners for the specified event. Does not call the parsers but the function directly and requires the args, kwargs passed
Will run all tasks before returning! Only supposed to be called in cases of special events!
:param event_name: The name of the event that should be triggered :param args: Args that will be passed to the coroutines :param kwargs: Kwargs that will be passed to the coroutines
Source code in openhivenpy\events.py
async def call_listeners(
self, event_name: str, args: tuple, kwargs: dict
) -> None:
"""
Dispatches all active EventListeners for the specified event.
Does not call the parsers but the function directly and requires the
args, kwargs passed
---
Will run all tasks before returning! Only supposed to be called in
cases of special events!
:param event_name: The name of the event that should be triggered
:param args: Args that will be passed to the coroutines
:param kwargs: Kwargs that will be passed to the coroutines
"""
listeners: List[DispatchEventListener] = self._active_listeners.get(
event_name.lower().replace('on_', '')
)
if listeners:
tasks = [listener(*args, **kwargs) for listener in listeners]
await asyncio.gather(*tasks)
cleanup_listeners(self)
¶
Cleanups the listeners and empties all active listeners
Source code in openhivenpy\events.py
def cleanup_listeners(self) -> None:
""" Cleanups the listeners and empties all active listeners """
self._active_listeners = {}
dispatch_event(self, event_name, args, kwargs)
¶
Manually adds an event to the event_buffer and triggers all listeners. Will return immediately and does not require asyncio unlike call_listeners which only calls the listeners;
Note that this will NOT call the event parsers but manually directly pass the args and kwargs to the functions!
:param event_name: The name of the event that should be triggered :param args: Args that will be passed to the coroutines :param kwargs: Kwargs that will be passed to the coroutines
Source code in openhivenpy\events.py
def dispatch_event(
self, event_name: str, args: tuple, kwargs: dict
) -> None:
"""
Manually adds an event to the event_buffer and triggers all listeners.
Will return immediately and does not require asyncio unlike
call_listeners which only calls the listeners;
Note that this will NOT call the event parsers but manually directly
pass the args and kwargs to the functions!
:param event_name: The name of the event that should be triggered
:param args: Args that will be passed to the coroutines
:param kwargs: Kwargs that will be passed to the coroutines
"""
self._validate_existence_of_event(event_name)
_: MessageBroker = getattr(self, 'message_broker')
buffer = _.get_buffer(event_name)
buffer.add_new_event(
{
"callee": event_name,
"caller": "client",
"time": time.time()
},
args,
kwargs
)
event(self, awaitable=None)
¶
Decorator used for registering Client Events
:param awaitable: Function that should be wrapped and registered
Source code in openhivenpy\events.py
def event(self, awaitable: Union[Callable, Coroutine] = None) -> Callable:
"""
Decorator used for registering Client Events
:param awaitable: Function that should be wrapped and registered
"""
def decorator(awaitable: Union[Callable, Coroutine]) -> Callable:
if not inspect.iscoroutinefunction(awaitable):
raise TypeError(
f"A coroutine was expected, got {type(awaitable)}"
)
self._validate_existence_of_event(awaitable.__name__)
func_name = awaitable.__name__.replace('on_', '')
self.add_multi_listener(func_name, awaitable)
logger.debug(f"[EVENTS] Event {func_name} registered")
# func can still be used normally outside the event listening
# process
return awaitable
if awaitable is None:
return decorator
else:
return decorator(awaitable)
remove_listener(self, listener)
¶
Removes the listener from the client cache
:param listener: The Listener that will be removed :raises KeyError: If the listener does not exist in the cache
Source code in openhivenpy\events.py
def remove_listener(self, listener: DispatchEventListener) -> None:
"""
Removes the listener from the client cache
:param listener: The Listener that will be removed
:raises KeyError: If the listener does not exist in the cache
"""
if self.active_listeners.get(listener.event_name):
self.active_listeners[listener.event_name].remove(listener)
else:
raise KeyError("The listener does not exist in the cache")
wait_for(self, event_name, awaitable=None)
async
¶
Waits for an event to be triggered and then returns the args and *kwargs passed
:param event_name: Name of the event to wait for :param awaitable: Coroutine that can be passed to be additionally triggered when received :raises UnknownEventError: If the event does not exist :return: A tuple of the args and kwargs => (args, kwargs)
Source code in openhivenpy\events.py
async def wait_for(
self,
event_name: str,
awaitable: Union[Callable, Coroutine, None] = None
) -> Tuple[tuple, dict]:
"""
Waits for an event to be triggered and then returns the *args and
**kwargs passed
:param event_name: Name of the event to wait for
:param awaitable: Coroutine that can be passed to be additionally
triggered when received
:raises UnknownEventError: If the event does not exist
:return: A tuple of the args and kwargs => (args, kwargs)
"""
event_name = event_name.replace('on_', '')
if event_name not in self.available_events:
raise UnknownEventError(
"The passed event type is invalid/does not exist"
)
listener = self.add_single_listener(event_name, awaitable)
while not listener.dispatched:
await asyncio.sleep(.05)
return listener.args, listener.kwargs