Skip to main content

Publisher

ruststream_fred::prelude

Trait Publisher 

Source 
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§

Source

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.

Source

type Error: Error + Send + Sync + 'static

The error type returned by publish.

Source

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§

Source

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§

Source

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".

Implementors§