Settle

Settle::AtMostOnce settles each delivery on receipt, so the message is gone whether the handler finished or not, and ack reports Unsupported.

AmqpAddress::settle sets the mode on the descriptor. Settle::AtLeastOnce is the default: the handler's outcome settles the delivery, as Dispositions describes. Settle::AtMostOnce accepts each delivery as it arrives, before the handler runs. The broker forgets the message at once, so a crash in the middle of the handler, a retry or a drop brings nothing back. The outcome the handler returns has nothing left to settle: the settlement reports AckError::Unsupported, and the runtime logs a warning that names the subscription. The mode suits traffic where a lost message costs less than a redelivered one, such as a reading that a sensor sends again a second later. The credit of an at-most-once subscription tops itself up as the deliveries arrive. In process, TestApp settles on receipt the same way, so a test sees a retry bring nothing back.

Example

Macros

examples/amqp_settlement_settle.rs

/// A sensor reports every second, so a lost reading costs nothing. Each delivery is settled on
/// receipt: a reading the store refused is not delivered again.
#[subscriber(AmqpAddress::queue("readings").settle(Settle::AtMostOnce))]
async fn record(reading: &Reading) -> HandlerOutcome {
    if let Err(err) = store(reading).await {
        tracing::warn!(reading.sensor, err, "reading lost");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/amqp_settlement_settle.rs

/// A sensor reports every second, so a lost reading costs nothing. Each delivery is settled on
/// receipt: a reading the store refused is not delivered again.
struct Record;

impl Handle<Reading> for Record {
    async fn handle(
        &self,
        reading: &Reading,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if let Err(err) = store(reading).await {
            tracing::warn!(reading.sensor, err, "reading lost");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn readings() -> AmqpAddress {
    AmqpAddress::queue("readings").settle(Settle::AtMostOnce)
}