Policy

Publish::default().ordering_key(..) orders every message of one mount site under one key, a reply included.

The key belongs to the policy, so it applies to every publish of that slot or reply. A reply has no call site, so the policy is where a fixed key for replies goes. It suits a slot that belongs to one entity: one ledger, one device. A key named by the call or by the partition-key header wins over it. The AsyncAPI document reports the key in the message's binding.

Example

Every ledger entry goes out under the key ledger.

Macros

examples/pubsub_publishing_ordering_policy.rs

/// Books every order in the ledger.
#[subscriber("orders-workers", reply)]
async fn book(order: &Order) -> LedgerEntry {
    LedgerEntry { order_id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // The ledger is one sequence, so every entry goes out under one key.
            b.include(book)
                .out_reply(Publish::default().ordering_key("ledger"));
        },
    )
}

Manual

examples/manual/pubsub_publishing_ordering_policy.rs

/// Books every order in the ledger.
struct Book;

impl Handle<Order, LedgerEntry> for Book {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<LedgerEntry, HandlerOutcome>> {
        ready(Ok(LedgerEntry { order_id: order.id }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // The ledger is one sequence, so every entry goes out under one key.
            b.include(subscriber("orders-workers", Book).reply().build())
                .out_reply(Publish::default().ordering_key("ledger"));
        },
    )
}