Skip to main content

Reliable publishing

This guide chooses how sure a service is that RabbitMQ took its messages, and sets the properties of one message.

The guarantee is a property of the publish policy at the mount site, not a flag on each call:

PolicyPrelude nameA publish resolves when
LapinPublishPublishthe frame is written to the connection
ConfirmsPublishTransactionalPublishRabbitMQ confirmed the message
ServerTxPublishServerTxPublishthe channel transaction commits

The example is an orders service. It turns an order into one shipment per parcel, all of them or none, and books a courier for each shipment, an expedited one ahead of the rest.

Confirmed transactions​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
    parcels: Vec<String>,
    expedited: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: u64,
    parcel: String,
    expedited: bool,
}

/// Publishes one shipment per parcel: every one of them, or none.
async fn ship_all(
    shipments: &impl TransactionalPublisher,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let scope = shipments.begin().await?;
    for parcel in &order.parcels {
        if parcel.is_empty() {
            scope.abort().await?;
            return Err("a parcel without a description".into());
        }
        let shipment = Shipment {
            order_id: order.id,
            parcel: parcel.clone(),
            expedited: order.expedited,
        };
        scope.message(&shipment).publish().await?;
    }
    // Resolves once RabbitMQ confirmed every shipment of the order.
    scope.commit().await?;
    Ok(())
}

#[subscriber("orders")]
async fn pack_order(
    order: &Order,
    Out(shipments): Out<impl TransactionalPublisher>,
) -> HandlerOutcome {
    match ship_all(shipments, order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, %err, "order not shipped");
            HandlerOutcome::drop()
        }
    }
}

pack_order bounds its slot Out<impl TransactionalPublisher>, so it mounts only on a policy with transactions. begin opens a scope, and every shipment published in it waits in the service. commit publishes them and resolves once RabbitMQ has confirmed each one. A parcel without a description aborts the scope, and no shipment of that order leaves. A commit that fails returns an error, and the order can be delivered again and packed again.

RabbitMQ confirms a message once it has taken responsibility for it: written to disk for a persistent message on a durable queue, accepted by a quorum queue's majority. RabbitMQ's confirms guide describes when each message is confirmed.

A confirm does not mean a queue received the message. RabbitMQ confirms a message that matches no binding and drops it, because the crate publishes without the mandatory flag. So a confirmed publish goes to an exchange whose bindings exist, or to a queue that exists: declare them with the service (Topology), or check them where operations own them.

Channel transactions​

ServerTxPublish, reached with Publish::default().server_tx(), uses AMQP channel transactions instead. The messages of a transaction become visible together at commit, which is a synchronous round trip to the server. Confirms are faster and cover most cases. Take a channel transaction when the consumers must never see half of a group, for example both legs of a transfer.

A handler bounded Out<impl TransactionalPublisher> mounts on either policy unchanged. A ConfirmsTransaction, opened with owned_transaction, owns its buffer instead of borrowing the publisher: several can be open at once, and each commits on its own. Channel transactions offer only the borrowed scope, because the transaction is the state of the channel itself.

Per-message properties​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "pickups")]
struct Pickup {
    order_id: u64,
    parcel: String,
}

/// The slot pickups leave through.
#[derive(OutSlot)]
#[publishes(Pickup)]
struct Pickups;

/// An expedited pickup jumps the couriers' queue and is worthless after an hour.
#[subscriber(RabbitQueue::new("shipments"))]
async fn book_courier(
    shipment: &Shipment,
    Out(pickups): Out<impl Publisher<Options = LapinPublishOptions>, Pickups>,
) -> HandlerOutcome {
    let pickup = Pickup {
        order_id: shipment.order_id,
        parcel: shipment.parcel.clone(),
    };
    let sent = if shipment.expedited {
        pickups
            .message(&pickup)
            .priority(9)
            .expiration(Duration::from_hours(1))
            .publish()
            .await
    } else {
        pickups.message(&pickup).publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

AMQP gives each message three properties of its own: the priority, the expiration and the delivery mode. The policy sets what every message through it carries, here priority 3. The publish builder's steps set them for one message: an expedited pickup goes out at priority 9 and expires after an hour unread. The slot is bounded on LapinPublishOptions, so the handler names no publisher type; the steps come from the crate's prelude.

None of the three travels as a header: each is an AMQP property, where RabbitMQ reads it. A delivery reports them back as headers, under PRIORITY_HEADER and EXPIRATION_HEADER.

A priority orders deliveries only on a queue declared with x-max-priority; on any other queue RabbitMQ ignores it. The couriers' pickups queue is declared with that argument by whoever owns it. Every policy publishes persistent messages unless persistent(false) says otherwise, so a message on a durable queue survives a broker restart.

The mount site​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Confirms: a publish resolves once RabbitMQ has taken the message.
        b.include(pack_order)
            .out(DefaultSlot, TransactionalPublish::default())
            .build();
        // Every pickup leaves at priority 3 unless the handler says otherwise.
        b.include(book_courier)
            .out(Pickups, Publish::default().priority(3))
            .build();
    })
}

TransactionalPublish::default() is ConfirmsPublish. Publish::default().priority(3) is the plain policy with a priority for every message. .confirms() and .server_tx() move a plain policy to the stronger ones and keep its settings, such as its exchange. Fire-and-forget publishes share one channel of the connection; a confirms or transactional publisher opens a channel of its own, because both modes are channel state.

A service publishes outside a handler too, for example from an HTTP endpoint, through a publisher the app hands out at startup (LapinPublish).

Testing​

The tests in the example file check that an order ships every parcel, that a bad parcel ships none, and which properties each pickup carries, read with tb.out::<Pickups>(). The in-process RabbitMQ buffers a transaction until it commits and frames each message as a server would. What a confirm means and the atomicity of a channel transaction are the server's, so they belong to a live test. Testing a RabbitMQ service covers the harness.