How can I implement NATS MESSAGE-ID?

[Muhammadali-Akbarov]

import json

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

broker = NatsBroker()
app = FastStream(broker)

CHANNEL_NAME = "telegram.billing"


@broker.subscriber(CHANNEL_NAME)
async def handler(msg: str, logger: Logger):
    """
    handler.
    """
    logger.info(msg)


@app.after_startup
async def test_send():
    """
    test send.
    """
    msg = {
        "channel_type": "billing",
        "message": "Test message"
    }
    msg = json.dumps(msg)

    for _ in range(0, 2):
        await broker.publish(msg, CHANNEL_NAME, correlation_id="123")

output

2024-05-10 10:46:47,488 INFO     - FastStream app starting...
2024-05-10 10:46:47,541 INFO     - telegram.billing |            - `Handler` waiting for messages
2024-05-10 10:46:47,541 INFO     - FastStream app started successfully! To exit, press CTRL+C
2024-05-10 10:46:47,555 INFO     - telegram.billing | aa7e66a2-9 - Received
2024-05-10 10:46:47,555 INFO     - telegram.billing | aa7e66a2-9 - **{"channel_type": "billing", "message": "Test message"}**
2024-05-10 10:46:47,555 INFO     - telegram.billing | aa7e66a2-9 - Processed
2024-05-10 10:46:47,555 INFO     - telegram.billing | 4d2a8a55-b - Received
2024-05-10 10:46:47,556 INFO     - telegram.billing | 4d2a8a55-b - **{"channel_type": "billing", "message": "Test message"}**
2024-05-10 10:46:47,556 INFO     - telegram.billing | 4d2a8a55-b - Processed

[sheldygg]

Hi, as far as I understand you, you want to send message with deduplication header, it is called Nats-Msg-Id.
You can pass headers as dict to broker.puslish method.

await broker.publish(message_body, headers={"Nats-Msg-Id": "your_deduplication_key"})

[Muhammadali-Akbarov]

Thanks you sheldygg, I'll try with it

[Muhammadali-Akbarov]

Lol it's not working

[Muhammadali-Akbarov]

its coming two times with similar nats message id

[Muhammadali-Akbarov]

is it jet stream, I'm I right?

[sheldygg]

By default FastStream are use Nats-Core instead of Nats-Jetstream, for enable JS you must pass stream parameter to your subscriber.
Minimal reproducible example:

import json
import uuid

from faststream import FastStream, Logger
from faststream.nats import NatsBroker, JStream, NatsMessage

broker = NatsBroker()
app = FastStream(broker)

CHANNEL_NAME = "telegram.billing"
publisher = broker.publisher(CHANNEL_NAME)


@broker.subscriber(CHANNEL_NAME, stream=JStream(name="stream_name"), durable="durable_name")
async def handler(msg: NatsMessage, logger: Logger):
    """
    handler.
    """
    logger.info(msg)
    await msg.ack()


@app.after_startup
async def test_send():
    """
    test send.
    """
    msg = {
        "channel_type": "billing",
        "message": "Test message"
    }
    msg = json.dumps(msg)
    deduplication_key = str(uuid.uuid4())

    for _ in range(0, 2):
        await publisher.publish(msg, headers={"Nats-Msg-Id": deduplication_key})

[Muhammadali-Akbarov]

Thanks dude, will try tomorrow