pub trait PublishTransform<K: ContextKind, Options = ()>: Send + Sync {
type Destination: DestinationUse;
// Required method
fn apply(
&self,
out: &mut Outgoing<'_>,
options: &mut Option<Options>,
cx: &K::View<'_>,
);
}Expand description
A static, compile-time publish transform: mutates an Outgoing before it is sent, with read
access to whatever its position hands it.
The publish-side counterpart to the consume-side Layer: zero-cost composition,
no dyn dispatch. Composed onto a position by the .transform(..) step of a mount site’s
chain. Use for
per-destination transforms that belong to the publisher itself - a Confluent / Avro envelope, a
fixed content-type header, or stamping the delivery’s trace / correlation id onto the reply
(read it from cx). For cross-cutting
observation across every publish (metrics), use the app-wide PublishLayer via
RustStream::publish_layer instead; the per-publisher
transforms run first (closest to the value), then the app-wide publish pipeline, then the send.
§Which positions a transform mounts on
K is the position’s ContextKind, and writing the impl is how a transform says where it
belongs. A transform that reads nothing is generic over the kind and over the options, and
mounts anywhere:
use ruststream::runtime::{ContextKind, Outgoing, PublishTransform, Reads};
struct Envelope;
impl<K: ContextKind, Options> PublishTransform<K, Options> for Envelope {
type Destination = Reads;
fn apply(
&self,
out: &mut Outgoing<'_>,
_options: &mut Option<Options>,
_cx: &K::View<'_>,
) {
out.headers_mut().insert("x-envelope", b"1".to_vec());
}
}One that reads the delivery names ForReply, and naming it on a slot is then a compile error
at the mount site:
use ruststream::runtime::{ForReply, Outgoing, PublishContext, PublishTransform, Reads};
struct StampSource;
impl<C, Options> PublishTransform<ForReply<C>, Options> for StampSource {
type Destination = Reads;
fn apply(
&self,
out: &mut Outgoing<'_>,
_options: &mut Option<Options>,
cx: &PublishContext<'_, C>,
) {
out.headers_mut().insert("x-source", cx.name().as_bytes().to_vec());
}
}§What a transform writes
Options is the broker’s per-message settings type (Publisher::Options),
and the position is None until something fills it: the publish policy’s own settings are
what apply. A transform that sets one writes the broker’s type into its impl, so it mounts
over that broker’s publisher and nowhere else:
use ruststream::runtime::{ForSlot, Outgoing, PublishTransform, Reads, SlotContext};
/// What one broker lets a single message differ in.
#[derive(Clone, Default)]
struct Priority {
level: Option<u8>,
}
struct Urgent;
impl PublishTransform<ForSlot, Priority> for Urgent {
type Destination = Reads;
fn apply(
&self,
_out: &mut Outgoing<'_>,
options: &mut Option<Priority>,
_cx: &SlotContext<'_>,
) {
options.get_or_insert_with(Priority::default).level = Some(9);
}
}On a slot the position starts as a copy of what the call site’s own builder steps set, so a
transform completes the call rather than replacing it: it reads a field the call left alone
and overrides one it filled. A reply has no call site, so there the position starts at None.
§What a transform may do to the destination
Destination is the third half of the declaration, beside the view and
the options: a
transform that leaves the destination alone declares Reads, one that names it declares
Names. Stable Rust has no default for an associated type, so every impl writes the line -
including the three above, which declare Reads.
A position offers the right to name only where nothing has declared the destination already:
a reply type that leaves it open, a slot whose whole #[publishes(..)] dictionary leaves it
open. Where it is not on offer, mounting a naming transform is a compile error at the mount
site, which is what keeps a declaration and the wire in step. Outgoing::set_name stays a
plain method; what is checked is the declaration, not the call.
A destination the delivery already carries travels as the delivery’s own buffer:
HeaderMap::get_shared hands the header value over as a
counted handle, Str::try_from checks it is UTF-8, and set_name
takes the result. Answering where the request asked therefore costs a reference count, not a copy of the
address.
use ruststream::Str;
use ruststream::runtime::{ForReply, Names, Outgoing, PublishContext, PublishTransform};
/// Answers where the request asked to be answered.
struct ReplyTo;
impl<C, Options> PublishTransform<ForReply<C>, Options> for ReplyTo {
type Destination = Names;
fn apply(
&self,
out: &mut Outgoing<'_>,
_options: &mut Option<Options>,
cx: &PublishContext<'_, C>,
) {
if let Some(to) = cx.headers().get_shared("reply-to")
&& let Ok(to) = Str::try_from(to)
{
out.set_name(to);
}
}
}Required Associated Types§
Sourcetype Destination: DestinationUse
type Destination: DestinationUse
Required Methods§
Sourcefn apply(
&self,
out: &mut Outgoing<'_>,
options: &mut Option<Options>,
cx: &K::View<'_>,
)
fn apply( &self, out: &mut Outgoing<'_>, options: &mut Option<Options>, cx: &K::View<'_>, )
Transforms out in place before it is sent, reading the position’s view through cx and
resolving the broker’s per-message settings through options.
options is None while nothing has set a field: the publish policy’s own settings
apply. Set one with options.get_or_insert_with(Default::default).field = Some(..).
Dyn Compatibility§
This trait is dyn compatible.
In older versions of Rust, dyn compatibility was called "object safety".