Subscriptions

A subject in #[subscriber] reads that Core NATS subject; a descriptor carries every other setting of its delivery model.

#[subscriber("orders.created")] reads the subject orders.created over Core NATS, with no queue group. It is the same subscription as CoreSubject::new("orders.created"). Core NATS hands a message to whoever is subscribed at that moment and stores nothing, so a message published before the service starts never reaches it. Every other setting belongs to a descriptor, one type per delivery model:

Descriptor	Reads	Its settings
CoreSubject	one Core NATS subject	queue_group
CoreWildcard	every Core NATS subject a pattern matches	queue_group
JetStreamSubject	a stream, through a pull consumer	durable, filter_subject, deliver_policy, max_ack_pending, ack_wait, pull_expires

A setting of the other model is not a method of the descriptor, so naming one does not compile. A descriptor goes where the subject goes, as the attribute's first argument. The manual form passes the subject, or the descriptor, to subscriber(source, body). The subscription's name is the subject the descriptor names: a test reads its deliveries under that name. All subscriptions of a service share the one connection of its NatsBroker.

Example

The subject orders.created, read over Core NATS.

Macros

examples/nats_subscriptions_name.rs

/// Reads the Core NATS subject `orders.created`.
#[subscriber("orders.created")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

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

Manual

examples/manual/nats_subscriptions_name.rs

/// Reads the Core NATS subject `orders.created`.
struct Place;

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

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