berconpy.EventDispatcher

class berconpy.EventDispatcher

Bases: object

Provides an event handler system for use with asyncio.

To listen for one of the following events, use the add_listener() method, the listen() decorator, or the type-safe on_* decorators:

dispatch = EventDispatcher()
dispatch.add_listener("on_login", event)

@dispatch.listen()
async def on_login():
    ...

@dispatch.listen("on_login")
async def my_listener():
    ...

@dispatch.on_login
async def my_listener():
    ...

Methods

add_listener(event, func)	Adds a listener for a given event, e.g. "on_login".
listen([event])	A decorator shorthand to add a listener for a given event, e.g. "on_login".
remove_listener(event, func)	Removes a listener from a given event, e.g. "on_login".
wait_for(event, *[, check, timeout])	Waits for a specific event to occur and returns the result.

Attributes

on_command(response, /)	Fired after receiving any command response from the server.
on_login()	Fired after a successful login to the server.
on_message(message, /)	Fired for messages sent by the server, e.g. player connections.
on_raw_event(packet, /)	Fired for every parsable packet received by the server.

add_listener(event, func)

Adds a listener for a given event, e.g. "on_login".

Parameters:

• event (str) – The event to listen for.

• func (Callable[..., Union[Any, Awaitable[Any]]]) – The function to dispatch when the event is received.

listen(event=None)

A decorator shorthand to add a listener for a given event, e.g. "on_login".

Parameters:

event (Optional[str]) – The event to listen for. If None, the function name is used as the event name.

Return type:

Callable[[TypeVar(T)], TypeVar(T)]

remove_listener(event, func)

Removes a listener from a given event, e.g. "on_login".

This method should be a no-op if the given event and function does not match any registered listener.

Parameters:

• event (str) – The event used by the listener.

• func (Callable[..., Union[Any, Awaitable[Any]]]) – The function used by the listener.

async wait_for(event, *, check=None, timeout=None)

Waits for a specific event to occur and returns the result.

This allows handling one-shot events in a simpler manner than with persistent listeners.

Parameters:

• event (str) – The event to wait for, e.g. "login" and "on_login".

• check (Optional[Callable[..., Union[Any, Awaitable[Any]]]]) – An optional predicate function to use as a filter. This can be either a regular or an asynchronous function. The function should accept the same arguments that the event normally takes.

• timeout (Union[float, int, None]) – An optional timeout for the function. If this is provided and the function times out, an asyncio.TimeoutError is raised.

Returns:

The same arguments received in the event.

Raises:

asyncio.TimeoutError – The timeout was exceeded while waiting.

on_command(response, /)

Fired after receiving any command response from the server.

This should only be used for debugging purposes as the send_command() method already returns the server’s response.

Parameters:

response (str) – The response received by the server.

Return type:

Any

on_login

Fired after a successful login to the server.

Return type:

Any

on_message(message, /)

Fired for messages sent by the server, e.g. player connections.

More specific events such as on_admin_login() are dispatched from this event.

Parameters:

response – The message that was sent by the server.

Return type:

Any

on_raw_event(packet, /)

Fired for every parsable packet received by the server.

Parameters:

packet (ServerPacket) – The packet that was received.

Return type:

Any