Api

Core Module

Core dispatch mechanism.

Dispatcher(maxsize=0)

Message dispatcher.

The dispatcher carries out the following steps:

• Receive and enqueue messages.
• Dequeue one message.
• Determine react method.
• Collect messages from running react method.
• Repeat or exit.

Parameters:

• maxsize (int, default: 0 ) –

  passed to queue

Initialize the dispatcher.

Source code in rsvp/core.py

def __init__(self, /, maxsize: int = 0) -> None:
   """Initialize the dispatcher."""
   self._message_queue: asyncio.Queue[Message] = asyncio.Queue(maxsize=maxsize)
   self._loop_task: asyncio.Task | None = None

dispatch(message) async

Dispatch a single message.

Parameters:

• message (Message) –

  the message for dispatch.

Source code in rsvp/core.py

async def dispatch(self, message: Message) -> None:
   """Dispatch a single message.

   :param message: the message for dispatch.
   """
   match message:
      case [receiver, *args]:
         await self._dispatch(receiver, args)
      case _:
         await self._dispatch(message, ())

dispose(receiver, args) async

Dispose a message without react method.

Feel free to override this method in order to change behavior.

Parameters:

• receiver (object) –

  the receiver of the message

• args (Sequence[object]) –

  the message arguments.

Source code in rsvp/core.py

async def dispose(
   self,
   receiver: object,
   args: Sequence[object],
) -> None:
   """Dispose a message without ``react`` method.

   Feel free to override this method in order
   to change behavior.

   :param receiver: the receiver of the message
   :param args: the message arguments.
   """
   pass

receive(message) async

Receive and enqueue a single message.

Parameters:

• message (Message) –

  the received message.

Source code in rsvp/core.py

async def receive(self, message: Message) -> None:
   """Receive and enqueue a single message.

   :param message: the received message.
   """
   await self._message_queue.put(message)

start()

Start the dispatcher loop as a new task.

If this dispatcher has already a loop running this method does nothing.

Returns:

• bool –

  whether a new loop was started

Source code in rsvp/core.py

def start(self) -> bool:
   """Start the dispatcher loop as a new task.

   If this dispatcher has already a loop running
   this method does nothing.

   :return: whether a new loop was started
   """
   if self._loop_task is not None:
      return False
   coro = self._loop()
   self._loop_task = asyncio.create_task(coro)
   return True

handler: python