Skip to main content

Streams and consumers

This guide reads a JetStream stream through durable and ephemeral consumers, chooses where a new consumer starts, reads the metadata of a delivery and takes messages in batches.

A stream stores the messages published to its subjects, each under a sequence number. A consumer reads a stream at its own pace and acknowledges message by message.

The example is a shop. It fulfils new orders, shows every order event on a dashboard, and reconciles paid orders with the bank.

The stream​

A stream is the server's state. The service finds it at startup and never creates it, so create it once with the NATS command-line tool:

nats stream add ORDERS --subjects 'orders.*' --defaults

ORDERS stores every message published to orders.created, orders.paid and any other subject that orders.* matches. A plain publish to such a subject is stored, whoever publishes it. A subscription to a stream the server does not have fails the service's start.

A durable consumer​

src/main.rs
/// What the warehouse is asked to ship. The sequence is the order's place in the stream.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments.requested")]
struct Shipment {
    order: u64,
    sequence: Option<u64>,
}

#[subscriber(
    JetStreamSubject::new("orders.created", "ORDERS").durable("fulfilment"),
    reply
)]
async fn fulfil(order: &Order, Ctx(sequence): Ctx<StreamSequence>) -> Shipment {
    tracing::info!(order.id, ?sequence, "order sent to the warehouse");
    Shipment {
        order: order.id,
        sequence,
    }
}

JetStreamSubject takes the subject the consumer reads and the stream that stores it. The subject is the consumer's filter, so fulfil reads only orders.created out of ORDERS. filter_subject sets another filter where the two should differ. The service creates the consumer when it starts, with every setting of the descriptor.

durable names the consumer fulfilment. The server keeps its position, so a restarted service continues where it stopped. Replicas of the service share the one durable consumer, and each message goes to one of them. That is the JetStream counterpart of a queue group.

The handler reads the delivery's place in the stream with Ctx<StreamSequence>. The JetStream context also carries the stream and consumer names, the consumer sequence, the delivery count and the number of messages still pending. In the manual form the handler takes Context<'_, JetStreamContext> and reads the same keys with ctx.context(..). On a Core NATS delivery every key reads None.

An ephemeral consumer​

src/main.rs
#[subscriber(JetStreamSubject::new("orders.*", "ORDERS").deliver_policy(DeliverPolicy::New))]
async fn show(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order event on the dashboard");
    HandlerOutcome::ack()
}

A descriptor without durable creates an ephemeral consumer. Each process gets its own, and the server removes it after the subscription closes. So every replica of the service shows every order event.

DeliverPolicy says where a new consumer starts. All, the default, starts at the first message the stream holds, and New at the next message published. Last, LastPerSubject, ByStartSequence and ByStartTime start at the last message, the last message of each subject, a sequence number or a point in time. The policy applies once, when the consumer is created; after that a durable consumer resumes from its own position.

Batches​

src/main.rs
#[subscriber(
    JetStreamSubject::new("orders.paid", "ORDERS")
        .durable("reconciler")
        .pull_expires(NonZeroDuration::from_secs(nonzero!(1)))
)]
async fn reconcile(orders: &[Order]) -> HandlerOutcome {
    let total: u64 = orders.iter().map(|order| order.amount_cents).sum();
    tracing::info!(orders = orders.len(), total, "paid orders reconciled");
    HandlerOutcome::ack()
}

A handler that takes &[Order] gets a batch of messages and runs once for the whole batch. The returned outcome settles every message of it.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(fulfil);
            b.include(show);
            b.include(reconcile.batch(nonzero!(100)));
        },
    )
}

The mount site names the batch size: batch(100) asks the server for up to a hundred messages in one pull. pull_expires on the descriptor says how long a pull waits for them, here one second, after which a smaller batch is handed over. The default is five seconds.

The reply of fulfil is published over Core NATS, the broker's default. Publishing to streams stores replies in a stream instead. Core and JetStream subscriptions share one connection, so one service mixes both models freely.

max_ack_pending caps how many deliveries a consumer holds unacknowledged; the default is 1024. Acknowledgement and delivery guarantees covers how each delivery settles.