class litestar.channels.backends.memory.MemoryChannelsBackend#

Bases: ChannelsBackend

An in-memory channels backend

__init__(history: int = 0) None#
async on_startup() None#

Called by the plugin on application startup

async on_shutdown() None#

Called by the plugin on application shutdown

async publish(data: bytes, channels: Iterable[str]) None#

Publish data to channels. If a channel has not yet been subscribed to, this will be a no-op.

  • data – Data to publish

  • channels – Channels to publish to




RuntimeError – If on_startup has not been called yet

async subscribe(channels: Iterable[str]) None#

Subscribe to channels, and enable publishing to them

async unsubscribe(channels: Iterable[str]) None#

Unsubscribe from channels

async stream_events() AsyncGenerator[tuple[str, Any], None]#

Return a generator, iterating over events of subscribed channels as they become available

async get_history(channel: str, limit: int | None = None) list[bytes]#

Return the event history of channel, at most limit entries