Acknowledgement

ack acknowledges a Pulsar delivery, retry negatively acknowledges it, and drop acknowledges it too: Pulsar has no terminal reject.

The handler's outcome settles each delivery on its subscription. An acknowledged message moves the subscription's cursor past it, and the subscription does not deliver it again. A negative acknowledgement hands the message back to the subscription, not to the consumer that gave up on it: on a Shared subscription another consumer may take it. A message that should leave the subscription for good goes through a dead-letter policy, which drop does not need. The client queues acknowledgements and sends them in the background, so a settlement that returns is queued on the consumer, not yet confirmed by the broker. A message nobody settles comes back after the subscription's ack_timeout. retry_after asks for the delivery back after a delay.

Acknowledgement settles the stock reservations of an inventory service by the handler's outcome.

Example

Macros
Manual

examples/pulsar_settlement_ack.rs
/// Reserves the stock of each order. `ack` acknowledges the message; `retry` negatively
/// acknowledges it and the subscription delivers it again; `drop` acknowledges it as well, so an
/// order of nothing is never seen again.
#[subscriber(PulsarSubscription::new("orders", "fulfilment"))]
async fn fulfil(order: &Order, State(warehouse): State<Warehouse>) -> HandlerOutcome {
    if order.quantity == 0 {
        tracing::warn!(order.id, "an order of nothing is dropped");
        return HandlerOutcome::drop();
    }
    match warehouse.reserve(order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, %err, "reservation failed, redelivering");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/pulsar_settlement_ack.rs
/// Reserves the stock of each order. `Ok` acknowledges the message; `Err(retry)` negatively
/// acknowledges it and the subscription delivers it again; `Err(drop)` acknowledges it as well,
/// so an order of nothing is never seen again.
struct Fulfil;

impl Handle<Order, (), (), (), AppState> for Fulfil {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if order.quantity == 0 {
            tracing::warn!(order.id, "an order of nothing is dropped");
            return Err(HandlerOutcome::drop());
        }
        ctx.state().warehouse.reserve(order).await.map_err(|err| {
            tracing::warn!(order.id, %err, "reservation failed, redelivering");
            HandlerOutcome::retry()
        })
    }
}

Example​

Example