NATS JetStream#

The default NATS usage is suitable for scenarios where:

• The publisher and consumer are always online.
• The system can tolerate messages loss.

If you need stricter restrictions, like:

• An availability of a message processing confirmation mechanism (ack/nack).
• Message persistence (messages will accumulate in the queue when the consumer is offline).

You should use the NATS JetStream extension.

In fact, the JetStream extension is the same as NATS, with the addition of a persistent layer above the file system. Therefore, all interfaces for publishing and consuming messages are similar to regular NATS usage.

However, the JetStream layer has many possibilities for configuration, from the policy of deleting old messages to the maximum stored messages number limit. You can find out more about all JetStream features in the official documentation.

Also, NATS JetStream has built-in key-value (similar to Redis) and object (similar to Minio) storages, which, in addition to the interface for put/get, have the ability to subscribe to events, which can be extremely useful in various scenarios.

FastStream does not provide access to this functionality directly, but it is covered by the nats-py library used. You can access the JS object from the application context:

 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, Logger
from faststream.nats import JStream, NatsBroker

broker = NatsBroker()
app = FastStream(broker)

stream = JStream(name="stream")

@broker.subscriber(
    "js-subject",
    stream=stream,
    deliver_policy="new",
)
async def handler(msg: str, logger: Logger):
    logger.info(msg)

@app.after_startup
async def test_send():
    await broker.publish("Hi!", "js-subject")
    # publish with stream verification
    await broker.publish("Hi!", "js-subject", stream="stream")