BatchPublishTransform

A batch publish transform is a step on the replies of a batch handler: each reply passes it on its own, with the batch's context.

A batch handler with a reply returns one Vec of replies per batch, and those replies pass their own steps. .batch_transform(..) adds one after .out_reply(policy), and a single-message mount rejects it at compile time. The per-message publish transforms do not run on batch replies, and batch transforms do not run on single replies. for_batch lifts a per-message transform onto the batch path, so one impl serves both. All replies of a batch see one PublishContext, the batch's: its name is the subscription, its headers are empty, and its context fields are the broker's batch context. A transform that reads the message it answers belongs on a single-message handler. A batch reply never offers the naming right. A batch transform has no macro form: the same impl serves both forms, and only the mounting differs.

Example

examples/batch_transform.rs

/// Marks every reply of a batch. It sees the batch's context, not one delivery's.
struct Batched;

impl<C, Options> BatchPublishTransform<C, Options> for Batched {
    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        _cx: &PublishContext<'_, C>,
    ) {
        out.headers_mut().insert("batched", "true");
    }
}

/// A per-message transform, reused on the batch path through `for_batch`.
struct Service(&'static str);

impl<K: ContextKind, Options> PublishTransform<K, Options> for Service {
    type Destination = Reads;

    fn apply(&self, out: &mut Outgoing<'_>, _options: &mut Option<Options>, _cx: &K::View<'_>) {
        out.headers_mut().insert("service", self.0);
    }
}

Macros

examples/batch_transform.rs

fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm.batch(nonzero!(64)))
            .out_reply(Publish)
            .batch_transform(Batched)
            .batch_transform(for_batch(Service("billing")));
    })
}

Manual

examples/manual/batch_transform.rs

fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber("orders", Confirm)
                .reply()
                .to("receipts")
                .batch(nonzero!(64))
                .build(),
        )
        .out_reply(Publish)
        .batch_transform(Batched)
        .batch_transform(for_batch(Service("billing")));
    })
}