Transform

A reply transform over PubSubPublishOptions sets the reply's ordering key per delivery, read off the request.

A reply has no call site, so a key that differs per reply is a transform's to write. The transform implements PublishTransform<ForReply<C>, PubSubPublishOptions> and reads the delivery being answered. It writes the key into the message's settings, where the ordering_key step would. Naming the settings type ties the transform to Pub/Sub publishers: it mounts on no other broker. The mount site adds it after the reply policy with .transform(..).

Example

Each receipt goes out under the key of the order it answers.

Macros

examples/pubsub_publishing_ordering_transform.rs

/// Sends each receipt under the key of the order it answers, which the delivery reports in its
/// `partition-key` header.
struct ReplyUnderTheOrdersKey;

impl<C> PublishTransform<ForReply<C>, PubSubPublishOptions> for ReplyUnderTheOrdersKey {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PubSubPublishOptions>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(key) = cx.headers().get_str(PARTITION_KEY_HEADER) {
            options
                .get_or_insert_with(PubSubPublishOptions::default)
                .ordering_key = Some(key.to_owned());
        }
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(receipt)
                .out_reply(Publish::default())
                .transform(ReplyUnderTheOrdersKey);
        },
    )
}

Manual

examples/manual/pubsub_publishing_ordering_transform.rs

/// Sends each receipt under the key of the order it answers, which the delivery reports in its
/// `partition-key` header.
struct ReplyUnderTheOrdersKey;

impl<C> PublishTransform<ForReply<C>, PubSubPublishOptions> for ReplyUnderTheOrdersKey {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PubSubPublishOptions>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(key) = cx.headers().get_str(PARTITION_KEY_HEADER) {
            options
                .get_or_insert_with(PubSubPublishOptions::default)
                .ordering_key = Some(key.to_owned());
        }
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(subscriber("orders-workers", ReceiptFor).reply().build())
                .out_reply(Publish::default())
                .transform(ReplyUnderTheOrdersKey);
        },
    )
}