ServerTxPublish

ServerTxPublish is the publish policy backed by AMQP channel transactions: tx.select, tx.commit and tx.rollback.

LapinPublish::server_tx reaches it with the same settings. It pairs into a ServerTxPublisher on a channel of its own. Between begin and commit the messages wait on the broker inside the channel transaction, and they become visible together at commit. abort rolls them back on the server. Each commit is a synchronous round trip, which makes the policy slower than ConfirmsPublish. Outside a transaction a publish behaves like a fire-and-forget one. The transaction is the channel's state, so the policy offers only the borrowed kind of the transaction scope.

Example

The debit and the credit of a transfer are published in one server transaction. A frozen account rolls it back, and the debit already sent is discarded on the broker.

Macros

examples/lapin_transactions_server_tx.rs

/// The debit and the credit of a transfer are published in one server transaction. A frozen
/// account rolls it back, and the debit already sent is discarded on the broker.
#[subscriber(RabbitQueue::new("transfers"))]
async fn book(
    transfer: &Transfer,
    Out(ledger): Out<impl TransactionalPublisher>,
) -> HandlerOutcome {
    let Ok(scope) = ledger.begin().await else {
        return HandlerOutcome::retry();
    };
    let legs = [
        (transfer.from, -transfer.cents),
        (transfer.to, transfer.cents),
    ];
    for (account, cents) in legs {
        if account == FROZEN {
            tracing::warn!(account, "frozen account, transfer rolled back");
            return match scope.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        if scope
            .message(&Entry { account, cents })
            .publish()
            .await
            .is_err()
        {
            return HandlerOutcome::retry();
        }
    }
    if scope.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

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

Manual

examples/manual/lapin_transactions_server_tx.rs

/// The debit and the credit of a transfer are published in one server transaction. A frozen
/// account rolls it back, and the debit already sent is discarded on the broker.
struct Book;

impl<Egress> Handle<Transfer, (), Outs<(Egress,)>> for Book
where
    Egress: OutEntry<DefaultSlot, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        transfer: &Transfer,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let scope = outs
            .get(DefaultSlot)
            .begin()
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        let legs = [
            (transfer.from, -transfer.cents),
            (transfer.to, transfer.cents),
        ];
        for (account, cents) in legs {
            if account == FROZEN {
                tracing::warn!(account, "frozen account, transfer rolled back");
                scope.abort().await.map_err(|_| HandlerOutcome::retry())?;
                return Err(HandlerOutcome::drop());
            }
            if scope
                .message(&Entry { account, cents })
                .publish()
                .await
                .is_err()
            {
                return Err(HandlerOutcome::retry());
            }
        }
        scope.commit().await.map_err(|_| HandlerOutcome::retry())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("transfers"), Book).build())
                .out(DefaultSlot, ServerTxPublish::default())
                .build();
        },
    )
}