warren.pubsub.memory
In-process pubsub backend: one broker, one event loop, no infrastructure.
Selected with backend: memory in RuntimeConfig. Single-process and
non-durable by construction. See warren/docs/memory.md.
Classes
Delivery
dataclassDelivery(body, routing_key)One message as a consumer receives it.
parambodybytesSerialised message. Bytes, not a dict, so every bound queue gets an independent copy once decoded.
paramrouting_keystrThe key the message was published with.
MemoryBroker
classMemoryBroker()Routes messages to bound queues in FIFO order; priority is ignored.
methodbind(exchange, queue_name, binding_key=None) -> asyncio.Queue[Delivery]Bind queue_name to exchange and return the queue.
Idempotent. Binding an existing queue name returns the same queue, so several consumers of one worker type compete for its messages.
paramexchangeRMQExchangeConfigparamqueue_namestrparambinding_keystr | None= Nonemethodpublish(exchange, routing_key, body) -> intDeliver body to every matching queue; return how many.
paramexchangeRMQExchangeConfigparamrouting_keystrparambodybytesMemoryConnectionManager
classMemoryConnectionManager(*, name=None)Holds the one MemoryBroker every publisher and consumer shares.
Exists so the memory backend has the same setup() / teardown()
shape as the RabbitMQ and Kafka connection managers and can sit in
RuntimeInfra.pubsub_connection_manager. Sharing a broker means
sharing this object: build one RuntimeInfra per process and inject
it into every runner.
paramnamestr | None= NoneattributebrokerMemoryBrokermethodsetupasync () -> Nonemethodteardownasync () -> NoneMemoryConsumerConfig
modelMemoryConsumerConfigHandler limits for an in-process consumer manager.
attributeconcurrencyint | None= NoneMaximum number of handlers that run at the same time. None means 1.
attributehandler_timeout_secondsfloat | None= NoneTime limit for one handler call. A handler that times out is a soft failure. /live fails when a handler runs for twice this value. None means no limit.
attributeon_shutdown_timeoutfloat= 30.0Seconds to wait for in-flight work on stop.
MemoryConsumerManager
classMemoryConsumerManager(connection_manager, consumer, *, exchange, queue_name, binding_key=None, binding_keys=(), data_publisher=None, control_publisher=None, observer_publisher=None, retry_config=None, extract_identity_func=None, publish_hard_failures=True, handler_timeout_seconds=None, concurrency=None, on_shutdown_timeout=30.0)Consumer manager for the in-process backend.
Bridges a MemoryBroker queue and an application worker. The worker
receives plain dicts and returns optional dicts, exactly as on the
RabbitMQ and Kafka backends.
Settlement semantics, the in-process mapping of ack/nack/reject:
- ack and reject(requeue=False): the delivery is simply not put back.
- nack(requeue=True): the delivery goes back on the queue.
Handler concurrency defaults to one per manager.
Single-process only. The broker lives in this process's memory, and
RetryWorker schedules retries with in-process timers, so nothing
survives a restart and nothing crosses a process boundary. This is the
only run mode the memory backend supports.
paramconnection_managerMemoryConnectionManagerparamconsumerMessageConsumerInterfaceparamexchangeRMQExchangeConfigparamqueue_namestrparambinding_keystr | None= Noneparambinding_keystuple[str, ...]= ()paramdata_publisherPublisherInterface | None= Noneparamcontrol_publisherPublisherInterface | None= Noneparamobserver_publisherPublisherInterface | None= Noneparamretry_configRetryConfig | None= Noneparamextract_identity_funcExtractMessageIdentityFunc | None= Noneparampublish_hard_failuresbool= Trueparamhandler_timeout_secondsfloat | None= Noneparamconcurrencyint | None= Noneparamon_shutdown_timeoutfloat= 30.0methodsetupasync () -> NoneSet up publishers and bind the queue.
Binding happens here, not in start_consuming(), so that every
runner in the process is bound before any of them publishes.
methodstart_consumingasync () -> Nonemethodstop_consumingasync () -> NoneStop taking messages, drain in-flight work, tear down publishers.
methodhealthasync (*, probe_timeout=1.0) -> ConsumerHealthNo connection to lose and no blocked state; liveness is the loop.
paramprobe_timeoutfloat= 1.0MemoryPublisher
classMemoryPublisher(connection_manager, exchange, *, route=None, route_func=None, name=None)Publishes JSON-encoded messages onto a MemoryBroker exchange.
Route resolution matches RMQPublisher: route_func when set,
otherwise the static route, otherwise the empty key.
Messages are encoded to JSON even though nothing leaves the process. That keeps two properties of the real backends: a payload that cannot be serialised fails at publish time, and every consumer decodes its own copy, which matters because consumer managers mutate the body when they stamp retry state.
paramconnection_managerMemoryConnectionManagerparamexchangeRMQExchangeConfigparamrouteRoute | None= Noneparamroute_funcRouteFunc | None= Noneparamnamestr | None= Nonemethodsetupasync () -> NoneNothing to open; the broker is owned by the connection manager.
methodteardownasync () -> NoneNothing to close.
method__call__async (message) -> NoneparammessagedictFunctions
topic_matches
functiontopic_matches(pattern, key) -> boolAMQP topic match: * is exactly one word, # is zero or more.
parampatternstrparamkeystr