LapinPublish

LapinPublish, Publish in the prelude, is the fire-and-forget publish policy: a publish resolves once the frame is written, with no broker confirm.

The message name is the routing key, and the exchange belongs to the policy. By default that is the default exchange, where a routing key addresses the queue of the same name; exchange names another. The policy holds no connection. At startup the runtime pairs it with the connected broker into a LapinPublisher, which publishes on the connection's shared publish channel. A publisher kept past shutdown answers every publish with AmqpError::Closed.

A mount site names the policy wherever a registration publishes: .out_reply(Publish::default()) for a reply, .out(Slot, Publish::default()) for a slot, .out_retry(Publish::default()) for retry copies. It is also the broker's default policy, so a registration that names none replies and retries through it. Where a reply goes is a property of the reply type, as the core's replies page describes. A publish outside a handler goes through after_startup on the broker scope: once the subscriptions are open, the runtime pairs the policy and hands the hook the live publisher.

The policy holds the defaults of the per-message properties. confirms and server_tx move to the transactional policies and keep these settings.

Example

Every order is answered with a confirmation, routed under the name the reply type declares. Once its subscriptions are open, the service announces itself.

Macros

examples/lapin_publishing_publish.rs

/// Every order is answered with a confirmation, routed under the name its type declares.
#[subscriber(RabbitQueue::new("orders"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { order_id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(confirm).out_reply(Publish::default());
            // Once the subscriptions are open, the service announces itself.
            b.after_startup(Publish::default(), async move |publisher| {
                let started = Started {
                    service: "orders".into(),
                };
                publisher.message(&started).publish().await
            });
        },
    )
}

Manual

examples/manual/lapin_publishing_publish.rs

/// Every order is answered with a confirmation, routed under the name its type declares.
struct Confirm;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            let confirm = subscriber(RabbitQueue::new("orders"), Confirm);
            b.include(confirm.reply().build())
                .out_reply(Publish::default());
            // Once the subscriptions are open, the service announces itself.
            b.after_startup(Publish::default(), async move |publisher| {
                let started = Started {
                    service: "orders".into(),
                };
                publisher.message(&started).publish().await
            });
        },
    )
}