# Queues (Point-to-Point) (/integrations/faststream/how-to/queues)



## Overview [#overview]

Queues are KubeMQ's reliable point-to-point pattern → see [Queues](/learn/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.

```python title="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())
```

<Callout type="info">
  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:

  <RunKubeMQ ports="[50000, 9090]" />

  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](/integrations/faststream/tutorials/getting-started) for a full walkthrough.
</Callout>

## Acknowledgement Policy [#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:

```python title="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. &#x2A;*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:

```python title="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()
```

<Callout type="info">
  `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.

  ```python title="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.
</Callout>

## Publishing Messages [#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.

```python title="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:

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

## Queue Policies [#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) [#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.

```python title="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 [#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.

```python title="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 [#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.

```python title="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.

<Callout type="info">
  The DLQ pattern is the foundation for production resilience workflows such as retry-with-backoff and saga compensation. The [resilient pipelines scenario](/integrations/faststream/scenarios/resilient-pipelines) builds on `dead_letter_processing.py` and related examples to show full pipelines.
</Callout>

## Batch Receive [#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.

```python title="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`:

```python title="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 [#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.

```python title="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}")
```

```python title="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 [#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.

<TypeTable
  type="{
  max_messages: {
    description: 'Maximum number of messages to receive per poll. With batch=True this is the batch size; the batch handler receives up to this many messages as a list.',
    type: 'int',
    default: '1',
  },
  wait_timeout: {
    description: 'How long, in seconds, a single poll waits for messages before returning.',
    type: 'int',
    default: '60',
  },
}"
/>

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.

## Related [#related]

<Cards>
  <Card title="Queues (core concept)" href="/learn/queues" description="At-least-once delivery, settlement, and dead-letter routing at the broker level." />

  <Card title="Composition" href="/integrations/faststream/how-to/composition" description="Group queue handlers into KubeMQRouter prefixes and auto-publish handler results with @broker.publisher." />

  <Card title="Resilient Pipelines" href="/integrations/faststream/scenarios/resilient-pipelines" description="Production resilience: dead-letter processing, retry with backoff, and saga compensation built on queues." />
</Cards>