CoreSubject

CoreSubject reads one Core NATS subject; it stores nothing, and a delayed copy of a delivery is published back to the subject itself.

CoreSubject::new takes the subject, and its one setting is the queue group. The server delivers a message to every subscription on the subject at that moment and keeps nothing for later. A subject is published to and subscribed to under one name, so a delayed retry needs nothing from the mount site. A retry_after(delay) settles the delivery, and after the delay the runtime publishes a copy back to the subject. The copy waits in the service's memory, so a process that exits before the timer fires loses it. A subject with * or > in it is a pattern, which is CoreWildcard; CoreSubject given one stops the service at startup. An empty subject stops it too. A stream that keeps messages for a consumer is JetStreamSubject.

Example

A payment the bank has not cleared yet comes back to payments.captured after 30 seconds.

Macros
Manual

examples/nats_subscriptions_core.rs
/// Books a cleared payment; one the bank has not cleared yet comes back in 30 seconds, as a copy
/// published to `payments.captured` again.
#[subscriber(CoreSubject::new("payments.captured"))]
async fn book(payment: &Payment) -> HandlerOutcome {
    if !payment.cleared {
        tracing::info!(payment.id, "not cleared yet, checking again later");
        return HandlerOutcome::retry_after(RECHECK);
    }
    tracing::info!(payment.id, "payment booked");
    HandlerOutcome::ack()
}

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

examples/manual/nats_subscriptions_core.rs
/// Books a cleared payment; one the bank has not cleared yet comes back in 30 seconds, as a copy
/// published to `payments.captured` again.
struct Book;

impl Handle<Payment> for Book {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !payment.cleared {
            tracing::info!(payment.id, "not cleared yet, checking again later");
            return ready(Err(HandlerOutcome::retry_after(RECHECK)));
        }
        tracing::info!(payment.id, "payment booked");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(subscriber(CoreSubject::new("payments.captured"), Book).build());
        },
    )
}

Example​

Example