Stream Acknowledgement#

When working with Redis streams in the FastStream library, it's important to manage message acknowledgements carefully to ensure that messages are not lost and that they have been processed as intended.

By default, when using FastStream with a Redis stream consumer group, the library will automatically acknowledge (ack) a message once it has been processed successfully. This follows the at least once processing guarantee.

Manual Acknowledgement#

In cases where you want explicit control over when a message is acknowledged, you can manually acknowledge a message by accessing the ack and nack methods provided:

from faststream.redis.annotations import RedisStreamMessage, Redis

# Setup broker and faststream app
...

@broker.subscriber(StreamSub("test-stream", group="test-group", consumer="1"))
async def base_handler(body: dict, msg: RedisStreamMessage, redis: Redis):
    # Process the message
    ...

    # Manually acknowledge the message
    await msg.ack(redis, group="test-group")
    # or, if processing fails and you want to reprocess later
    await msg.nack()

Using ack will mark the message as processed in the stream, while nack is useful for situations where you might need to reprocess a message due to a handling failure.

Interrupt Process#

If the need arises to instantly interrupt message processing at any point in the call stack and acknowledge the message, you can achieve this by raising the faststream.exceptions.AckMessage exception:

 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.redis import RedisBroker, StreamSub

broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)


@broker.subscriber(stream=StreamSub("test-stream", group="test-group", consumer="1"))
async def handle(body):
    processing_logic(body)


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


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

By raising AckMessage, FastStream will halt the current message processing routine and immediately acknowledge it. Analogously, raising NackMessage would prevent the message from being acknowledged and could lead to its subsequent reprocessing by the same or a different consumer.