Attribute Macro subscriber 

#[subscriber]

Turns an async fn handler into a mountable subscriber definition.

/// Processes incoming orders.
#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerResult { HandlerResult::Ack }
// later: broker_scope.include(handle);

// reply form: the return value is encoded and published to "responses" through the
// TypedPublisher (broker + reply codec) passed at wiring time.
#[subscriber("requests", publish("responses"))]
async fn reply(req: &Request) -> Response { /* ... */ }
// later: broker_scope.include_publishing(reply, typed_publisher);

// reply form with explicit ack control: `Ok` publishes the reply, `Err` skips it and the
// dispatcher acts on the returned HandlerResult.
#[subscriber("requests", publish("responses"))]
async fn confirm(req: &Request) -> Result<Response, HandlerResult> { /* ... */ }

// batch form: the handler takes the whole decoded batch as a slice; the source's
// subscriber must implement BatchSubscriber. Mounted with include_batch.
#[subscriber(batch("orders"))]
async fn bill(orders: &[Order]) -> HandlerResult { /* settles the whole batch */ }

Without publish(..) the handler returns any IntoHandlerResult (a HandlerResult, (), or Result<_, E>). With publish(..) it returns the reply value to publish, or Result<Reply, HandlerResult> to control acknowledgement: Err(result) publishes nothing and returns result to the dispatcher. The Result form is detected syntactically, so spell it out in the signature (a type alias is treated as a plain reply type).

Wrapping the source in batch(..) switches the definition to a BatchDef: the handler takes &[T] and runs once per batch pulled from the broker’s BatchSubscriber (use the Buffered adapter for brokers without native batching). It returns any IntoBatchResult - one outcome for the whole batch (HandlerResult, (), Result<_, E>), or Vec<HandlerResult> to settle element i of the slice with outcome i. The source type is recovered from the constructor path, so a generic source spells its parameters: batch(Buffered::<Name>::new(Name::new("orders"))).

Combining batch(..) with publish(..) produces a BatchPublishingDef (mounted with include_batch_publishing): the handler returns Vec<Reply> (or Result<Vec<Reply>, HandlerResult> for explicit ack control, all-or-nothing - selective outcomes do not compose with a transaction), every reply is published to the reply name, and the whole batch is acked after. Hand the mount a TypedPublisher for independent reply publishes, or .transactional() for one transaction per batch.

A workers(n) clause processes up to n deliveries (or batches) of this subscriber concurrently, each in its own task; global processing order is lost by design, and back-pressure holds at n in-flight deliveries. workers(n, by_key) switches to n sequential lanes keyed by the message’s partition key, preserving per-key ordering (single-message forms only). The default is the sequential loop.

In both forms the handler may declare an optional second parameter, the per-delivery &mut Context, to read app state or publish manually.