durable

durable names a JetStream consumer: the server keeps its position across restarts, and subscriptions sharing it take messages in turn.

A named consumer lives on the server, apart from any subscription. It remembers which messages were acknowledged, so a restarted service resumes after the last one and reads what arrived while it was down. A consumer without a name is ephemeral and starts from its deliver policy on every startup. Every replica of a service that reads through one durable shares its messages: each message goes to one of them, as in a queue group. At startup the subscription creates the consumer, or updates it when it exists. So the settings the last subscription opened with are the consumer's settings, and replicas that share a durable name one filter and one policy.

Example

Every replica bills through the consumer billing, and a restarted service resumes where the consumer stopped.

Macros
Manual

examples/nats_subscriptions_durable.rs
/// Every replica reads through the consumer `billing`: each invoice is billed once, and a
/// restarted service resumes where the consumer stopped.
#[subscriber(JetStreamSubject::new("invoices.*", "INVOICES").durable("billing"))]
async fn bill(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.id, "invoice billed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(bill);
        },
    )
}

examples/manual/nats_subscriptions_durable.rs
/// Every replica reads through the consumer `billing`: each invoice is billed once, and a
/// restarted service resumes where the consumer stopped.
struct Bill;

impl Handle<Invoice> for Bill {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(invoice.id, "invoice billed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let invoices = JetStreamSubject::new("invoices.*", "INVOICES").durable("billing");
            b.include(subscriber(invoices, Bill).build());
        },
    )
}

Example​

Example