pub trait Publisher: Send + Sync {
type Payload: PayloadForm;
type Error: Error + Send + Sync + 'static;
type Options: Clone + Send + Sync + 'static;
// Required method
fn publish(
&self,
msg: OutgoingMessage<'_, <Self::Payload as PayloadForm>::Form<'_>>,
options: Option<&Self::Options>,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
// Provided method
fn base_headers(&self) -> Option<&HeaderMap> { ... }
}Expand description
A producer that sends messages into the broker.
Publisher is Send + Sync so a single instance can be shared across tasks. Implementations
are expected to be cheap to clone; expensive shared state (connection pool, batch buffers)
should live behind an Arc.
§Examples
use ruststream::{OutgoingMessage, Publisher};
async fn emit<P: Publisher>(publisher: &P) -> Result<(), P::Error> {
// Bytes the caller holds, in whichever form this publisher declared.
let msg = OutgoingMessage::new("orders.created", b"{}".as_slice());
publisher.publish(msg, None).await
}Required Associated Types§
Sourcetype Payload: PayloadForm
type Payload: PayloadForm
How this transport consumes the payload: Lend where it reads the bytes and keeps
nothing, Take where its client keeps them.
The declaration decides what publish receives - &[u8] under Lend, a
BytesMut of its own under Take - and what the framework does with its own buffers
above it: a lending publisher inside a dispatch loop is lent the one encode buffer that
loop reuses, so a reply through it allocates nothing per message, while a taking one is
handed a fresh buffer it may keep. Declare Lend unless the client keeps the bytes past
the call.
Sourcetype Options: Clone + Send + Sync + 'static
type Options: Clone + Send + Sync + 'static
The broker’s per-message settings: a QoS, a priority, an ordering key, whatever this
transport lets one message differ from the next in.
Every field is optional, because a publish carries only what its call site adjusted: the
rest keeps what the PublishPolicy fixed when it paired this publisher. A broker with
no per-message setting writes type Options = ();.
The type is the broker’s own, and so are the builder steps that fill it: a broker ships an
extension trait over PublishBuilder bounded on this
type, so its steps appear on a builder over its own publisher and nowhere else.
Clone and 'static are what the test harness needs: it copies the options a slot
publish carried and hands them back to the test as this type
(tb.out::<Marker>().with_options(..) under the testing feature).
Required Methods§
Sourcefn publish(
&self,
msg: OutgoingMessage<'_, <Self::Payload as PayloadForm>::Form<'_>>,
options: Option<&Self::Options>,
) -> impl Future<Output = Result<(), Self::Error>> + Send
fn publish( &self, msg: OutgoingMessage<'_, <Self::Payload as PayloadForm>::Form<'_>>, options: Option<&Self::Options>, ) -> impl Future<Output = Result<(), Self::Error>> + Send
Publishes a message to the broker, with the per-message settings the call site adjusted.
This is the contract a broker implements, and the direct call a broker crate used on its
own - without this one - is written against. Inside a service built on ruststream it is
the layer underneath: what a handler sends goes through the publish builder
(message), which resolves the destination, the
codec and the headers and assembles the
OutgoingMessage itself. Reach for this one where the message is already built: a
publish transform, a middleware, a post-settle hook.
options is None on every path with no call site to adjust them - a reply, a deferred
redelivery - and the policy’s own settings apply.
The message arrives by value, and so do its payload and its header map. Read them with
OutgoingMessage::payload and OutgoingMessage::headers; a transport that consumes
the message takes its parts with OutgoingMessage::into_parts - the destination, the
payload and the map in one move, nothing copied.
What the payload is follows Payload. A Lend publisher is handed
&[u8] valid for the length of the call: read it, write it into your frame, and hold
nothing afterwards. A Take publisher is handed the BytesMut the framework wrote,
as it was written, so Vec::from(payload) and BytesMut::freeze turn it into what the
client speaks with no copy of the body; only a publish lending bytes the framework does
not own is copied into that buffer.
§Cancel safety
Cancel safety is implementation-defined: most brokers will leave a message in an indeterminate state if the future is dropped mid-flight. Implementors must document the guarantees their broker provides.
§Errors
Returns Self::Error when the broker rejects the message, the connection is lost, or
the operation times out.
Provided Methods§
Sourcefn base_headers(&self) -> Option<&HeaderMap>
fn base_headers(&self) -> Option<&HeaderMap>
The headers this publisher contributes to every message it sends, underneath whatever the message itself names.
None by default: a plain broker publisher contributes nothing and the outgoing map starts
empty. This is the place for a constant of the publisher itself - a tenant, a producer
name, a schema id every message of this handle carries - laid down first, with the
message’s own headers written over it key by key, so the call site wins over the handle.
A delivery setting is not one of those: it belongs to the message, and a broker carries it
as a field of Options that the policy defaults and the call site
adjusts.
Every message the runtime sends through this publisher starts from that base: a publish
through the builder, and the reply a reply("dest") handler returns (whose
PublishTransform stack then writes over it, like any
other call site). A publish a handler body issues on a value it obtained outside the
framework is that body’s own call and reaches nothing here.
The map is borrowed, never rebuilt per publish, so a handle that has one keeps it in its own state.
§Examples
use ruststream::memory::MemoryBroker;
use ruststream::runtime::PublishExt;
use ruststream::{HeaderMap, Outgoing, OutgoingFor, Publisher, Serialized};
// A handle that tags every message it sends, without touching the message itself.
struct Tenanted<P>(P, HeaderMap);
impl<P: Publisher> Publisher for Tenanted<P> {
type Payload = P::Payload;
type Error = P::Error;
type Options = P::Options;
async fn publish(
&self,
msg: OutgoingFor<'_, Self::Payload>,
options: Option<&Self::Options>,
) -> Result<(), Self::Error> {
self.0.publish(msg, options).await
}
fn base_headers(&self) -> Option<&HeaderMap> {
Some(&self.1)
}
}
// Bytes that are already the payload, so this example needs no codec feature.
#[derive(Outgoing, Serialized)]
struct Order(Vec<u8>);
let broker = MemoryBroker::new();
let base = [("tenant", "acme")].into_iter().collect();
let publisher = Tenanted(broker.publisher(), base);
publisher.message(&Order(b"{}".to_vec())).to("orders").publish().await?;Dyn Compatibility§
This trait is not dyn compatible.
In older versions of Rust, dyn compatibility was called "object safety".