Hooks

A post-settle hook registered on the context runs once the delivery is settled, only for the settlement it names.

ctx.after_ack(fut) registers a hook that runs only when the delivery is acked. ctx.after(outcome) returns an After, and its then(fut) runs only when the settlement matches that outcome by kind; retry_after matches any delay. ctx.after_settle(fut) runs whatever the settlement. Hooks accumulate, and every matching one runs. On a batch the context belongs to the whole batch: only after_settle hooks run there, once, after every element is settled. A hook is at most once, like a continuation of and_after: a lost or panicking hook never brings the delivery back. A graceful shutdown waits for running hooks within the app's shutdown_timeout, and a test waits for them with drain.

Example

Macros

examples/post_settle.rs

#[subscriber("orders")]
async fn accept(order: &Order, ctx: &mut Context<'_, (), Arc<Notifier>>) -> HandlerOutcome {
    let id = order.id;
    let notifier = Arc::clone(ctx.state());
    ctx.after_ack(async move { notifier.send(format!("shipping {id}")).await });
    let notifier = Arc::clone(ctx.state());
    ctx.after(HandlerOutcome::drop())
        .then(async move { notifier.send(format!("rejected {id}")).await });
    let notifier = Arc::clone(ctx.state());
    ctx.after_settle(async move { notifier.send(format!("settled {id}")).await });

    if id == 0 {
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/post_settle.rs

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 id = order.id;
        let notifier = Arc::clone(ctx.state());
        ctx.after_ack(async move { notifier.send(format!("shipping {id}")).await });
        let notifier = Arc::clone(ctx.state());
        ctx.after(HandlerOutcome::drop())
            .then(async move { notifier.send(format!("rejected {id}")).await });
        let notifier = Arc::clone(ctx.state());
        ctx.after_settle(async move { notifier.send(format!("settled {id}")).await });

        if id == 0 {
            return ready(Err(HandlerOutcome::drop()));
        }
        ready(Ok(()))
    }
}