and_after

and_after attaches a continuation to an outcome: work that runs on a task of its own once the delivery is settled.

HandlerOutcome::ack().and_after(fut) settles the delivery first, then runs fut. A continuation attaches to any of the outcomes, drop() and retry() included; a second and_after replaces the first. On a batch each element carries its own continuation, so only the elements that asked for one run it. Post-settle work is at most once: the delivery is already settled, so a lost or panicking continuation never brings it back. Work whose loss must redeliver the message belongs in the body, before the settlement. A graceful shutdown waits for running continuations within the app's shutdown_timeout. A test waits for them with drain. A hook on the context runs after the settlement too, gated by which settlement it was.

Example

Macros

examples/post_settle.rs

#[subscriber("orders")]
async fn accept(order: &Order, ctx: &mut Context<'_, (), Arc<Notifier>>) -> HandlerOutcome {
    let notifier = Arc::clone(ctx.state());
    let id = order.id;
    // Runs once the ack is through; losing it never brings the order back.
    HandlerOutcome::ack()
        .and_after(async move { notifier.send(format!("accepted {id}")).await })
}

Manual

examples/manual/post_settle.rs

/// The outcome with its continuation travels as `Err(outcome)`: a plain `Ok(())` acks with
/// none.
struct Accept;

impl Handle<Order, (), (), (), Arc<Notifier>> for Accept {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), Arc<Notifier>>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let notifier = Arc::clone(ctx.state());
        let id = order.id;
        // Runs once the ack is through; losing it never brings the order back.
        ready(Err(HandlerOutcome::ack().and_after(async move {
            notifier.send(format!("accepted {id}")).await;
        })))
    }
}

One continuation per element

Macros

examples/post_settle.rs

/// Each element carries its own continuation: the dropped order gets none.
#[subscriber("orders")]
async fn accept(
    orders: &[Order],
    ctx: &mut Context<'_, (), Arc<Notifier>>,
) -> Vec<HandlerOutcome> {
    orders
        .iter()
        .map(|order| {
            if order.id == 0 {
                return HandlerOutcome::drop();
            }
            let notifier = Arc::clone(ctx.state());
            let id = order.id;
            HandlerOutcome::ack()
                .and_after(async move { notifier.send(format!("accepted {id}")).await })
        })
        .collect()
}

Manual

examples/manual/post_settle.rs

/// Each element carries its own continuation: the dropped order gets none.
struct Accept;

impl Handle<[Order], (), (), (), Arc<Notifier>> for Accept {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        ctx: &mut Context<'_, (), Arc<Notifier>>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        ready(Err(orders
            .iter()
            .map(|order| {
                if order.id == 0 {
                    return HandlerOutcome::drop();
                }
                let notifier = Arc::clone(ctx.state());
                let id = order.id;
                HandlerOutcome::ack().and_after(async move {
                    notifier.send(format!("accepted {id}")).await;
                })
            })
            .collect()))
    }
}

fn app(broker: MemoryBroker<Retaining>) -> impl App<State = Arc<Notifier>> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Arc::new(Notifier::default())))
        .with_broker(broker, |b| {
            b.include(
                subscriber("orders", Accept)
                    .batch(nonzero!(3))
                    .start_at(MemoryPosition::start())
                    .build(),
            );
        })
}