ConfirmsPublish

ConfirmsPublish, TransactionalPublish in the prelude, is the publish policy whose every publish resolves once the broker has confirmed the message.

LapinPublish::confirms reaches it with the same settings: exchange and property defaults. It pairs into a ConfirmsPublisher with a confirm channel of its own. A message the broker refuses fails the publish, and the handler decides what follows. RabbitMQ confirms a message once it has routed it, and a message that matches no queue is confirmed and dropped. So a confirm says the broker took the message, not that a queue holds it.

Inside a transaction the publisher buffers in the client. commit publishes the buffer in order and resolves once every message is confirmed, and abort discards the buffer without touching the broker. A failed commit loses the buffer, and the delivery's redelivery is the way to recover. The policy offers the borrowed kind of the transaction scope and the owned kind of ConfirmsTransaction.

Example

A payment is acknowledged once the broker has confirmed its receipt. A receipt the broker refused brings the payment back.

Macros
Manual

examples/lapin_transactions_confirms.rs
/// The payment is acknowledged once the broker has confirmed its receipt. A receipt the broker
/// refused brings the payment back.
#[subscriber(RabbitQueue::new("payments"))]
async fn settle(payment: &Payment, Out(receipts): Out<impl Publisher>) -> HandlerOutcome {
    let receipt = Receipt {
        payment_id: payment.id,
    };
    if let Err(err) = receipts.message(&receipt).publish().await {
        tracing::warn!(payment.id, %err, "receipt not confirmed");
        return HandlerOutcome::retry();
    }
    tracing::info!(payment.id, payment.cents, "payment settled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(settle)
            .out(DefaultSlot, TransactionalPublish::default())
            .build();
    })
}

examples/manual/lapin_transactions_confirms.rs
/// The payment is acknowledged once the broker has confirmed its receipt. A receipt the broker
/// refused brings the payment back.
struct Settle;

impl<Egress> Handle<Payment, (), Outs<(Egress,)>> for Settle
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        payment: &Payment,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let receipt = Receipt {
            payment_id: payment.id,
        };
        let receipts = outs.get(DefaultSlot);
        if let Err(err) = receipts.message(&receipt).publish().await {
            tracing::warn!(payment.id, %err, "receipt not confirmed");
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(payment.id, payment.cents, "payment settled");
        Ok(())
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(RabbitQueue::new("payments"), Settle).build())
            .out(DefaultSlot, TransactionalPublish::default())
            .build();
    })
}

Example​

Example