Subscribers

#[subscriber] turns an async fn over a reference to the message into a handler that include mounts on a broker.

The first parameter is the message: &T handles one message, &[T] a whole batch. The runtime decodes the payload with the scope's codec before the body runs; a serde type is all the input needs. A payload that does not decode never reaches the body: the decode failure policy settles it, and drops it by default. An optional &mut Context comes second, and any further parameter is an extractor resolved before the body runs: the context page covers both. What the body returns settles the delivery: HandlerOutcome, () (ack), Result<(), E> (ack on Ok, drop on Err) or Result<HandlerOutcome, E>, as the outcomes page shows. The attribute expands into a definition value named after the function, and include on a broker scope of the application or a router mounts it. The attribute's first argument is the subscription source; settings it leaves out are chained at the mount site. The manual form writes the body as an impl Handle<T> of a named type, binds it with subscriber(source, body) and closes the chain with .build(). There Ok(()) acks and Err(outcome) settles by that outcome.

Example

Macros

examples/subscriber_handler.rs

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
    })
}

Manual

examples/manual/subscriber_handler.rs

struct Accept;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Accept).build());
    })
}