Pull Subscriber#

Overview#

NATS JetStream supports two different ways to consume messages: Push and Pull consumers.

The Push consumer is used by default to consume messages with FastStream. It means that the NATS server delivers messages to your consumer as fast as possible by itself. However, it also means that NATS should control all current consumer connections and increase server load.

Thus, the Pull consumer is the recommended way to consume JetStream messages by the NATS TEAM. Using it, you simply ask NATS for new messages at some interval. It may sound a little less convenient than automatic message delivery, but it provides several advantages, such as:

• Consumer scaling without a queue group
• Handling messages in batches
• Reducing NATS server load

So, if you want to consume a large flow of messages without strict time limitations, the Pull consumer is the right choice for you.

FastStream Details#

The Pull consumer is just a regular Stream consumer, but with the pull_sub argument, which controls consuming messages with batch size and block interval.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from faststream import FastStream, Logger
from faststream.nats import NatsBroker, PullSub

broker = NatsBroker()
app = FastStream(broker)


@broker.subscriber(
    subject="test",
    stream="stream",
    pull_sub=PullSub(batch_size=10),
)
async def handle(msg, logger: Logger):
    logger.info(msg)

The batch size doesn't mean that your msg argument is a list of messages, but it means that you consume up to 10 messages for one request to NATS and call your handler for each message concurrently in an anyio task group.

Batch Pull Subscriber Example#

This example demonstrates how to use batch=True with a Pull subscriber. The message bodies are passed as a list into bodies, while the native NATS messages are available via msg.raw_message (type NatsBatchMessage), allowing per‑message ack() inside a batch.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from typing import Annotated

from faststream import Context, FastStream, Logger
from faststream.nats import NatsBroker, PullSub, message

broker = NatsBroker()

@broker.subscriber(
    "test",
    stream="test",
    durable="test",
    pull_sub=PullSub(batch=True, batch_size=3),
)
async def handler(
    bodies: list[str],
    logger: Logger,
    msg: Annotated[message.NatsBatchMessage, Context("message")],
) -> None:
    logger.info(bodies)
    for m in msg.raw_message:
        await m.ack()

app = FastStream(broker)

@app.after_startup
async def after_startup() -> None:
    for i in range(3):
        await broker.publish(f"Hello, world! {i}", "test", stream="test")

So, your subject will be processed much faster, without blocking for each message processing.

With batch=True, FastStream collects up to batch_size messages before calling the handler. It waits until the batch is full or timeout expires (5 seconds by default), then delivers any collected messages as a partial batch. Set timeout=None to wait until the batch is full.