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

Macros

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()
}

Manual

examples/manual/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)
    }
}

struct Reserve;

impl Handle<Order, (), (), (), Arc<Warehouse>> for Reserve {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), Arc<Warehouse>>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.sku.is_empty() {
            // Nothing to reserve, and a redelivery would not change that.
            return ready(Err(HandlerOutcome::drop()));
        }
        if !ctx.state().try_reserve(&order.sku) {
            // A passing condition: the broker brings the order back.
            return ready(Err(HandlerOutcome::retry()));
        }
        ready(Ok(()))
    }
}

Nothing to decide

Macros

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");
}

Manual

examples/manual/handler_outcome.rs

/// A body with nothing to decide returns `Ok(())`, and the delivery is acked.
struct Audit;

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

An error drops the delivery

Macros

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(())
}

Manual

examples/manual/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),
    }
}

/// The body turns the error into a drop itself.
struct Price;

impl Handle<Order> for Price {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let priced = price_of(&order.sku).map(|cents| {
            tracing::info!(order.id, cents, "priced");
        });
        ready(priced.map_err(|_| HandlerOutcome::drop()))
    }
}

An outcome or an error

Macros

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())
}

Manual

examples/manual/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),
    }
}

/// The settlement the body chose travels as `Err(outcome)`, and the error becomes a drop.
struct Ship;

impl Handle<Order> for Ship {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        ready(match stock_of(&order.sku) {
            Err(_) => Err(HandlerOutcome::drop()),
            // Restocking takes an hour: ask for the order back then.
            Ok(0) => Err(HandlerOutcome::retry_after(Duration::from_secs(3600))),
            Ok(_) => Ok(()),
        })
    }
}