Transactions

AmqpTransactionalPublish, TransactionalPublish in the prelude, posts under an AMQP transaction: what a publisher sends between begin and commit becomes visible together, or not at all.

The policy needs the transaction feature. The runtime pairs it into an AmqpTxnPublisher, which implements TransactionalPublisher. A handler binds its slot as Out<impl TransactionalPublisher>, and the mount site names the policy with .out(DefaultSlot, TransactionalPublish).build(). begin_transaction declares a transaction on the broker. Each publish after it is posted under that transaction and waits for the broker's outcome, as a plain publish does. commit makes every posted message visible at once; abort discards them. A publish with no transaction open goes out at once.

The handle holds one transaction at a time. A second begin_transaction while one is open fails and leaves the open one untouched. commit or abort with nothing open fails too. A commit or an abort that fails has still ended the transaction, so the next begin_transaction starts a new one. Each failure is an AmqpError::Transaction.

The slot's publisher is paired once, at startup, so every delivery of the registration uses the same handle. A registration that runs deliveries side by side with workers(n) has them compete for the one transaction, and a second begin_transaction fails until the first ends.

A transaction covers what the publisher posts. The delivery the handler answers settles on its own after the handler returns, so a delivery redelivered after a commit posts its messages again.

Transactions bills an order with its ledger entries and invoice in one transaction.

Example

The dependency line:

ruststream-amqp = { version = "0.7", features = ["transaction"] }

The handler posts one invoice line per order line and commits them together; an empty line aborts the order.

Macros
Manual

examples/amqp_transaction.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "invoices")]
struct InvoiceLine {
    order_id: u64,
    sku: String,
    quantity: u32,
}

#[subscriber(AmqpAddress::queue("orders"))]
async fn bill(order: &Order, Out(invoices): Out<impl TransactionalPublisher>) -> HandlerOutcome {
    if invoices.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for line in &order.lines {
        // An empty line voids the order, and the abort takes back the lines posted before it.
        if line.quantity == 0 {
            tracing::warn!(order.id, sku = %line.sku, "empty line, order voided");
            return match invoices.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        let invoice_line = InvoiceLine {
            order_id: order.id,
            sku: line.sku.clone(),
            quantity: line.quantity,
        };
        if invoices.message(&invoice_line).publish().await.is_err() {
            let _ = invoices.abort().await;
            return HandlerOutcome::retry();
        }
    }
    match invoices.commit().await {
        Ok(()) => {
            tracing::info!(order.id, lines = order.lines.len(), "order billed");
            HandlerOutcome::ack()
        }
        Err(_) => HandlerOutcome::retry(),
    }
}

examples/manual/amqp_transaction.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "invoices")]
struct InvoiceLine {
    order_id: u64,
    sku: String,
    quantity: u32,
}

struct Bill;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Bill
where
    Entry: OutEntry<DefaultSlot, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let invoices = outs.get(DefaultSlot);
        invoices
            .begin_transaction()
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        for line in &order.lines {
            // An empty line voids the order, and the abort takes back the lines posted before it.
            if line.quantity == 0 {
                tracing::warn!(order.id, sku = %line.sku, "empty line, order voided");
                invoices
                    .abort()
                    .await
                    .map_err(|_| HandlerOutcome::retry())?;
                return Err(HandlerOutcome::drop());
            }
            let invoice_line = InvoiceLine {
                order_id: order.id,
                sku: line.sku.clone(),
                quantity: line.quantity,
            };
            if invoices.message(&invoice_line).publish().await.is_err() {
                let _ = invoices.abort().await;
                return Err(HandlerOutcome::retry());
            }
        }
        invoices
            .commit()
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        tracing::info!(order.id, lines = order.lines.len(), "order billed");
        Ok(())
    }
}

The mount site gives the slot the transactional policy.

Macros
Manual

examples/amqp_transaction.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            // The transactional policy is what gives the slot begin, commit and abort.
            b.include(bill)
                .out(DefaultSlot, TransactionalPublish)
                .build();
        },
    )
}

examples/manual/amqp_transaction.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            let bill = subscriber(AmqpAddress::queue("orders"), Bill);
            // The transactional policy is what gives the slot begin, commit and abort.
            b.include(bill.build())
                .out(DefaultSlot, TransactionalPublish)
                .build();
        },
    )
}

Example​

Example