Transactions

A handler bounded Out<impl TransactionalPublisher> opens a transaction scope with begin, publishes through it, and ends it with commit or abort.

The scope's publishes become visible together at commit, and abort discards them. The handler names the capability, TransactionalPublisher, and the mount site names the policy that provides it. ConfirmsPublish, TransactionalPublish in the prelude, buffers in the client and commits by awaiting a broker confirm for every message. ServerTxPublish runs an AMQP channel transaction on the server. LapinPublish pairs into a publisher with no transaction surface, so a transactional handler mounted with it does not compile.

A publisher runs one such transaction at a time. A second begin while one is open, or a commit or abort with none open, fails with AmqpError::Transaction and leaves the open transaction as it was. A publisher is a shared handle: every clone uses one channel and one transaction state. Several independent groups of messages in one handler take the owned kind, ConfirmsTransaction.

Example

An order ships one message per item, in one transaction. An item without a name aborts the transaction, and the shipments already published are discarded.

Macros
Manual

examples/lapin_transactions_scope.rs
/// One shipment per item, in one transaction. An item without a name aborts it, and the shipments
/// already published are discarded with it.
#[subscriber(RabbitQueue::new("orders"))]
async fn ship(order: &Order, Out(shipments): Out<impl TransactionalPublisher>) -> HandlerOutcome {
    let Ok(scope) = shipments.begin().await else {
        return HandlerOutcome::retry();
    };
    for item in &order.items {
        if item.is_empty() {
            tracing::warn!(order.id, "an item without a name, nothing shipped");
            return match scope.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        let shipment = Shipment {
            order_id: order.id,
            item: item.clone(),
        };
        if scope.message(&shipment).publish().await.is_err() {
            return HandlerOutcome::retry();
        }
    }
    if scope.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/lapin_transactions_scope.rs
/// One shipment per item, in one transaction. An item without a name aborts it, and the shipments
/// already published are discarded with it.
struct Ship;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for Ship
where
    Egress: OutEntry<DefaultSlot, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let scope = outs
            .get(DefaultSlot)
            .begin()
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        for item in &order.items {
            if item.is_empty() {
                tracing::warn!(order.id, "an item without a name, nothing shipped");
                scope.abort().await.map_err(|_| HandlerOutcome::retry())?;
                return Err(HandlerOutcome::drop());
            }
            let shipment = Shipment {
                order_id: order.id,
                item: item.clone(),
            };
            if scope.message(&shipment).publish().await.is_err() {
                return Err(HandlerOutcome::retry());
            }
        }
        scope.commit().await.map_err(|_| HandlerOutcome::retry())
    }
}

Macros
Manual

examples/lapin_transactions_scope.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            // The confirms policy; `LapinPublish::default().server_tx()` is the other one.
            b.include(ship)
                .out(DefaultSlot, TransactionalPublish::default())
                .build();
        },
    )
}

examples/manual/lapin_transactions_scope.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            // The confirms policy; `LapinPublish::default().server_tx()` is the other one.
            b.include(subscriber(RabbitQueue::new("orders"), Ship).build())
                .out(DefaultSlot, TransactionalPublish::default())
                .build();
        },
    )
}

Example​

Example