pak.packets.packet_handler

Tools for handling Packets.

packet_listener(*packet_types, **flags)

A decorator for Packet listeners.

See also

PacketHandler.register_packet_listener()

Parameters:

• *packet_types (subclass of Packet) – The Packets to listen for.

• **flags – The flags which must match for the listener to be returned by PacketHandler.listeners_for_packet().

Examples

>>> import pak
>>> class Example(pak.PacketHandler):
...     @pak.packet_listener(pak.Packet)
...     def listener_example(self, packet):
...         # Do things with 'packet' here.
...         pass
...
>>> ex = Example()
>>> ex.is_listener_registered(ex.listener_example)
True

most_derived_packet_listener(most_general_packet_type, **flags)

A decorator for Packet listeners that dispatch only to the listener corresponding to the most derived Packet type.

Parameters:

• most_general_packet_type (subclass of Packet) – The most general Packet for the listener, i.e. the base class for all the Packets the listener will handle.

• **flags – See PacketHandler.listeners_for_packet().

Examples

>>> import pak
>>> class GeneralPacket(pak.Packet):
...     pass
...
>>> class DerivedPacket(GeneralPacket):
...     pass
...
>>> class MoreDerivedPacket(DerivedPacket):
...     pass
...
>>> class MyHandler(pak.PacketHandler):
...     @pak.most_derived_packet_listener(GeneralPacket)
...     def most_derived(self):
...         return "general"
...
...     @most_derived.derived_listener(DerivedPacket)
...     def most_derived(self):
...         return "derived"
...
...     @most_derived.derived_listener(MoreDerivedPacket)
...     def most_derived(self):
...         return "more derived"
...
>>> handler = MyHandler()
>>> handler.listeners_for_packet(GeneralPacket())[0]()
'general'
>>> handler.listeners_for_packet(DerivedPacket())[0]()
'derived'
>>> handler.listeners_for_packet(MoreDerivedPacket())[0]()
'more derived'
>>> # You can pass 'override=True' to 'derived_listener'
>>> # to override a previously set listener.
>>> class MyDerivedHandler(MyHandler):
...     @MyHandler.most_derived.derived_listener(MoreDerivedPacket, override=True)
...     def most_derived(self):
...         return "overridden more derived"
...
>>> handler = MyDerivedHandler()
>>> handler.listeners_for_packet(MoreDerivedPacket())[0]()
'overridden more derived'

class PacketHandler

Bases: object

An object which handles Packets by dispatching them to listener methods or functions.

On construction, methods decorated with packet_listener() are passed to register_packet_listener().

has_packet_listener(packet, **flags)

Gets whether there is a listener for a certain Packet.

See also

listeners_for_packet()

Parameters:

• packet (Packet or subclass of Packet) – The Packet to check listeners for.

• **flags – The flags which must match the flags a listener was registered with for it to be an applicable listener.

Returns:

Whether there is a listener for packet.

Return type:

bool

Examples

>>> import pak
>>> class MyPacket(pak.Packet):
...     pass
...
>>> class Example(pak.PacketHandler):
...     @pak.packet_listener(MyPacket)
...     def listener_example(self, packet):
...         pass
...
>>> ex = Example()
>>> ex.has_packet_listener(MyPacket())
True
>>> ex.has_packet_listener(pak.Packet())
False
>>>
>>> # You can also get whether a 'Packet' class has a listener:
>>> ex.has_packet_listener(MyPacket)
True
>>> ex.has_packet_listener(pak.Packet)
False

is_listener_registered(listener)

Gets whether a Packet listener is registered.

Parameters:

listener – The Packet listener possibly passed to register_packet_listener().

Returns:

Whether listener is a registered Packet listener.

Return type:

bool

listeners_for_packet(packet, **flags)

Gets the listeners for a certain Packet.

It is the caller’s responsibility to send the Packet to the returned listeners.

If a Packet listener has a to_real_listener attribute, then that attribute will be called with the packet parameter to get the real Packet listener to be returned.

Parameters:

• packet (Packet or subclass of Packet) – The Packet to get listeners for.

• **flags – The flags which must match the flags a listener was registered with for the listener to be returned.

Returns:

The list of listeners for packet.

Return type:

list

Examples

>>> import pak
>>> class MyPacket(pak.Packet):
...     pass
...
>>> class Example(pak.PacketHandler):
...     @pak.packet_listener(MyPacket)
...     def listener_example(self, packet):
...         # Do things with 'packet' here.
...         pass
...     def __repr__(self):
...         return "Example()"
...
>>> ex = Example()
>>> ex.listeners_for_packet(MyPacket())
[<bound method Example.listener_example of Example()>]
>>>
>>> # You can also request listeners for 'Packet' classes:
>>> ex.listeners_for_packet(MyPacket)
[<bound method Example.listener_example of Example()>]

register_packet_listener(listener, *packet_types, **flags)

Registers a Packet listener.

See also

listeners_for_packet()

Parameters:

• listener – The Packet listener to register.

• *packet_types (subclass of Packet) – The Packets to listen for.

• **flags – The flags which must match for the listener to be returned by PacketHandler.listeners_for_packet().

Raises:

• TypeError – If *packet_types is empty.

• ValueError – If listener is already registered.

Examples

>>> import pak
>>> def listener_example(packet):
...     # Do things with 'packet' here.
...     pass
...
>>> handler = pak.PacketHandler()
>>> handler.register_packet_listener(listener_example, pak.Packet)
>>> handler.is_listener_registered(listener_example)
True

unregister_packet_listener(listener)

Unregisters a Packet listener.

Parameters:

listener – The Packet listener passed to register_packet_listener().

class AsyncPacketHandler

Bases: PacketHandler

A PacketHandler that handles Packets asynchronously.

This class doesn’t really have different semantics from PacketHandler, but it has extra facilities for asynchronously handling Packets.

async end_listener_tasks(*, timeout=1)

Ends any outstanding listener tasks created with listener_task_group().

Parameters:

timeout (int or float or None) –

How long to wait before canceling outstanding listener tasks.

Passed to asyncio.wait_for().

listener_task_group(*, listen_sequentially)

Creates an asynchronous context manager in which listener tasks should be created.

The manager has a create_task method that takes a coroutine, like so:

handler = pak.AsyncPacketHandler()
packet  = ...

async with handler.listener_task_group(listen_sequentially=False) as group:
    for listener in handler.listeners_for_packet(packet):
        group.create_task(packet)

await handler.end_listener_tasks()

Note

This interface is similar to the asyncio.TaskGroup class in Python 3.11+.

Warning

The end_listener_tasks() method should be called when all listening should end.

Parameters:

listen_sequentially (bool) –

Whether the listeners should be called sequentially.

If True, listeners responding to the same Packet will still be run asynchronously, however they will all be awaited before listening to another Packet.

Also when True, the tasks are never canceled.

Examples

>>> import pak
>>> import asyncio
>>> class ExampleHandler(pak.AsyncPacketHandler):
...     @pak.packet_listener(pak.Packet)
...     async def slow_listener(self, packet):
...         await asyncio.sleep(1)
...         print("slow_listener")
...
...     @pak.packet_listener(pak.Packet)
...     async def fast_listener(self, packet):
...         print("fast_listener")
...
>>> async def main():
...     handler = ExampleHandler()
...     packet  = pak.Packet()
...     async with handler.listener_task_group(listen_sequentially=False) as group:
...         for listener in handler.listeners_for_packet(packet):
...             group.create_task(listener(packet))
...
...     await handler.end_listener_tasks(timeout=2)
...
>>> asyncio.run(main())
fast_listener
slow_listener