ConfirmsTransaction

ConfirmsTransaction is an owned confirm transaction: a value that holds its own publish buffer.

A handler bounded Out<impl OwnedTransactions, Slot> opens one with transaction() on its slot, and every call opens another. Any number can be open on one publisher, and settling one never touches another. The publisher keeps publishing directly while they are open. commit publishes the buffer in order and awaits every confirm, and abort discards it without touching the broker. Both consume the value, so a double commit or a publish after settling does not compile. A failed commit loses the buffer, and the delivery's redelivery is the way to recover. Only ConfirmsPublish offers the owned kind.

Example

The shipment and the audit record of an order go out in two transactions. An order out of stock aborts the shipment and still commits its audit record.

Macros
Manual

examples/lapin_transactions_owned.rs
#[derive(OutSlot)]
#[publishes(Shipment, AuditRecord)]
struct Records;

/// The shipment and the audit record go out in two transactions. The shipment waits in its own
/// buffer while the stock is checked: an order out of stock aborts it and still commits its audit
/// record.
#[subscriber(RabbitQueue::new("orders"))]
async fn record(
    order: &Order,
    Out(records): Out<impl OwnedTransactions, Records>,
) -> HandlerOutcome {
    let (Ok(mut shipment), Ok(mut audit)) =
        (records.transaction().await, records.transaction().await)
    else {
        return HandlerOutcome::retry();
    };
    let shipped = Shipment { order_id: order.id };
    if shipment.message(&shipped).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    let settled = if order.in_stock {
        shipment.commit().await
    } else {
        shipment.abort().await
    };
    if settled.is_err() {
        return HandlerOutcome::retry();
    }
    let entry = AuditRecord {
        order_id: order.id,
        shipped: order.in_stock,
    };
    if audit.message(&entry).publish().await.is_err() || audit.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(record)
                .out(Records, TransactionalPublish::default())
                .build();
        },
    )
}

examples/manual/lapin_transactions_owned.rs
#[derive(OutSlot)]
#[publishes(Shipment, AuditRecord)]
struct Records;

/// The shipment and the audit record go out in two transactions. The shipment waits in its own
/// buffer while the stock is checked: an order out of stock aborts it and still commits its audit
/// record.
struct Record;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for Record
where
    Egress: OutEntry<Records, Wire: OwnedTransactions>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let records = outs.get(Records);
        let (Ok(mut shipment), Ok(mut audit)) =
            (records.transaction().await, records.transaction().await)
        else {
            return Err(HandlerOutcome::retry());
        };
        let shipped = Shipment { order_id: order.id };
        if shipment.message(&shipped).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        let settled = if order.in_stock {
            shipment.commit().await
        } else {
            shipment.abort().await
        };
        if settled.is_err() {
            return Err(HandlerOutcome::retry());
        }
        let entry = AuditRecord {
            order_id: order.id,
            shipped: order.in_stock,
        };
        if audit.message(&entry).publish().await.is_err() || audit.commit().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("orders"), Record).build())
                .out(Records, TransactionalPublish::default())
                .build();
        },
    )
}

Example​

Example