Operations

A send operation says how it posts: confirmed for Publish, transactional for TransactionalPublish.

The extension of a send operation has one field, posting, and the publish policy at its mount site decides it. Publish posts confirmed: each transfer waits for the broker's outcome, and anything but accepted fails the publish. TransactionalPublish posts transactional: the transfers carry a transaction and become visible at its commit. A dead-letter destination has a send operation of its own, and it posts confirmed.

A reply has no send operation. What a reply policy adds to the document is on its reply.

Example

The operation send_events_orders_dead posts confirmed, and send_payments_ledger posts transactional.

Macros

examples/amqp_asyncapi.rs

fn service(broker: AmqpBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("billing", "1.0.0")).with_broker_labeled("amqp", broker, |b| {
        b.include(confirm).out_reply(Publish).transform(ToReplyTo);
        b.include(watch)
            .max_attempts(nonzero!(3u32))
            .dead_letter("orders.dead");
        b.include(audit);
        b.include(book).out(Ledger, TransactionalPublish).build();
    })
}

Manual

examples/manual/amqp_asyncapi.rs

fn service(broker: AmqpBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("billing", "1.0.0")).with_broker_labeled("amqp", broker, |b| {
        let confirm = subscriber(orders(), Confirm);
        b.include(confirm.reply().to("receipts").build())
            .out_reply(Publish)
            .transform(ToReplyTo);
        b.include(subscriber(events(), Watch).build())
            .max_attempts(nonzero!(3u32))
            .dead_letter("orders.dead");
        b.include(subscriber(audit_queue(), Audit).build());
        let book = subscriber(AmqpAddress::queue("payments"), Book);
        b.include(book.build())
            .out(Ledger, TransactionalPublish)
            .build();
    })
}