Consuming Acknowledgements#

As you may know, Nats employs a rather extensive Acknowledgement policy.

In most cases, FastStream automatically acknowledges (acks) messages on your behalf. When your function executes correctly, including sending all responses, a message will be acknowledged (and rejected in case of an exception).

However, there are situations where you might want to use different acknowledgement logic.

Retries#

If you prefer to use a nack instead of a reject when there's an error in message processing, you can specify the retry flag in the @broker.subscriber(...) method, which is responsible for error handling logic.

By default, this flag is set to False, indicating that if an error occurs during message processing, the message can still be retrieved from the queue:

@broker.subscriber("test", retry=False) # don't handle exceptions
async def base_handler(body: str):
    ...

If this flag is set to True, the message will be nacked and placed back in the queue each time an error occurs. In this scenario, the message can be processed by another consumer (if there are several of them) or by the same one:

@broker.subscriber("test", retry=True)  # try again indefinitely
async def base_handler(body: str):
    ...

Manual Acknowledgement#

If you want to acknowledge a message manually, you can get access directly to the message object via the Context and call the method.

from faststream.nats.annotations import NatsMessage

@broker.subscriber("test")
async def base_handler(body: str, msg: NatsMessage):
    await msg.ack()
    # or
    await msg.nack()
    # or
    await msg.reject()

FastStream will see that the message was already acknowledged and will do nothing at the end of the process.

Interrupt Process#

If you want to interrupt message processing at any call stack, you can raise faststream.exceptions.AckMessage

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from faststream import FastStream
from faststream.exceptions import AckMessage
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")
app = FastStream(broker)


@broker.subscriber("test-subject", stream="test-stream")
async def handle(body):
    smth_processing(body)


def smth_processing(body):
    if True:
        raise AckMessage()


@app.after_startup
async def test_publishing():
    await broker.publish("Hello!", "test-subject")

This way, FastStream interrupts the current message processing and acknowledges it immediately. Also, you can raise NackMessage and RejectMessage too.