PublishTransform

A publish transform is a static step on one publish position: it reads the delivery or the slot, and changes the outgoing message.

.transform(..) adds one after the position it belongs to: after .out_reply(policy) for a reply, after .out(marker, policy) for a slot. Its apply gets the outgoing message, the broker's per-message settings and a view of the position, and returns nothing: it cannot refuse a publish. An impl declares three things. The kind is what it reads. ForReply<C> reads the delivery being answered through a PublishContext: its channel, headers and the broker's context fields. ForSlot reads a SlotContext, the slot's name and nothing else, because the body that publishes has already read its delivery. The Options parameter is the settings it writes: (), or the broker's own type, which mounts it only over that broker's publisher. type Destination is what it may do to the destination: Reads leaves it alone, and Names sets it. A transform that reads and sets nothing is generic over both, impl<K: ContextKind, Options>, and mounts anywhere.

Transforms run in the order the chain writes them, then the app's publish layers, then the send. The chain is one type, built when the code compiles, so a transform costs only its own code. A reply of a batch handler takes batch transforms instead. A transform has no macro form: the same impl serves both forms, and only the mounting differs.

Example

A reply transform carries the request's correlation id onto the reply.

examples/publish_transform.rs
/// Carries the request's correlation id onto its reply. `ForReply` reads the delivery being
/// answered, so this transform mounts on a reply position.
struct Correlate;

impl<C, Options> PublishTransform<ForReply<C>, Options> for Correlate {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        // The delivery's own buffer, shared rather than copied.
        if let Some(id) = cx.headers().get_shared("correlation-id") {
            out.headers_mut().insert("correlation-id", id);
        }
    }
}

A slot transform stamps a constant on what leaves the slot.

examples/publish_transform.rs
/// Stamps the schema version on what leaves a slot. `ForSlot` reads the slot's name and nothing
/// else: the body that publishes has already read its delivery.
struct SchemaVersion(&'static str);

impl<Options> PublishTransform<ForSlot, Options> for SchemaVersion {
    type Destination = Reads;

    fn apply(&self, out: &mut Outgoing<'_>, _options: &mut Option<Options>, _cx: &SlotContext<'_>) {
        out.headers_mut().insert("schema-version", self.0);
    }
}

Macros
Manual

examples/publish_transform.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // A transform follows the position it belongs to.
        b.include(confirm).out_reply(Publish).transform(Correlate);
        b.include(audit)
            .out(Trail, Publish)
            .transform(SchemaVersion("2"))
            .build();
    })
}

examples/manual/publish_transform.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // A transform follows the position it belongs to.
        b.include(subscriber("orders", Confirm).reply().to("receipts").build())
            .out_reply(Publish)
            .transform(Correlate);
        b.include(subscriber("payments", AuditPayment).build())
            .out(Trail, Publish)
            .transform(SchemaVersion("2"))
            .build();
    })
}

Example​

Example