Skip to main content

Publishing to streams

This guide publishes replies and messages into JetStream streams, stores a repeated message once, and refuses a write made from a stale version.

A plain publish goes out over Core NATS and nobody confirms it. JetStreamPublish waits for the stream to acknowledge each message. So a message the stream refuses is an error the handler sees, not a silent loss.

The example is an orders service. It confirms each order, archives it, and keeps the history of each customer account. The streams it writes to exist on the server:

nats stream add ORDERS --subjects 'orders.placed' --defaults
nats stream add CONFIRMATIONS --subjects 'confirmations.orders' --defaults
nats stream add ARCHIVE --subjects 'archive.orders' --defaults
nats stream add ACCOUNTS --subjects 'accounts.*' --defaults

A reply into a stream​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "confirmations.orders")]
struct Confirmation {
    order: u64,
}

#[subscriber(
    JetStreamSubject::new("orders.placed", "ORDERS").durable("confirmer"),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { order: order.id }
}

confirm answers each order with a Confirmation on confirmations.orders. The app below sends that reply through JetStreamPublish::default().expect_stream("CONFIRMATIONS"). expect_stream names the stream that must store the subject, and a publish that another stream would take is refused. The order is acknowledged only after the stream has stored its confirmation. A refused confirmation fails the delivery, and the order comes back.

The reply's subject is outside the ORDERS subjects. A reply stored in the stream a handler reads would reach that handler again.

Storing a message once​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "archive.orders")]
struct Archived {
    order: u64,
    amount_cents: u64,
}

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

#[subscriber(JetStreamSubject::new("orders.placed", "ORDERS").durable("archiver"))]
async fn archive(
    order: &Order,
    Out(out): Out<impl Publisher<Options = JetStreamOptions>, Archive>,
) -> HandlerOutcome {
    let archived = Archived {
        order: order.id,
        amount_cents: order.amount_cents,
    };
    // The same id on every delivery of this order: the stream stores the first copy only.
    if out
        .message(&archived)
        .message_id(format!("order-{}", order.id))
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[subscriber(JetStreamSubject::new("archive.orders", "ARCHIVE").durable("indexer"))]
async fn index(archived: &Archived) -> HandlerOutcome {
    tracing::info!(archived.order, "archived order indexed");
    HandlerOutcome::ack()
}

An order can be delivered twice, and then archive publishes its copy twice. message_id gives every copy of one order the same id, which travels as the Nats-Msg-Id header. The stream stores the first copy and acknowledges a repeat without storing it, so index reads each order once. The stream remembers ids for its duplicate window, two minutes unless the stream sets another.

message_id describes one message, so it is a step of the publish, not a setting of the mount site. The steps come from JetStreamPublishSteps in the crate's prelude. A body that calls one bounds its slot Out<impl Publisher<Options = JetStreamOptions>, _>. That bound ties the handler to JetStream, which is the one place a handler names the broker.

Refusing a stale write​

src/main.rs
/// A change of an account's e-mail, made from the version of the account the client last read:
/// the stream sequence of its last event, or 0 for an account with no events yet.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "commands.change-email")]
struct ChangeEmail {
    account: u64,
    version: u64,
    email: String,
}

/// An event of an account's history, stored under the account's own subject.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct EmailChanged {
    account: u64,
    email: String,
}

#[derive(OutSlot)]
#[publishes(EmailChanged)]
struct History;

#[subscriber("commands.change-email")]
async fn change_email(
    change: &ChangeEmail,
    Out(history): Out<impl Publisher<Options = JetStreamOptions>, History>,
) -> HandlerOutcome {
    let event = EmailChanged {
        account: change.account,
        email: change.email.clone(),
    };
    let stored = history
        .message(&event)
        .to(format!("accounts.{}", change.account))
        .expect_last_subject_sequence(change.version)
        .publish()
        .await;
    if stored.is_err() {
        tracing::warn!(
            change.account,
            change.version,
            "change made from a stale version refused"
        );
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

The history of account 42 is the stream's subject accounts.42. A client changes the account's e-mail from the version it last read: the stream sequence of the account's last event. expect_last_subject_sequence makes the stream store the event only if the subject's last message is still at that sequence. Two clients that read the same version cannot both write: the second write is refused, and change_email drops it. expect_last_sequence checks the whole stream instead of one subject, and expect_last_message_id checks the stream's last message id.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(confirm)
                .out_reply(JetStreamPublish::default().expect_stream("CONFIRMATIONS"));
            b.include(archive)
                .out(
                    Archive,
                    JetStreamPublish::default().expect_stream("ARCHIVE"),
                )
                .build();
            b.include(index);
            b.include(change_email)
                .out(
                    History,
                    JetStreamPublish::default().expect_stream("ACCOUNTS"),
                )
                .build();
        },
    )
}

Each JetStream publish waits for the stream's acknowledgement. Outside a delivery, a startup hook takes the policy and publish_ack returns that acknowledgement: the stream, the sequence, and whether the duplicate window had seen the message. A Core NATS publish is the right choice where losing a message is acceptable, such as the dashboard of Acknowledgement and delivery guarantees.

The tests in the example file run the app in process: the in-process stream checks message ids and expectations as a server does.