Publish

AmqpPublish, Publish in the prelude, waits for the broker's disposition and reports anything but accepted as PublishNotAccepted.

The policy has no fields, so a mount site writes it bare. It is named wherever a registration publishes: .out_reply(Publish) for a reply, .out(Slot, Publish).build() for a slot, .out_retry(Publish) for retry copies. The policy holds no connection; the runtime pairs it with the connected broker at startup into an AmqpPublisher. It is also the broker's default policy, so a reply registration that names no policy publishes through it.

The publishers share one AMQP session. A sender link is attached to an address on its first publish and kept for the next ones. Each publish waits for the broker's outcome of the transfer. An accepted outcome completes it; rejected, released or modified fails it with the address and the outcome, so a refusal never passes as sent.

The message carries the encoded value as one data section. Headers ride the message's properties and application-properties sections, and the partition key rides group-id. A publish carries no per-message settings: the message goes out without the header section, so it is not durable and keeps the default priority and no time to live.

Where a reply goes is a property of its type: #[outgoing(name = "receipts")] fixes the address. A reply to an address the request names is the request-reply pattern.

Example

Macros
Manual

examples/amqp_publishing_publish.rs
/// The receipt names its own address, so the mount site adds nothing to it.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

#[subscriber(AmqpAddress::queue("orders"), reply)]
async fn issue_receipt(order: &Order) -> Receipt {
    tracing::info!(order.id, "receipt issued");
    Receipt { order_id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            // The reply waits for the broker's disposition; anything but `accepted` fails it.
            b.include(issue_receipt).out_reply(Publish);
        },
    )
}

examples/manual/amqp_publishing_publish.rs
/// The receipt names its own address, so the mount site adds nothing to it.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

struct IssueReceipt;

impl Handle<Order, Receipt> for IssueReceipt {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(order.id, "receipt issued");
        ready(Ok(Receipt { order_id: order.id }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            let issue_receipt = subscriber(AmqpAddress::queue("orders"), IssueReceipt);
            // The reply waits for the broker's disposition; anything but `accepted` fails it.
            b.include(issue_receipt.reply().build()).out_reply(Publish);
        },
    )
}

Example​

Example