Dispositions

A handler's outcome is the delivery's disposition: ack accepts, retry sends modified with delivery-failed, drop rejects.

In AMQP 1.0 a receiver settles each delivery with a disposition, and the broker acts on its outcome. ack sends accepted, and the broker forgets the message. retry sends modified with delivery-failed set and undeliverable-here cleared. The broker adds the failed attempt to the message's delivery-count and delivers it again, to this subscription or to another receiver on the address. A retry never sends released: that outcome says the delivery was not acted on, so the broker leaves the count as it was, and a cap on the attempts reads that count. drop sends rejected, which ends the delivery: the broker's dead-letter policy decides where it goes, and ActiveMQ Artemis moves it to its DLQ address by default. A delivery that is never settled, because the process stopped, stays on its link until the link detaches, and the broker then hands it to the next receiver. AMQP 1.0 has no delayed redelivery, so retry_after is the runtime's own copy. Under Settle::AtMostOnce the delivery is settled on receipt, and the outcome reaches no broker. In process, TestApp keeps the same rules: a retry comes back with one more counted attempt, and a drop removes the delivery.

Acknowledgement and delivery guarantees settles the deliveries of a warehouse.

Example

Macros

examples/amqp_settlement_dispositions.rs

/// A charged payment is accepted, a timed-out one goes back to the broker for another try, and a
/// malformed one is rejected.
#[subscriber(AmqpAddress::queue("payments"))]
async fn charge(payment: &Payment, State(gateway): State<PaymentGateway>) -> HandlerOutcome {
    if payment.amount_cents <= 0 {
        // No retry can fix it: the broker's dead-letter policy decides where it goes.
        return HandlerOutcome::drop();
    }
    match gateway.charge(payment).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(payment.id, err, "charge failed, retrying");
            HandlerOutcome::retry()
        }
    }
}

Manual

examples/manual/amqp_settlement_dispositions.rs

/// A charged payment is accepted, a timed-out one goes back to the broker for another try, and a
/// malformed one is rejected.
struct Charge;

impl Handle<Payment, (), (), (), AppState> for Charge {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if payment.amount_cents <= 0 {
            // No retry can fix it: the broker's dead-letter policy decides where it goes.
            return Err(HandlerOutcome::drop());
        }
        if let Err(err) = ctx.state().gateway.charge(payment).await {
            tracing::warn!(payment.id, err, "charge failed, retrying");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}