Overview

Queues are KubeMQ's reliable point-to-point pattern → see Queues for at-least-once delivery, settlement, and dead-letter routing at the broker level.

In kubemq-faststream, you register a queue consumer with the queues= keyword on @broker.subscriber(...) and send messages with broker.publish(..., queues=...). Settlement is governed by FastStream's AckPolicy, so the same acknowledgement model you use with other FastStream brokers applies here. This page documents the queue-specific API: ack policies, publish-side queue policies (TTL, delay, DLQ), batch send/receive, and out-of-band queue operations.

basic_send_receive.py

import asyncio
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.queues.basic"


@broker.subscriber(queues=CHANNEL)
async def handle_task(msg: dict) -> None:
    """Process a queue message (auto-acked on success)."""
    print(f"Received task: {msg}")
    print(f"Processing order {msg.get('order_id')}...")


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"order_id": 1001, "item": "laptop", "qty": 2},
        queues=CHANNEL,
    )
    print("Published order to queue")
    await asyncio.sleep(2)
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())

Queues connect over KubeMQ's native gRPC port 50000, the same transport as every other kubemq-faststream pattern. There is no HTTP connector to enable. If you don't have a broker running yet, start one with Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 9090 is the shared HTTP server used by the HTTP connectors (REST, CloudEvents, MCP, A2A); it is not required for FastStream but is harmless to expose. See the Getting Started guide for a full walkthrough.

Acknowledgement Policy

The default subscriber auto-acks a message after the handler returns successfully and nacks it (requeues for redelivery) if the handler raises an exception. Set the policy explicitly with ack_policy to choose different settlement behavior:

queues.py

from faststream.middlewares import AckPolicy
from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(queues="tasks", ack_policy=AckPolicy.ACK)
async def process_task(task: dict) -> None:
    print(f"Processing: {task}")
    # Message is auto-acked on success, nacked on exception

AckPolicy is FastStream's own enum. You can import it from faststream.middlewares or, for convenience, directly from the package — kubemq_faststream re-exports it (from kubemq_faststream import AckPolicy, also available as from kubemq_faststream.schemas import AckPolicy).

AckPolicy	Behavior
`AckPolicy.ACK`	Ack on success, nack (requeue) on handler error. Default.
`AckPolicy.NACK_ON_ERROR`	Nack (requeue) on handler error.
`AckPolicy.REJECT_ON_ERROR`	Reject on handler error (server-configured disposition).
`AckPolicy.ACK_FIRST`	Ack immediately before the handler runs (at-most-once).
`AckPolicy.MANUAL`	No automatic ack — the handler calls `msg.ack()` / `msg.nack()` itself.

The example below registers a handler per policy on its own channel so you can see each settlement mode side by side. Note the MANUAL handler pulls the underlying message off the FastStream context and settles it explicitly:

ack_nack_reject.py

from faststream import FastStream, context
from kubemq_faststream import KubeMQBroker
from kubemq_faststream.schemas import AckPolicy

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)


@broker.subscriber(queues="example.queues.ack_auto", ack_policy=AckPolicy.ACK)
async def auto_ack_handler(msg: dict) -> None:
    """Auto-acknowledged after handler returns without error."""
    print(f"[ACK] Processed: {msg}")


@broker.subscriber(queues="example.queues.ack_nack", ack_policy=AckPolicy.NACK_ON_ERROR)
async def nack_on_error_handler(msg: dict) -> None:
    """Message is requeued if handler raises an exception."""
    if msg.get("fail"):
        raise ValueError("Simulated failure — message will be requeued")
    print(f"[NACK_ON_ERROR] Success: {msg}")


@broker.subscriber(queues="example.queues.ack_reject", ack_policy=AckPolicy.REJECT_ON_ERROR)
async def reject_on_error_handler(msg: dict) -> None:
    """Message is rejected (not requeued) if handler raises an exception."""
    if msg.get("fail"):
        raise ValueError("Simulated failure — message will be rejected")
    print(f"[REJECT_ON_ERROR] Success: {msg}")


@broker.subscriber(queues="example.queues.ack_manual", ack_policy=AckPolicy.MANUAL)
async def manual_handler(msg: dict) -> None:
    """Explicit ack/nack/reject — full control over message disposition."""
    message = context.get("message")
    if msg.get("action") == "reject":
        await message.reject()
    elif msg.get("action") == "nack":
        await message.nack()
    else:
        await message.ack()

AckPolicy.ACK_FIRST acks the message before the handler runs, giving at-most-once delivery — if the handler then fails, the message is not requeued. Use it only when reprocessing is more harmful than dropping a message.

ack_first_policy.py

from kubemq_faststream import AckPolicy, KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(queues="example.queues.ack_first", ack_policy=AckPolicy.ACK_FIRST)
async def handle_task(msg: dict) -> None:
    # The message was already acked before this runs. A failure here
    # does NOT requeue the message.
    print(f"[ACK_FIRST] Processing: {msg}")

Ack/nack operations are no-ops for the Events and Events Store patterns, which are fire-and-forget. AckPolicy is meaningful only for Queues.

Publishing Messages

Send a single message with broker.publish(..., queues=...). To send several messages in one round-trip, use broker.publish_batch(*messages, queues=...) — the messages are passed positionally and an optional headers dict is applied to the batch.

queues_batch_send.py

# Publish a single message
await broker.publish({"type": "email"}, queues="tasks")

# Publish a batch in a single call
messages = [{"item": i, "batch": True} for i in range(10)]
await broker.publish_batch(
    *messages,
    queues="tasks",
    headers={"source": "batch-send-demo"},
)

Or pass the messages inline, exactly as shown in the README quick start:

queues.py

await broker.publish_batch(
    {"type": "email"},
    {"type": "sms"},
    {"type": "push"},
    queues="tasks",
)

Queue Policies

broker.publish(...) accepts queue-specific delivery policies as keyword arguments. They are ignored by the other patterns and apply only when you publish with queues=.

Message TTL (Expiration)

expiration_in_seconds sets a time-to-live on the message. If it is not consumed within that window, the broker discards it. This is useful for time-sensitive work that becomes meaningless once stale.

expiration_policy.py

# Discarded if not consumed within 5 seconds
await broker.publish(
    {"id": 1, "note": "expires in 5 seconds"},
    queues="example.queues.expiration",
    expiration_in_seconds=5,
)

# No expiration — stays in the queue until consumed
await broker.publish(
    {"id": 2, "note": "no expiration"},
    queues="example.queues.expiration",
)

Delayed Delivery

delay_in_seconds holds the message invisible to consumers until the delay elapses. Use it to schedule future work, defer processing, or implement a back-off between retries.

delay_policy.py

# Visible immediately
await broker.publish(
    {"id": 1, "note": "immediate delivery"},
    queues="example.queues.delay",
)

# Becomes visible to consumers only after 5 seconds
await broker.publish(
    {"id": 3, "note": "delayed by 5 seconds"},
    queues="example.queues.delay",
    delay_in_seconds=5,
)

Dead-Letter Queues

A poison message that repeatedly fails processing should not block the queue forever. Set max_receive_count to cap how many times a message may be delivered without being acked, and max_receive_queue to name the dead-letter queue (DLQ) the broker routes it to once that cap is exceeded. A separate subscriber on the DLQ channel then captures the failed messages for inspection or replay.

Both fields must be set together — passing only max_receive_count or only max_receive_queue has no effect; the broker needs both the cap and a destination before it will divert a message.

max_receive_dlq.py

from faststream import FastStream
from kubemq_faststream import KubeMQBroker
from kubemq_faststream.schemas import AckPolicy

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

MAIN_QUEUE = "example.queues.dlq_main"
DLQ_QUEUE = "example.queues.dlq_dead"


@broker.subscriber(queues=MAIN_QUEUE, ack_policy=AckPolicy.NACK_ON_ERROR)
async def handle_order(msg: dict) -> None:
    """Fails every time, so the message is redelivered until it hits the DLQ."""
    print(f"[MAIN] Attempt to process: {msg}")
    raise ValueError("Simulated processing failure")


@broker.subscriber(queues=DLQ_QUEUE)
async def handle_dead_letter(msg: dict) -> None:
    """Capture messages that exceeded their max receive count."""
    print(f"[DLQ] Dead-letter received: {msg}")


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"order_id": 999, "item": "fragile-widget"},
        queues=MAIN_QUEUE,
        max_receive_count=3,
        max_receive_queue=DLQ_QUEUE,
    )
    print(f"Published order with max_receive_count=3, DLQ={DLQ_QUEUE}")

With max_receive_count=3, the broker delivers the message to handle_order three times. Because the handler keeps raising (and NACK_ON_ERROR requeues it each time), on the fourth delivery the broker instead routes the message to example.queues.dlq_dead, where handle_dead_letter picks it up.

The DLQ pattern is the foundation for production resilience workflows such as retry-with-backoff and saga compensation. The resilient pipelines scenario builds on dead_letter_processing.py and related examples to show full pipelines.

Batch Receive

A batch subscriber receives up to max_messages messages at once as a list, rather than one message per handler invocation. Set batch=True and tune max_messages and wait_timeout. Internally this constructs a BatchQueuesSubscriber. Batch handlers settle the whole batch together — on success every message is acked; if the handler raises, none are acked and they become available for redelivery.

batch_receive.py

@broker.subscriber(
    queues="example.queues.batch_receive",
    batch=True,
    max_messages=5,
    wait_timeout=10,
)
async def handle_batch(msgs: list[dict]) -> None:
    """Process a batch of queue messages."""
    print(f"[BATCH] Received {len(msgs)} messages: {msgs}")

For finer control, combine batch=True with ack_policy=AckPolicy.MANUAL and settle the batch explicitly from the FastStream context:

queues_batch_receive.py

from faststream import context
from kubemq_faststream import AckPolicy

@broker.subscriber(
    queues="example.batch.queues_recv_manual",
    batch=True,
    max_messages=5,
    wait_timeout=10,
    ack_policy=AckPolicy.MANUAL,
)
async def handle_batch_manual(msgs: list[dict]) -> None:
    message = context.get("message")
    for msg in msgs:
        ...  # process each message
    await message.ack()  # acknowledge the whole batch

Inspecting and Clearing a Queue

Two broker methods operate on a queue out-of-band, without a subscriber:

broker.peek_queue_messages(queues=..., max_messages=...) — read up to max_messages messages without consuming them. The messages remain in the queue for a real subscriber to process later. Useful for monitoring, debugging, or building admin tooling.
broker.ack_all_queue_messages(queues=...) — acknowledge every pending message in a queue at once, clearing it. Useful for draining a backlog of messages that are already processed or no longer relevant.

peek_messages.py

# Inspect up to 3 messages — they stay in the queue
result = await broker.peek_queue_messages(
    queues="example.queues.peek",
    max_messages=3,
)
print(f"Peeked messages: {result}")

ack_all.py

# Drain the queue — acknowledge everything pending at once
await broker.ack_all_queue_messages(queues="example.queues.ack_all")
print("Acknowledged all messages in queue")

Subscriber Tuning

Two subscriber configuration fields control how a queue consumer polls the broker. They are passed as keyword arguments to @broker.subscriber(...) and back the KubeMQSubscriberConfig fields.

Prop

Type

For strictly ordered processing, keep a single consumer with max_messages=1 so messages are handled one at a time. To raise throughput, increase max_messages (with batch=True) so each poll returns a batch.

Queues (Point-to-Point)