Skip to main content

Subjects, wildcards and queue groups

This guide names Core NATS subjects, reads many of them with a wildcard, shares the work between replicas in a queue group and keeps the order of each key in parallel lanes.

A subject is a name of tokens separated by dots, such as orders.eu.created. Core NATS hands a message to every subscription whose subject matches at that moment, and stores nothing.

The example is an orders service. It packs the European orders, bills them, and audits every order event of every region.

A queue group​

pack reads orders.eu.created. The service runs in several replicas, and each order must be packed once.

src/main.rs
#[subscriber(CoreSubject::new("orders.eu.created").queue_group("packers"))]
async fn pack(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order packed");
    HandlerOutcome::ack()
}

CoreSubject reads one subject. queue_group puts the subscription into the group packers. The server hands each message to one member of a group, so each replica packs a share of the orders. A group name is one group for the server, even across different subjects. A subscription outside the group, or in another group, still gets its own copy.

Parallel lanes per customer​

bill reads the same subject without a group, so it gets every order that pack gets. It bills up to eight orders at once and keeps the order of each customer.

src/main.rs
/// The customer an order belongs to, carried in the partition-key header.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct CustomerKey {
    #[serde(rename = "nats-partition-key")]
    customer: String,
}

/// The invoice of one order, keyed by the customer like the order it bills.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoices.eu", headers = CustomerKey)]
struct Invoice {
    order: u64,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Invoice)]
struct Invoices;

#[subscriber("orders.eu.created", workers(8, by_key))]
async fn bill(
    order: &Order,
    Headers(key): Headers<CustomerKey>,
    Out(invoices): Out<impl Publisher, Invoices>,
) -> HandlerOutcome {
    let invoice = Invoice {
        order: order.id,
        amount_cents: order.amount_cents,
    };
    if invoices
        .message(&invoice)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, customer = %key.customer, "order billed");
    HandlerOutcome::ack()
}

workers(8, by_key) runs eight lanes. Deliveries with the same key run one after another on one lane. In the manual form the same setting is workers_by_key. NATS has no native key, so the key travels in the nats-partition-key header (PARTITION_KEY_HEADER), and the sender sets it. CustomerKey is a header contract with one field under that name. Headers<CustomerKey> reads the key of the delivery, and with_headers(&key) writes it on the invoice. So the invoices of one customer keep their order for the next reader too. An order without the header fails the contract and is settled by the decode policy, which drops it by default.

A NATS header is text: the name is printable ASCII without a colon, and the value is one line. A header that does not fit fails the publish.

A wildcard​

audit reads every order event of every region.

src/main.rs
#[subscriber(CoreWildcard::new("orders.>"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order event audited");
    HandlerOutcome::ack()
}

CoreWildcard reads every subject a pattern matches. * matches one token: orders.*.created reads the creations of every region. > matches one or more trailing tokens: orders.> reads every subject under orders.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(pack);
            b.include(bill).out(Invoices, Publish).build();
            // A pattern is no subject to publish a retry copy to, so the mount site names one.
            b.include(audit).out_retry(Publish).to("orders.retry");
        },
    )
}

A pattern reads many subjects, so a delayed copy of its delivery has no single subject to go to. The mount site names one with out_retry(Publish).to(..), here orders.retry, which the pattern also reads. Every registration over a pattern names it, even when its handler never retries, and the service refuses to start without it. Retries and dead letters explains the copies.

At most once​

Core NATS delivers a message to the subscriptions that exist when it arrives. A message published while no subscription matches is gone. A delivery has no acknowledgement either, so a handler that fails does not get the message again from the server. When every message counts, the subject belongs to a stream: Streams and consumers shows how to read one.

The tests in the example file run the app in process and check each subscription's copy, the pattern, and the order of the invoices per customer.