DeliverPolicy

DeliverPolicy says where a newly created consumer starts: all, new, last, last per subject, a sequence or a time.

deliver_policy sets it on the descriptor. The type is the one async-nats defines, re-exported by the crate. All, the default, starts with the oldest message the stream holds. New starts with the first message published after the consumer was created. Last starts with the newest message of the stream, and LastPerSubject with the newest message of each subject the filter matches. ByStartSequence starts at a stream sequence, and ByStartTime at the first message stored at or after a time. The policy applies once, when the consumer is created. A durable consumer that already exists resumes from its own position. An ephemeral consumer is created on every startup, so it starts from its policy each time.

Example

A live ticker starts with the next price, and the prices the stream held before it started stay behind.

Macros
Manual

examples/nats_subscriptions_deliver.rs
/// A live ticker: each new consumer starts with the next price, not with the stream's history.
#[subscriber(JetStreamSubject::new("prices.*", "PRICES").deliver_policy(DeliverPolicy::New))]
async fn tick(price: &Price) -> HandlerOutcome {
    tracing::info!(price.product, price.cents, "price changed");
    HandlerOutcome::ack()
}

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

examples/manual/nats_subscriptions_deliver.rs
/// A live ticker: each new consumer starts with the next price, not with the stream's history.
struct Tick;

impl Handle<Price> for Tick {
    fn handle(
        &self,
        price: &Price,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(price.product, price.cents, "price changed");
        ready(Ok(()))
    }
}

fn prices() -> JetStreamSubject {
    JetStreamSubject::new("prices.*", "PRICES").deliver_policy(DeliverPolicy::New)
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("ticker", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(subscriber(prices(), Tick).build());
        },
    )
}

Example​

Example