Trait NatsSubscription

Source

pub trait NatsSubscription: Sealed {
    // Required method
    fn subject(&self) -> &str;

    // Provided method
    fn ensure_subject(&self) -> Result<(), NatsError> { ... }
}

Expand description

A NATS subscription descriptor: CoreSubject, CoreWildcard or JetStreamSubject.

Sealed, because the two delivery models NATS has are the two this crate ships.

The three differ in where a delayed copy of a delivery goes. A CoreSubject is its own destination, so the framework publishes the copy back to it. A CoreWildcard names many subjects and none in particular, so its mount site names one with out_retry(..).to(..). A JetStreamSubject needs no copy: the server holds the message for the delay.

§Examples

use std::time::Duration;

use ruststream_nats::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Payment {
    id: u64,
    settled: bool,
}

#[subscriber("payments.settled")]
async fn reconcile(payment: &Payment) -> HandlerOutcome {
    if payment.settled {
        return HandlerOutcome::ack();
    }
    HandlerOutcome::retry_after(Duration::from_secs(30))
}

#[subscriber(CoreWildcard::new("payments.*"))]
async fn audit(payment: &Payment) -> HandlerOutcome {
    tracing::warn!(payment.id, "audit pending");
    HandlerOutcome::retry_after(Duration::from_secs(30))
}

#[subscriber(JetStreamSubject::new("payments.stored", "PAYMENTS").durable("reconciler"))]
async fn store(payment: &Payment) -> HandlerOutcome {
    tracing::warn!(payment.id, "store pending");
    HandlerOutcome::retry_after(Duration::from_secs(30))
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // One subject is its own destination.
            b.include(reconcile)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dead");

            // A pattern is not, so the mount site names where a copy lands.
            b.include(audit)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead.payments")
                .out_retry(Publish)
                .to("payments.settled");

            // The server holds the message, so nothing is published at all.
            b.include(store)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dead");
        },
    )
}

Required Methods§

Source

fn subject(&self) -> &str

The subject pattern this subscription receives messages on.

Provided Methods§

Source

fn ensure_subject(&self) -> Result<(), NatsError>

The one thing about a subscription the types cannot settle: the subject is a string the caller supplies, often from configuration, and a descriptor constructor is the expression a #[subscriber(..)] attribute writes, so it cannot hand back a Result. Every subscribe implementation calls this before it does any work.