Skip to main content

Outcomes

What the handler body returns settles the delivery: a HandlerOutcome, nothing at all, or a Result.

HandlerOutcome::ack() acknowledges the delivery, and the broker forgets it. HandlerOutcome::retry() asks for it again at once; on the in-memory broker it returns to the same subscriber. HandlerOutcome::drop() rejects it without a redelivery; a broker with a dead-letter policy of its own applies it. HandlerOutcome::retry_after(delay) is the fourth settlement, described on the retry_after page. A body that returns () acks every delivery. A body that returns Result<(), E> acks on Ok and drops on Err; the error value is discarded, so log it in the body when it matters. A body that returns Result<HandlerOutcome, E> settles by the inner outcome on Ok and drops on Err. A batch body answers the same way, with one HandlerOutcome for the whole slice or a Vec<HandlerOutcome> with one per element. A manual body always returns Result<(), HandlerOutcome>: Ok(()) acks, and Err(outcome) settles by that outcome. A retry() with no cap declared comes back for as long as the body keeps asking; the attempt cap ends that. Any outcome can carry work for after the settlement with and_after. A panic or a payload that does not decode is settled by the failure policy instead. A test reads the settlements with outcomes.

Example​

examples/handler_outcome.rs
/// Stands in for a stock service whose first reservation finds the shelf locked.
struct Warehouse {
    locked: AtomicBool,
}

impl Warehouse {
    fn try_reserve(&self, _sku: &str) -> bool {
        !self.locked.swap(false, Ordering::SeqCst)
    }
}

#[subscriber("orders")]
async fn reserve(order: &Order, ctx: &mut Context<'_, (), Arc<Warehouse>>) -> HandlerOutcome {
    if order.sku.is_empty() {
        // Nothing to reserve, and a redelivery would not change that.
        return HandlerOutcome::drop();
    }
    if !ctx.state().try_reserve(&order.sku) {
        // A passing condition: the broker brings the order back.
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Nothing to decide​

examples/handler_outcome.rs
/// A body with nothing to decide returns nothing, and the delivery is acked.
#[subscriber("orders")]
async fn audit(order: &Order) {
    tracing::info!(order.id, sku = %order.sku, "seen");
}

An error drops the delivery​

examples/handler_outcome.rs
#[derive(Debug)]
struct UnknownSku;

impl fmt::Display for UnknownSku {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("unknown sku")
    }
}

impl Error for UnknownSku {}

fn price_of(sku: &str) -> Result<u64, UnknownSku> {
    match sku {
        "lamp" => Ok(40),
        _ => Err(UnknownSku),
    }
}

/// `Ok` acks and `Err` drops, so `?` is how the body gives up on an order.
#[subscriber("orders")]
async fn price(order: &Order) -> Result<(), UnknownSku> {
    let cents = price_of(&order.sku)?;
    tracing::info!(order.id, cents, "priced");
    Ok(())
}

An outcome or an error​

examples/handler_outcome.rs
#[derive(Debug)]
struct UnknownSku;

impl fmt::Display for UnknownSku {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("unknown sku")
    }
}

impl Error for UnknownSku {}

fn stock_of(sku: &str) -> Result<u32, UnknownSku> {
    match sku {
        "lamp" => Ok(3),
        "vase" => Ok(0),
        _ => Err(UnknownSku),
    }
}

/// `Ok` carries the settlement the body chose, and `Err` drops.
#[subscriber("orders")]
async fn ship(order: &Order) -> Result<HandlerOutcome, UnknownSku> {
    if stock_of(&order.sku)? == 0 {
        // Restocking takes an hour: ask for the order back then.
        return Ok(HandlerOutcome::retry_after(Duration::from_secs(3600)));
    }
    Ok(HandlerOutcome::ack())
}