Attribute Macro subscriber 

#[subscriber]

Turns an async fn handler into a mountable subscriber definition.

The attribute carries the declarative settings of the subscription; the shape of the handler - one message (&T) or a whole batch of them (&[T]) - is read off the signature, which is the only place it was ever really written. Which lane a type rides is the type’s own business: a serde type decodes through the scope codec, a #[derive(Deserialized)] type constructs itself from the payload bytes, and a reply encodes through the reply codec unless its type is #[derive(Serialized)], in which case its bytes leave as they are.

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

// reply form: the return value is encoded and published through the reply publisher
// attached at the include site (the broker's default without it), at the destination its
// own type declares.
#[derive(Serialize, Outgoing)]
#[outgoing(name = "responses")]
struct Response { /* ... */ }

#[subscriber("requests", reply)]
async fn reply(req: &Request) -> Response { /* ... */ }

// a reply type declaring no name takes the clause's:
#[subscriber("requests", reply("responses"))]
async fn reply_elsewhere(req: &Request) -> Receipt { /* ... */ }

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

// batch form: the slice parameter says the handler takes the whole decoded batch; the
// subscription's subscriber must implement BatchSubscriber (natively, or through the
// buffer the mount site adds).
#[subscriber("orders")]
async fn bill(orders: &[Order]) -> HandlerOutcome { /* settles the whole batch */ }

// raw form: no codec, no serde - the input type deserializes itself from the payload
// bytes. `Deserialize` means the framework's codec does it; `Deserialized` means it is
// already done by the user's own type.
#[derive(Deserialized)]
struct Frame<'a>(&'a [u8]);

#[subscriber("frames")]
async fn on_frame(frame: &Frame<'_>) -> HandlerOutcome { /* parse it yourself */ }

// raw batch: the batch shape on the same lane; the views borrow the batch's own messages
// for the duration of the call.
#[subscriber("frames")]
async fn ingest(frames: &[Frame<'_>]) -> HandlerOutcome { /* parse them yourself */ }

// the settings the attribute leaves out are filled in at the mount site, through the same
// builder the attribute expands into:
#[subscriber]
async fn audit(order: &Order) -> HandlerOutcome { HandlerOutcome::ack() }
// later: broker_scope.include(audit.name(subject).workers(nonzero!(4)));

// raw reply: the reply type carries its own bytes, so they are published as they are -
// no codec on the way out. `Serialize` means the framework's codec does it; `Serialized`
// means it is already done by the user's own type. The same reply clause serves both
// wires, and Result<Export, HandlerOutcome> gives the usual explicit ack control.
#[derive(Outgoing, Serialized)]
struct Export(Vec<u8>);

#[subscriber("frames", reply("frames-out"))]
async fn mirror(frame: &Frame<'_>) -> Export { Export(frame.0.to_vec()) }

// the lanes mix freely (the gateway shape): the input decodes with the scope codec, the
// returned bytes still go out unencoded - and the other diagonal (a Deserialized input
// with an encoded serde reply) works the same way.
#[subscriber("orders", reply("orders-wire"))]
async fn encode(order: &Order) -> Export { /* your wire format */ }

A reply type derives Outgoing: #[outgoing(name = "..")] on it is the destination, and a derive without a name takes the clause’s reply("..") as the destination instead. A declared name wins over the clause’s, and startup logs a warning when the two differ.

Without a reply clause the handler returns any accepted outcome shape (a HandlerOutcome, (), or Result<_, E>). Attach a post-settle continuation with HandlerOutcome::ack().and_after (any outcome works), which runs after the message is settled. With one it returns the reply value to publish, or Result<Reply, HandlerOutcome> to control acknowledgement: Err(outcome) publishes nothing and settles by outcome. The Result form is detected syntactically, so spell it out in the signature (a type alias is treated as a plain reply type). A failed reply publish nacks the delivery with requeue on both wires, so a reply is retried rather than silently lost.

A &[T] message parameter makes the definition a batch one: the handler runs once per batch pulled from the subscription’s BatchSubscriber (added by the mount site’s buffer for subscriptions without native batching). It returns one outcome for the whole batch (HandlerOutcome, (), Result<_, E>), or a per-element Vec<HandlerOutcome> to settle element i of the slice with outcome i, each element carrying its own optional and_after continuation. A &[Message<H, T>] parameter pairs each element with its typed header contract: the core decodes both under the subscriber’s decode policy, and the body reads element.headers next to element.body (the single-message counterpart stays the Headers<T> extractor parameter). A batch body may name the broker’s subscription-scoped batch context (ctx: &mut Context<'_, MemoryBatchContext>-style) to reach handles every delivery of the subscription shares, like a seeker.

Combining a batch handler with reply(..) produces a batch reply definition: the handler returns Vec<Reply> (or Result<Vec<Reply>, HandlerOutcome> 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. Name the reply’s policy with .out(Reply, ..) at the mount site for independent reply publishes, and chain .transactional() for one transaction per batch.

A workers(n) clause processes up to n deliveries (or batches) of this subscriber concurrently on long-lived workers; 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.

A threads(n) clause is the same position for a handler that computes: the subscription’s deliveries (or batches) are handled on n dedicated threads of its own, each running a current-thread runtime, so the computation holds up none of the app runtime’s threads. threads(n, by_key) keeps a key on one thread. A subscription names workers(..) or threads(..), not both. What the delivery leaves behind (continuations, hooks, a retry timer) stays on its thread; the handler sends work to the app’s runtime through the spawn and run methods of a Ctx(main): Ctx<MainRuntime> parameter.

#[subscriber("images.resize", threads(8))]
async fn resize(job: &Resize) -> HandlerOutcome {
    render(job);
    HandlerOutcome::ack()
}

Clause values need not be literals: workers(..) and threads(..) take any usize expression (a constant, a static, a function call - an integer literal keeps the compile-time zero rejection, a runtime value of zero panics at registration), reply(..) takes a &'static str expression, and on_failure(..) keys accept a FailurePolicy expression next to the keyword vocabulary.

An Out(out): Out<P> parameter injects a live publisher, paired at startup from the policy attached at the include site (b.include(f).out(marker, policy).build()); its optional third position declares the message set the handler publishes (Out<impl Publisher, Marker, (A, B)> - a tuple, a single type, or a #[derive(OutMessages)] set enum), narrowing what the publish builder accepts on it, and the generated document reports that narrowed list as the handler’s send operations. Out parameters combine freely in one handler: with each other, with a byte input, with a batch handler, and with every reply form (the two reply spellings and the batch publishing form). An Out parameter’s attachment is required at the include site: one .out(marker, policy) per slot, next to the reply’s own optional .out(Reply, policy), and a .build() to commit. Repositioning a subscription from inside the body is not a parameter but a broker context field: the broker publishes its position and seek-handle keys, and the handler reads them with the Ctx extractor like any other broker field.

A start_at(<position>) clause opens the subscription at that position instead of the broker’s default, seeking before the first delivery (“start from the latest on deploy”, “replay the whole log”). The position is the broker’s own type, named by its constructor (MemoryPosition::start(), a Kafka-style latest()); the source’s subscriber must implement the Seekable capability, so the clause does not compile against a broker without a replayable log. The position is forced on every startup; without the clause the subscription simply opens at the broker’s default, and conditional defaults stay on the broker’s own subscription descriptor.

// Opens at the start of the log: entries published before the service started are
// replayed into the fresh subscription.
#[subscriber("audit", start_at(MemoryPosition::start()))]
async fn record(entry: &Entry) -> HandlerOutcome { /* ... */ }

In both forms the handler may declare an optional second parameter, the per-delivery &mut Context, to read app state or publish manually. Any further parameter is an extractor: its type must implement FromContext, and the generated handler resolves it from the delivery context (in declaration order) before the body runs, so dependencies arrive as arguments. A failed extraction settles the delivery by the rejection’s HandlerOutcome without running the body.

// `State<Db>` is resolved from the application state before the body runs.
#[subscriber("orders")]
async fn handle(order: &Order, State(db): State<Db>) -> HandlerOutcome { /* ... */ }

The attribute expands into the definition and then the settings builder over it, so the clauses above are exactly the calls a user would write at the mount site (see SubscriberSettings in the core crate). A setting the attribute names is fixed in the definition’s type and its builder method no longer applies, so the two can never disagree.