Публикация «всё или ничего»

Это руководство публикует несколько сообщений из одного обработчика вместе или не публикует ни одного.

Транзакции входят в протокол AMQP 1.0: отправитель передаёт сообщения внутри транзакции, и брокер делает их видимыми при фиксации. Их включает фича transaction:

Cargo.toml

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

Пример - сервис биллинга. Он превращает каждый заказ в проводки и счёт, и до брокера доходят либо все они, либо ни одна.

Обработчик

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "billing.orders")]
struct Order {
    id: u64,
    lines: Vec<Line>,
}

#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Line {
    sku: String,
    cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "ledger")]
struct LedgerEntry {
    order: u64,
    sku: String,
    cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoices")]
struct Invoice {
    order: u64,
    total_cents: u64,
}

#[derive(OutSlot)]
#[publishes(LedgerEntry, Invoice)]
struct Billing;

#[subscriber(AmqpAddress::queue("billing.orders"))]
async fn bill(
    order: &Order,
    Out(billing): Out<impl TransactionalPublisher, Billing>,
) -> HandlerOutcome {
    if billing.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for line in &order.lines {
        if line.cents == 0 {
            tracing::warn!(
                order.id,
                line.sku,
                "a line without a price, nothing is billed"
            );
            // A failed abort has still ended the transaction, so there is nothing left to undo.
            let _ = billing.abort().await;
            return HandlerOutcome::drop();
        }
        let entry = LedgerEntry {
            order: order.id,
            sku: line.sku.clone(),
            cents: line.cents,
        };
        if billing.message(&entry).publish().await.is_err() {
            let _ = billing.abort().await;
            return HandlerOutcome::retry();
        }
    }
    let invoice = Invoice {
        order: order.id,
        total_cents: order.lines.iter().map(|line| line.cents).sum(),
    };
    if billing.message(&invoice).publish().await.is_err() {
        let _ = billing.abort().await;
        return HandlerOutcome::retry();
    }
    if billing.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, invoice.total_cents, "order billed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    lines: Vec<Line>,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "billing.orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

impl MessageInfo for Order {
    const NAME: &'static str = "Order";
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Line {
    sku: String,
    cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct LedgerEntry {
    order: u64,
    sku: String,
    cents: u64,
}

impl OutgoingDestination for LedgerEntry {
    type Form = FixedName;

    const DESTINATION: &'static str = "ledger";
}

impl MessageHeaders for LedgerEntry {
    type Contract = NoHeaders;
}

impl MessageInfo for LedgerEntry {
    const NAME: &'static str = "LedgerEntry";
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Invoice {
    order: u64,
    total_cents: u64,
}

impl OutgoingDestination for Invoice {
    type Form = FixedName;

    const DESTINATION: &'static str = "invoices";
}

impl MessageHeaders for Invoice {
    type Contract = NoHeaders;
}

impl MessageInfo for Invoice {
    const NAME: &'static str = "Invoice";
}

struct Billing;

impl OutSlot for Billing {
    const NAME: &'static str = "Billing";
    type Destination = Reads;
}

impl PublishedThrough<Billing> for LedgerEntry {}
impl PublishedThrough<Billing> for Invoice {}

struct Bill;

impl<Sent> Handle<Order, (), Outs<(Sent,)>> for Bill
where
    Sent: OutEntry<Billing, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Sent,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let billing = outs.get(Billing);
        if billing.begin_transaction().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        for line in &order.lines {
            if line.cents == 0 {
                tracing::warn!(
                    order.id,
                    line.sku,
                    "a line without a price, nothing is billed"
                );
                // A failed abort has still ended the transaction, so there is nothing left to undo.
                let _ = billing.abort().await;
                return Err(HandlerOutcome::drop());
            }
            let entry = LedgerEntry {
                order: order.id,
                sku: line.sku.clone(),
                cents: line.cents,
            };
            if billing.message(&entry).publish().await.is_err() {
                let _ = billing.abort().await;
                return Err(HandlerOutcome::retry());
            }
        }
        let invoice = Invoice {
            order: order.id,
            total_cents: order.lines.iter().map(|line| line.cents).sum(),
        };
        if billing.message(&invoice).publish().await.is_err() {
            let _ = billing.abort().await;
            return Err(HandlerOutcome::retry());
        }
        if billing.commit().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(order.id, invoice.total_cents, "order billed");
        Ok(())
    }
}

Слот Billing ограничен как Out<impl TransactionalPublisher, _>: тело обработчика требует транзакций и не называет брокер. begin_transaction открывает транзакцию на издателе слота. Каждая публикация после этого отправляется внутри транзакции, и брокер придерживает её. commit делает проводки и счёт видимыми одновременно. По заказу со строкой без цены нельзя выставить счёт, и abort отбрасывает отправленное, поэтому ничего не доходит до ledger и invoices.

Фиксация или прерывание, вернувшие ошибку, всё равно завершают транзакцию, поэтому следующий begin_transaction начинает новую. Если фиксация вернула ошибку, bill повторяет заказ.

Приложение

Макросы

src/main.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| {
            b.include(bill).out(Billing, TransactionalPublish).build();
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(subscriber(AmqpAddress::queue("billing.orders"), Bill).build())
                .out(Billing, TransactionalPublish)
                .build();
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

TransactionalPublish - политика с поддержкой транзакций, AmqpTransactionalPublish под именем из прелюдии. Её издатель держит одну транзакцию за раз. Второй begin_transaction при открытой транзакции возвращает ошибку и не трогает открытую. Каждая регистрация получает собственного издателя, поэтому два обработчика могут держать по транзакции одновременно. Публикация на этом издателе вне транзакции уходит сразу.

Транзакция, которая начинает работу сервиса и не обрабатывает никакую доставку, выполняется из хука запуска: b.after_startup(TransactionalPublish, ..) передаёт хуку того же издателя.

Доставка самого заказа

Транзакция охватывает то, что публикует обработчик. Доставка заказа урегулируется после возврата из обработчика, вне транзакции. Сбой между фиксацией и урегулированием доставляет заказ повторно, и его проводки и счёт публикуются второй раз. Дайте сообщениям идентификатор, по которому читатели отбросят повтор, например номер заказа, как здесь.

Тестирование транзакции

Тесты в файле примера запускают сервис внутри процесса. Брокер внутри процесса ничего не публикует до фиксации и отбрасывает сообщения прерванной транзакции, поэтому тесты читают исход по тому, что опубликовано. Руководство Тестирование сервиса на AMQP разбирает обвязку.