queue_group

queue_group puts a Core NATS subscription in a queue group: each message goes to one member, and one name is one group across subjects.

Without a group, every subscription on a subject receives every message. Replicas of a service that subscribe in one group share the messages instead, and the server picks one member for each. A subscription outside the group still receives everything, so an audit can read beside the workers. The server keys a group by its name alone, so subscriptions with one group name on two subjects or patterns are one group. A message both of them match is delivered once, to one member. CoreSubject::queue_group and CoreWildcard::queue_group take the same name. Core NATS keeps nothing for a group: a message published while no member is subscribed is lost. On a stream the same sharing comes from a durable consumer that several subscriptions read through.

Example

Every replica fulfils orders in the group fulfilment, and the audit reads every order outside it.

Macros
Manual

examples/nats_subscriptions_queue.rs
/// Every replica of the service joins `fulfilment`, so each order is fulfilled once.
#[subscriber(CoreSubject::new("orders.created").queue_group("fulfilment"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order fulfilled");
    HandlerOutcome::ack()
}

/// Outside the group: the audit reads every order.
#[subscriber(CoreWildcard::new("orders.*"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order audited");
    HandlerOutcome::ack()
}

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

examples/manual/nats_subscriptions_queue.rs
/// Every replica of the service joins `fulfilment`, so each order is fulfilled once.
struct Fulfil;

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

/// Outside the group: the audit reads every order.
struct Audit;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let fulfilment = CoreSubject::new("orders.created").queue_group("fulfilment");
            b.include(subscriber(fulfilment, Fulfil).build());
            b.include(subscriber(CoreWildcard::new("orders.*"), Audit).build())
                .out_retry(Publish)
                .to("orders.audit");
        },
    )
}

Example​

Example