filter_subject

filter_subject narrows the subjects of the stream a consumer reads; it defaults to the subject the descriptor names.

A stream often stores several kinds of message under one prefix, such as every order event under orders.>. The consumer's filter decides which of them it is handed; the server skips the rest for this consumer. Without filter_subject the filter is the subject given to JetStreamSubject::new. With it, the descriptor's subject stays the subscription's name, and the filter decides what arrives. A filter may be a pattern with * and >. A durable consumer has one filter, the one its last subscription opened with.

Example

The stream ORDERS stores every order event, and the refunds consumer reads orders.refunded only.

Macros

examples/nats_subscriptions_filter.rs

/// Reads only the refunds of the `ORDERS` stream, which stores every order event.
#[subscriber(
    JetStreamSubject::new("orders.>", "ORDERS")
        .durable("refunds")
        .filter_subject("orders.refunded")
)]
async fn refund(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order refunded");
    HandlerOutcome::ack()
}

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

Manual

examples/manual/nats_subscriptions_filter.rs

/// Reads only the refunds of the `ORDERS` stream, which stores every order event.
struct Refund;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("refunds", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let refunds = JetStreamSubject::new("orders.>", "ORDERS")
                .durable("refunds")
                .filter_subject("orders.refunded");
            b.include(subscriber(refunds, Refund).build());
        },
    )
}