CoreWildcard

CoreWildcard reads every Core NATS subject a pattern matches, * for one token and > for the rest, and the mount site names where a delayed copy goes.

A subject is a list of tokens separated by dots. orders.* matches orders.created and orders.cancelled, but not orders.eu.created. orders.> matches all three, and > alone matches every subject. CoreWildcard::new takes the pattern, and its one setting is the queue group. The subscription's name is the pattern itself. A pattern names many subjects and none of them in particular, so the crate does not choose where a delayed retry copy goes. The mount site names that subject with out_retry(Publish).to(subject). A registration over a pattern that names none stops the service at startup, with an error naming the subscription. A subject the pattern matches brings the copy back to the same handler.

Example

Every one-token order event is audited, and a delayed copy goes to orders.audit.

Macros
Manual

examples/nats_subscriptions_wildcard.rs
/// Audits every one-token order event: `orders.created`, `orders.cancelled`, and the rest.
#[subscriber(CoreWildcard::new("orders.*"))]
async fn audit(event: &OrderEvent) -> HandlerOutcome {
    tracing::info!(event.order_id, "order event audited");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // A pattern is no destination, so the mount site names where a delayed copy goes.
            b.include(audit).out_retry(Publish).to("orders.audit");
        },
    )
}

examples/manual/nats_subscriptions_wildcard.rs
/// Audits every one-token order event: `orders.created`, `orders.cancelled`, and the rest.
struct Audit;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // A pattern is no destination, so the mount site names where a delayed copy goes.
            b.include(subscriber(CoreWildcard::new("orders.*"), Audit).build())
                .out_retry(Publish)
                .to("orders.audit");
        },
    )
}

Example​

Example