message_id

message_id tags a publish with Nats-Msg-Id, so a repeat inside the stream's deduplication window is stored once.

The stream remembers the ids it stored for its deduplication window, two minutes unless the stream is configured otherwise. A second message with the same id inside the window is acknowledged and not stored, and its acknowledgement reports duplicate. An id derived from the input, such as order-{id}, makes a redelivered input publish the same message again without storing it twice. The id describes one message, so the handler body states it as a step on the publish builder. The step comes from JetStreamPublishSteps in this crate's prelude, and the body bounds its slot Out<impl Publisher<Options = JetStreamOptions>, Slot>. That bound ties the body to JetStream, and its signature says so. The step writes into the call's JetStreamOptions; the publish still leaves through the mount site's policy and codec. A test reads the value back with tb.out::<Slot>().with_options(&JetStreamOptions { .. }).

Example

Every shipped order is archived under an id derived from the order, so a redelivered order is stored once.

Macros
Manual

examples/nats_jetstream.rs
/// The copy of an order kept in the archive stream.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "archive.orders")]
struct Archived {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Archived)]
struct Archive;

/// Archives every shipped order under a deduplication id, so a redelivered order is stored once.
/// The id belongs to the message, so the body states it, and its slot bound names the `JetStream`
/// options type.
#[subscriber(JetStreamSubject::new("orders.shipped", "ORDERS").durable("archiver"))]
async fn archive(
    order: &Order,
    Out(copies): Out<impl Publisher<Options = JetStreamOptions>, Archive>,
) -> HandlerOutcome {
    let sent = copies
        .message(&Archived { id: order.id })
        .message_id(format!("order-{}", order.id))
        .publish()
        .await;
    if let Err(err) = sent {
        tracing::warn!(order.id, %err, "archive copy refused, redelivering");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/nats_jetstream.rs
/// The copy of an order kept in the archive stream.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "archive.orders")]
struct Archived {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Archived)]
struct Archive;

/// Archives every shipped order under a deduplication id, so a redelivered order is stored once.
/// The id belongs to the message, so the body states it, and its slot bound names the `JetStream`
/// options type.
struct ArchiveOrder;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for ArchiveOrder
where
    Entry: OutEntry<Archive, Wire: Publisher<Options = JetStreamOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        outs.get(Archive)
            .message(&Archived { id: order.id })
            .message_id(format!("order-{}", order.id))
            .publish()
            .await
            .map_err(|err| {
                tracing::warn!(order.id, %err, "archive copy refused, redelivering");
                HandlerOutcome::retry()
            })
    }
}

Example​

Example