Settings

Settings the attribute leaves open are chained where the handler is mounted: name, workers, failure policy, batch size and start position.

Each setting has two spellings: a clause in the attribute, or a method of SubscriberSettings at the mount site.

At the mount site	In the attribute	What it sets
`.name(name)`	`#[subscriber("name")]`	the subscription a source left open
`.workers(n)`, `.workers_by_key(n)`	`workers(n)`, `workers(n, by_key)`	how many deliveries run at once (workers)
`.threads(n)`, `.threads_by_key(n)`	`threads(n)`, `threads(n, by_key)`	dedicated threads for a handler that computes (threads)
`.on_failure(policies)`	`on_failure(panic = .., decode = ..)`	what a panic or an undecodable payload does (failures)
`.batch(n)`	none	the batch size of a `&[T]` handler (batches)
`.start_at(position)`	`start_at(position)`	where a replayable subscription opens (`start_at`)

A setting the attribute named is fixed in the definition's type, so naming it again at the mount site does not compile. The mount site is where a value known only at run time goes: a name from configuration, a pool size per deployment. Counts are NonZeroUsize, and the nonzero! macro writes one: nonzero!(4) compiles and nonzero!(0) does not. on_failure takes a FailurePolicies value: one FailurePolicy for a panic and one for a decode failure. A subscription takes workers or threads, not both: they fill one position. batch exists only on a batch handler, and a batch handler does not mount without it. start_at compiles only against a broker whose subscriptions are replayable. A broker crate adds steps of its own to the same chain, such as a consumer group, in the broker's vocabulary. In the manual form the same methods chain on subscriber(source, body) before .build().

Example

Macros
Manual

examples/subscriber_settings.rs
/// Everything is left to the mount site: the name, the pool, the failure policy.
#[subscriber]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "billed");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            bill.name("orders")
                .workers(nonzero!(4))
                // A payload that does not decode is acked and skipped, not dropped.
                .on_failure(FailurePolicies::default().with_decode(FailurePolicy::Skip)),
        );
    })
}

examples/manual/subscriber_settings.rs
struct Bill;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            subscriber(Unnamed::<Name>::new(), Bill)
                .name("orders")
                .workers(nonzero!(4))
                // A payload that does not decode is acked and skipped, not dropped.
                .on_failure(FailurePolicies::default().with_decode(FailurePolicy::Skip))
                .build(),
        );
    })
}

Example​

Example