JetStreamSubject

JetStreamSubject reads a stream through a pull consumer; the stream exists on the server before the service starts.

A stream stores the messages published to its subjects, and a consumer reads them at its own pace. JetStreamSubject::new(subject, stream) takes the subject to read and the name of the stream that stores it. At startup the subscription finds the stream and creates its consumer; a stream the server does not have stops the service. The crate creates no stream: the service's deployment does, for example with nats stream add. Without durable the consumer is ephemeral, and the server discards it once the subscription closes. Each delivery settles on the server: an ack acknowledges it, and a retry or a delay is a negative acknowledgement the server acts on. The settings and their defaults:

Setting	Default	What it does
`durable`	none, an ephemeral consumer	names the consumer, which keeps its position across restarts
`filter_subject`	the subject	narrows what the consumer reads from the stream
`deliver_policy`	`DeliverPolicy::All`	where a new consumer starts
`max_ack_pending`	1024	how many deliveries the consumer holds unacknowledged
`ack_wait`	30 seconds	how long the server waits for a settlement before it delivers again
`pull_expires`	5 seconds	how long one fetch of a batch waits for messages

The subscription's name is the subject the descriptor names. A delivery carries its place in the stream, which a handler reads through the context keys.

Example

The stream ORDERS stores orders.*, and an ephemeral consumer reads it.

Macros
Manual

examples/nats_subscriptions_jetstream.rs
/// Reads `orders.*` from the stream `ORDERS`; the ack settles each order on the server.
#[subscriber(JetStreamSubject::new("orders.*", "ORDERS"))]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

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

examples/manual/nats_subscriptions_jetstream.rs
/// Reads `orders.*` from the stream `ORDERS`; `Ok` acknowledges each order on the server.
struct Ship;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let orders = JetStreamSubject::new("orders.*", "ORDERS");
            b.include(subscriber(orders, Ship).build());
        },
    )
}

Example​

Example