Writing a broker¶

A broker is an independent crate that implements the core traits. It depends on ruststream with default features off, so it pulls in the trait surface and runtime without the bundled JSON codec or any other broker:

[dependencies]
ruststream = { version = "0.7", default-features = false }

This page is the contract. Implement the required traits, expose your own Config, add capability traits for the features your broker supports, and prove the result with the conformance harness. For a complete implementation on a real client, see the worked NATS example.

The required traits¶

`Broker` and `ConnectedBroker`¶

The broker is pure lifecycle: each state is a distinct type, and a transition consumes self and returns the next state, so calls made out of order do not compile. The broker names neither a subscriber type nor a publisher type, so one application can mix brokers of different kinds.

pub trait Broker: Send + Sync + Sized {
    type Error: std::error::Error + Send + Sync + 'static;
    type Connected: ConnectedBroker;
    async fn connect(self) -> Result<Self::Connected, Self::Error>;
}

pub trait ConnectedBroker: Send + Sync + Sized + 'static {
    type Error: std::error::Error + Send + Sync + 'static;
    type Closed: Send;
    async fn shutdown(self) -> Result<Self::Closed, Self::Error>;
}

shutdown must never block or panic: do all teardown that can return an error here, and return a Result. Closed is the shutdown witness: carry teardown diagnostics (flush results, drop counts) in it as plain data, or use ().

Construction is synchronous and I/O-free: new(addrs) only records the configuration. All network work happens in connect, which the runtime calls once at startup. The connected form holds the live client directly, so its operations never check a "maybe connected" state.

A broker may additionally keep a shared cell that connect fills, or shareable in-process state, as the in-memory broker does. Publishers can then be handed out while the application is still being assembled, before connect runs: the cell serves those early handles, not the connected form.

The conformance harness proves the whole sequence of transitions, and the NATS example walks it on a real client.

A broker you already shut down has nothing left to call, neither publish nor subscribe, so misuse by the owner does not compile. Sharing the connection is checked at run time: handles that share it (publishers handed out from the connected form, clones of a shareable broker) must return an error after shutdown and must never succeed silently against a dead connection. The lifecycle check covers that path too.

The in-memory broker walks the whole lifecycle in a few lines, and every example of it on this page is cut from that same file, so a contract that moves takes the page's code with it:

src/memory/mod.rs

impl<Log: LogMode> Broker for MemoryBroker<Log> {
    type Error = MemoryError;
    type Connected = ConnectedMemoryBroker<Log>;

    /// Connecting is free for an in-process bus. A shut-down bus (a clone lineage may have shut
    /// the shared state down) is revived with a fresh, empty registration map, so the connected
    /// form always starts live; a live bus keeps its registrations.
    fn connect(self) -> impl Future<Output = Result<Self::Connected, Self::Error>> {
        {
            let mut bus = self
                .state
                .subscribers
                .lock()
                .expect("memory broker mutex poisoned");
            if matches!(*bus, Bus::ShutDown) {
                *bus = Bus::Live(HashMap::new());
            }
        }
        ready(Ok(ConnectedMemoryBroker {
            state: self.state,
            mode: PhantomData,
        }))
    }
}

/// The connected form of [`MemoryBroker`]: the typed witness that [`Broker::connect`] ran.
///
/// Cheap to clone: the in-memory bus is shared state by nature, so the connected form is a
/// shareable handle on it, exactly like the unconnected broker. Subscriptions (the
/// [`Subscribe`] capability, [`MemorySource`]) resolve against this form, and carry over the
/// broker's log mode: only a [`Retaining`] one opens repositionable subscriptions.
pub struct ConnectedMemoryBroker<Log = Discarding> {
    state: Arc<MemoryState>,
    mode: PhantomData<Log>,
}

impl<Log> Clone for ConnectedMemoryBroker<Log> {
    fn clone(&self) -> Self {
        Self {
            state: Arc::clone(&self.state),
            mode: PhantomData,
        }
    }
}

impl<Log> fmt::Debug for ConnectedMemoryBroker<Log> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ConnectedMemoryBroker")
            .finish_non_exhaustive()
    }
}

impl<Log: LogMode> ConnectedMemoryBroker<Log> {
    /// Returns a publisher bound to this broker.
    #[must_use]
    pub fn publisher(&self) -> MemoryPublisher {
        MemoryPublisher {
            state: Arc::clone(&self.state),
            txn: Mutex::new(None),
        }
    }

    /// Returns a request / reply-capable publisher bound to this broker.
    ///
    /// See [`MemoryBroker::requester`] for why its operations report [`RequestError`] rather
    /// than [`MemoryError`].
    #[must_use]
    pub fn requester(&self) -> MemoryRequester {
        MemoryRequester::new(Arc::clone(&self.state))
    }
}

impl<Log: LogMode> ConnectedBroker for ConnectedMemoryBroker<Log> {
    type Error = MemoryError;
    type Closed = ClosedMemoryBroker;

    /// Enters the terminal shut-down state: the bus itself flips to its `ShutDown` variant, so
    /// every aliased handle that would touch it (a publisher's publish or transaction commit, a
    /// request) errors with [`MemoryError::ShutDown`]. Consuming `self` makes any further use
    /// of this handle a compile error; the returned witness reports how many subscriber
    /// registrations the teardown dropped.
    fn shutdown(self) -> impl Future<Output = Result<Self::Closed, Self::Error>> {
        let dropped = {
            let mut bus = self
                .state
                .subscribers
                .lock()
                .expect("memory broker mutex poisoned");
            match std::mem::replace(&mut *bus, Bus::ShutDown) {
                Bus::Live(subscribers) => subscribers.values().map(Vec::len).sum(),
                Bus::ShutDown => 0,
            }
        };
        ready(Ok(ClosedMemoryBroker {
            subscribers_dropped: dropped,
        }))
    }
}

ClosedMemoryBroker is the witness with the teardown diagnostics described above: it reports how many subscriber registrations the shutdown dropped.

`Subscribe`¶

Implement Subscribe on the connected form so a service can subscribe by the name of a topic, a subject or a queue. #[subscriber("name")] subscribes through it.

pub trait Subscribe: ConnectedBroker {
    type Subscriber: Subscriber;

    // Who publishes the retry copies of a by-name subscription, and who names where they
    // go. AddressedCopies where a publish under a subscribe name reaches the subscription
    // opened by it - a subject, a topic, a stream, a queue name - and the name is then the
    // address. NamedCopies where it is not: an MQTT filter reads many topics and names none.
    type Copies: CopyPath;

    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error>;

    // Defaulted: what a registration mounted by this name declared about its retries.
    // Map the cap and the destination onto the subscription the name opens, where the
    // broker has a mechanism for them.
    fn declare_retry(&self, name: &str, declaration: &RetryDeclaration)
        -> Result<(), DeclareRetryError>;
}

Opening a subscription and saying where a publish reaches it is all it has to do:

src/memory/mod.rs

impl<Log: LogMode> Subscribe for ConnectedMemoryBroker<Log> {
    type Subscriber = MemorySubscriber<Log>;
    // One subject is both ends of the bus here, so a publish under the name a subscription reads
    // reaches that subscription, and the name is the address.
    type Copies = AddressedCopies;

    fn subscribe(&self, name: &str) -> impl Future<Output = Result<Self::Subscriber, Self::Error>> {
        let (tx, rx) = mpsc::unbounded_channel();
        let name = name.to_owned();
        if let Err(err) = self.state.register(&name, tx.clone()) {
            return ready(Err(err));
        }
        ready(Ok(MemorySubscriber {
            name,
            rx,
            requeue: tx,
            state: Arc::clone(&self.state),
            seek: Arc::new(SeekControl::default()),
            mode: PhantomData,
        }))
    }
}

type Copies is what the by-name source reports on your broker, so it decides how #[subscriber("orders")] retries here: with AddressedCopies the name is the address and nothing else is written, with NamedCopies the mount site names where a copy goes.

Answer NamedCopies where a subscribe name is not a publish destination. An MQTT topic filter is one: devices/+/telemetry reads every device's topic and names none of them, so the mount site names where a copy goes, with .out_retry(policy).to(name) or a publish transform that names one per delivery. A subscription that needs more than a name to exist takes a descriptor of your own instead.

`Subscriber`¶

A subscriber is a Stream of incoming messages. Back-pressure comes from the stream itself.

pub trait Subscriber: Send {
    type Message: IncomingMessage;
    type Error: std::error::Error + Send + Sync + 'static;
    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_;
}

stream takes &mut self, so any state buffered between polls lives behind the mutable borrow, which keeps it cancel-safe.

`IncomingMessage`¶

A delivered message exposes its payload and its headers, and is acknowledged with ack or rejected with nack. ack consumes self, so a double ack is a compile error.

Every accessor here hands back a borrow, so one delivery needs one reference count and not one per field. A broker's delivery wraps the client's own message, whose name, payload and headers are usually three counted things; put them in one block behind one Arc and keep outside it what differs per copy, such as a log position or an attempt count. A hand-over - the copy per subscriber, a requeue, a replay - is then one atomic increment rather than three, which a service whose client thread and dispatch task sit on different cores pays for on every message. MemoryBroker is written that way.

pub trait IncomingMessage: Send + Sync {
    fn payload(&self) -> &[u8];
    fn headers(&self) -> &HeaderMap;
    async fn ack(self) -> Result<(), AckError>;
    async fn nack(self, requeue: bool) -> Result<(), AckError>;

    // Defaulted: false. The runtime reads this first and never calls
    // nack_after without it, so override the pair together.
    fn supports_nack_after(&self) -> bool;

    // Defaulted: AckError::Unsupported. Override when the transport has native
    // delayed redelivery (JetStream NAK with delay); handlers reach it through
    // HandlerOutcome::retry_after.
    async fn nack_after(self, delay: Duration) -> Result<(), AckError>;

    // Defaulted: None. Override (with the Partitioned capability) to feed the
    // runtime's keyed worker lanes, workers(n, by_key).
    fn partition_key(&self) -> Option<&[u8]>;

    // Defaulted: None. Override where the transport counts its own deliveries
    // (JetStream num_delivered, SQS ApproximateReceiveCount, Pub/Sub
    // delivery_attempt): a registration's max_attempts(..) cap then counts the
    // broker's redeliveries and not only the copies the runtime published.
    // The first delivery of a message answers 1.
    fn redelivery_count(&self) -> Option<u64>;
}

Delayed redelivery is two methods, and the runtime asks supports_nack_after. Override nack_after alone and the flag stays false, so the override is never called. By default nack_after returns AckError::Unsupported instead of settling the delivery with a plain nack(true): a transport that cannot hold a message back has to say so, or the pause before a retry turns into a storm of redeliveries.

A broker that overrides none of the four defaulted methods still works with every runtime feature. Where there is no native delayed redelivery the runtime runs retry_after itself: it drops the delivery and, after the delay, publishes a copy through the registration's retry publisher, with an incremented retry-count header. That copy goes to the address your subscription reports. Keyed worker lanes hand out keyless messages round-robin.

Where redelivery_count stays None, that header is the only count there is, and a max_attempts(..) cap is read from it. Override the method and the broker's count becomes the one count the cap reads: the framework's header is never added to it.

Honour a delay by publishing a copy yourself - a wait queue behind a dead-letter exchange, a retry topic - and increment RETRY_COUNT_HEADER (exported from ruststream::runtime) on that copy, but only where the transport counts nothing. Where it counts, your copy is a new message and the broker counts it from one again, so a delivery that has been round the wait queue reaches the cap as a first attempt. That is the behaviour of your delay scheme: document it, and leave the header out.

There is no broker to point at for "overrides nothing": every broker in this workspace overrides these methods. So the core pins the behaviour with a test:

src/message.rs

struct Stub {
    payload: Vec<u8>,
    headers: HeaderMap,
}

impl IncomingMessage for Stub {
    fn payload(&self) -> &[u8] {
        &self.payload
    }

    fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    fn ack(self) -> impl Future<Output = Result<(), AckError>> {
        ready(Ok(()))
    }

    fn nack(self, _requeue: bool) -> impl Future<Output = Result<(), AckError>> {
        ready(Ok(()))
    }
}

let stub = Stub {
    payload: b"body".to_vec(),
    headers: HeaderMap::new(),
};
assert_eq!(stub.payload(), b"body");
// The default partition_key is None (no key).
assert!(stub.partition_key().is_none());
// The default reports no native delayed redelivery, so the runtime uses its fallback.
assert!(!stub.supports_nack_after());
// The default nack_after signals "not honored" rather than silently degrading.
assert!(matches!(
    stub.nack_after(Duration::from_secs(1)).await,
    Err(AckError::Unsupported)
));

The Unsupported answer is what lets the runtime tell a transport with no delayed redelivery from one that honoured the delay, and run its own fallback.

`Publisher`¶

pub trait Publisher: Send + Sync {
    /// How your transport consumes the payload: `Lend` when it reads the bytes, `Take` when
    /// your client keeps them.
    type Payload: PayloadForm;

    type Error: std::error::Error + Send + Sync + 'static;

    /// Your broker's per-message settings. Every field optional; `()` when you have none.
    type Options: Clone + Send + Sync + 'static;

    async fn publish(
        &self,
        msg: OutgoingFor<'_, Self::Payload>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error>;

    /// Defaulted: headers this handle contributes under every publish.
    fn base_headers(&self) -> Option<&HeaderMap> { None }
}

Declare Take when your client keeps the payload past the call - it takes a Vec<u8>, a Bytes, anything it owns - and Lend when your transport only reads the bytes, because it writes them into a frame, a batch or a socket buffer of its own. Nearly every wire protocol is the second kind.

The message you receive follows the declaration, and so does what the framework does above you:

// Take: the buffer the framework wrote, yours to keep.
async fn publish(&self, msg: OutgoingMessage<'_, BytesMut>, options: Option<&()>) -> Result<(), Error>

// Lend: the bytes where they already are, valid for the length of the call.
async fn publish(&self, msg: OutgoingMessage<'_, &[u8]>, options: Option<&()>) -> Result<(), Error>

A Take publisher is handed the codec's own buffer, as it was written. Vec::from(payload) is free, because that buffer is the vector; payload.freeze() costs the one block that makes ownership shareable; and only a publish lending bytes the framework does not own is copied into the buffer on the way in.

A Lend publisher is handed &[u8] and nothing to release. Inside a dispatch loop the framework lends it one buffer per loop, cleared and written again for every message, so a reply through your broker allocates nothing at all per delivery. That is the whole reason the declaration exists: the runtime cannot lend what you might keep.

OutgoingMessage borrows its name and carries the payload and a header map your transport may take rather than copy. Read them with msg.payload() and msg.headers(), which answer &[u8] and &HeaderMap whichever form you declared. A transport that consumes the message takes its parts with msg.into_parts() - the destination, the payload and the map in one move, nothing copied.

A service writes the builder, not this method: publisher.message(&value).publish() picks the destination, the codec and the headers, and makes exactly one call to publish. Implement publish and the whole builder works on top of it.

Options holds what belongs to the message rather than to the handle: a QoS, a priority, an ordering key, an expiration. A call carries only the fields it adjusted, and the rest is what the policy fixed when it paired this publisher. Resolving the two is the first thing your publish does.

options is None on every path with no call site to adjust them - a reply, a deferred redelivery - and the policy's settings apply.

Clone and 'static are what the test harness asks of the type: it copies the options of a publish through an Out slot and hands them back to the test as this type. A service testing your broker then asserts on the value your publish received, not on the protocol field it became. Derive Debug and PartialEq as well, and the assertion reads with_options(&YourOptions { .. }) (asserting on Out slots).

base_headers is for a constant of the publisher itself: a tenant, a producer name, a schema id every message of this handle carries. The builder starts the outgoing headers from that base and writes the call site's headers over it key by key, so on a shared key the call site's value stays (see where the headers come from).

Transaction names an Options and a Payload of its own, and carries the same defaulted base_headers. A transaction is a publish surface of its own, so it may honour settings the publisher it was opened from does not, and a client buffer keeps the payload where the direct publish only reads it. Most brokers name the publisher's own types in both places. A handle with no constant of its own leaves base_headers defaulted in both places.

`PublishPolicy`¶

A broker publisher is a policy (an exchange, a queue timeout, a transactional id) and the live connection. Ship a separate policy type: it is constructible anywhere and holds the builder options.

Implement PublishPolicy on it: the policy constructs the live publisher on the connected form, and pair is that constructor. It is async and can return an error, so a broker that has to initialize a transactional producer does it here.

pub trait PublishPolicy<C: ConnectedBroker> {
    type Live; // the live publisher (or live wiring form, for combinator stacks)
    async fn pair(self, connected: &C) -> Result<Self::Live, PairError>;
}

The error is the type-erased PairError: wrap your broker's error with PairError::new. The policy instantiates the publisher once, at startup, so pair never reaches the hot path.

Ship one policy and live form per genuine publishing mode, and make the choice of mode a transition of the policy type rather than a runtime flag. The plain policy constructs the plain publisher, and a transactional_id(..) builder step moves it to a distinct transactional policy type whose live form implements TransactionalPublisher. The plain publisher then has no transactional surface at all.

The minimal reference is the in-memory broker's MemoryPublish and MemoryRequest: they have no options, so they are empty structs.

The core's typed combinators implement PublishPolicy functorially, so users compose codecs and transforms over your policy before it constructs the publisher.

When the plain policy is usable with its defaults (most are), also implement DefaultPublish on the connected form and name the policy there. The runtime then instantiates the reply publisher itself when a publish("dest") handler is mounted without an explicit .out_reply(..), and b.include(def) compiles on its own. Brokers whose publishers always need explicit options do not implement it, and their users specify the policy at every handler registration.

pub trait DefaultPublish: ConnectedBroker {
    type Policy: PublishPolicy<Self> + Default + Send + 'static;
}

Both halves, on a broker whose policy carries no options at all:

src/memory/mod.rs

impl<Log: LogMode> PublishPolicy<ConnectedMemoryBroker<Log>> for MemoryPublish {
    type Live = MemoryPublisher;

    fn pair(
        self,
        connected: &ConnectedMemoryBroker<Log>,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(Ok(connected.publisher()))
    }
}

impl<Log: LogMode> DefaultPublish for ConnectedMemoryBroker<Log> {
    type Policy = MemoryPublish;
}

Subscription sources¶

Subscribe covers the case where a name is all a subscription needs. When it needs options of your own broker (a consumer group, a durable name, a delivery policy), ship a descriptor type that implements SubscriptionSource:

pub trait SubscriptionSource<C: ConnectedBroker> {
    type Subscriber: Subscriber;

    // Who publishes the copies this subscription's retries are made of, and who names
    // where they go: AddressedCopies, NamedCopies or BrokerMoves.
    type Copies: CopyPath;

    fn name(&self) -> &str;
    fn subscribe(self, connected: &C) -> impl Future<Output = Result<Self::Subscriber, C::Error>> + Send;

    // Defaulted: the descriptor unchanged. Read the registration's max_attempts(..)
    // and dead_letter(..) here and apply them to the subscription you are about to
    // open, where the broker has a mechanism for them.
    fn declare_retry(self, declaration: &RetryDeclaration) -> Self;

}

// The other half of AddressedCopies: the destination is a property of the type, not an
// answer checked at startup. Ask the broker where only the live connection knows.
pub trait RedeliveryAddressed<C>: SubscriptionSource<C, Copies = AddressedCopies> {
    async fn redelivery_address(&self, connected: &C) -> Result<RedeliveryAddress, C::Error>;
}

Give the descriptor an associated constructor (OrdersStream::new(..)) rather than a free function: a user then names it directly in the attribute, #[subscriber(OrdersStream::new("orders", "workers"))].

The macro reads the type out of the constructor call, and accepts a builder chain on it as well (#[subscriber(OrdersStream::new("orders").durable("workers"))]), as long as each method returns Self.

type Subscriber is declared on the source, so one broker can offer several kinds of subscription (pub/sub and streams) with different subscriber types, or serve them all from one descriptor that branches inside, as the NATS example does.

Derive Clone on the descriptor: the mount rebuilds the configuration per registration, so one definition can be mounted on two brokers at once.

Who publishes a retry copy¶

Every descriptor declares one of three things with type Copies.

AddressedCopies says this process publishes the copies a retry needs, and the descriptor knows where they go. It implements RedeliveryAddressed beside SubscriptionSource, so the address is a property of the type rather than an answer checked at startup. This is the answer for a subject, a topic, a stream and a queue - one subscription, one destination the service can publish back to.

NamedCopies says this process publishes them but the descriptor cannot address them: a wildcard subject, an MQTT filter, a Pulsar pattern, a list of topics. One such subscription reads many destinations, so the mount site names one, statically with .out_retry(policy).to(name) or per delivery with a publish transform.

BrokerMoves says the server or the client library moves the delivery itself: a quorum queue with x-delivery-limit and an x-dead-letter-exchange, a Pub/Sub subscription with a dead-letter policy, an SQS redrive policy, a Pulsar consumer with a DeadLetterPolicy. Nothing is published from the service, so .out_retry(..) is a compile error at every mount site of that descriptor, and the error names it.

The first two pair a retry publisher for every registration from the broker's DefaultPublish policy, so a descriptor declaring either over a broker with no DefaultPublish does not compile.

Subscribe declares the same thing for the by-name form: type Copies there is what #[subscriber("orders")] reports on your broker. Answer AddressedCopies where a publish under a subscribe name reaches the subscription opened by it, which a subject, a topic, a stream and a queue name usually are - the name is then the address, and nothing else has to be written.

Where the nativeness depends on a field's value rather than on the type - a RabbitMQ queue without .delay(..) has no native delayed redelivery of its own - keep the path open.

What the registration declares¶

declare_retry hands you the cap and the destination the mount site declared, once per registration, before subscribe runs. A descriptor that has a mechanism of its own turns them into topology there, and only when both are declared, because a native dead-letter policy needs the limit and the address together. A descriptor without one keeps the default and the runtime applies the declaration on the retry path.

A registration mounted by a bare name declares the same thing, and the descriptor it gets is the core's Name, which carries no topology to put it in. Subscribe::declare_retry is where your broker takes it instead, at the same point and for the name it is about to open. Map it there the way your descriptor does, and only when both halves are declared.

The default accepts a registration that declared nothing. It accepts any declaration where your type Copies says this process publishes the copies, because the runtime applies the cap and the destination itself there. On a BrokerMoves broker it refuses a non-empty one at startup, naming the subscription and where the declaration belongs: nothing else would apply it, and a message whose cap silently went missing outlives its own dead-letter policy. Implement the method where the broker has a mechanism a bare name reaches - a Pub/Sub dead-letter policy, an SQS redrive policy, a Pulsar DeadLetterPolicy - and leave it alone everywhere else.

Where a retry copy is published¶

Without native delayed redelivery, the runtime honours retry_after by publishing a copy of the message once the delay is over. An AddressedCopies descriptor says where that copy goes.

src/memory/mod.rs

impl<Log: LogMode> SubscriptionSource<ConnectedMemoryBroker<Log>> for MemorySource {
    type Subscriber = MemorySubscriber<Log>;
    // The bus moves nothing on its own, and one subject is both ends of it, so a copy goes back
    // to the subject the subscription reads.
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        &self.name
    }

    async fn subscribe(
        self,
        connected: &ConnectedMemoryBroker<Log>,
    ) -> Result<Self::Subscriber, MemoryError> {
        Subscribe::subscribe(connected, &self.name).await
    }
}

impl<Log: LogMode> RedeliveryAddressed<ConnectedMemoryBroker<Log>> for MemorySource {
    fn redelivery_address(
        &self,
        _connected: &ConnectedMemoryBroker<Log>,
    ) -> impl Future<Output = Result<RedeliveryAddress, MemoryError>> + Send {
        // One subject is both ends of the bus, and no lookup is needed to say so.
        ready(Ok(RedeliveryAddress::new(self.name.clone())))
    }
}

Answer with the name a publisher bound to your broker uses to reach this subscription again: the subject on NATS, the topic on Kafka, the stream key on Redis.

On NATS JetStream a consumer is bound to a stream rather than to a subject, so the answer is a subject that stream is published on, and a descriptor built from the stream name asks the server for it. The runtime asks once, at startup, and a .to(name) at the mount site overrides it.

A .to(name) names a channel the registration sends to, so the generated document reports it with a send operation. The address you answer with is not reported: it is the subscription's own channel, which the document already carries.

harness::redelivery_address checks the answer you give: a publish to the reported address must arrive at the subscription that reported it. A NamedCopies descriptor has no answer to check, and harness::lifecycle covers the rest of the ladder for both.

Naming a kind by one string¶

A kind identified by a name and nothing else also implements FromName: its single constructor builds the value from that name.

src/memory/mod.rs

impl FromName for MemorySource {
    fn from_name(name: impl Into<Cow<'static, str>>) -> Self {
        Self::new(name.into().into_owned())
    }
}

#[subscriber(OrdersStream)] is then legal: the attribute names the kind, and the mount site supplies the value. A kind that needs more than one name (a topic and a subscription name) does not implement FromName, and that form does not compile for it.

Settings in your own vocabulary¶

The core does not know that a subscription has a stream, a durable name or a consumer group, so it gives you one hook: map_source, a transform over the source the mount site is building. You put your own trait on top and bind it to your source type:

use ruststream::runtime::{Declared, SubscriberBuilder, SubscriberSettings};

pub trait NatsSubscriber {
    fn jetstream(self, stream: impl Into<String>) -> Self;
    fn durable(self, name: impl Into<String>) -> Self;
}

// The four state slots are (workers, failure policies, start position, batch size); `Codec` is
// the registration's own decode override, `()` until one is named. Both travel unchanged.
impl<Def, Workers, Failures, StartPosition, Batch, Codec> NatsSubscriber
    for SubscriberBuilder<Def, SubscribeOptions, (Workers, Failures, StartPosition, Batch), Codec>
where
    Def: Declared,
{
    fn jetstream(self, stream: impl Into<String>) -> Self {
        self.map_source(|source| source.jetstream(stream))
    }

    fn durable(self, name: impl Into<String>) -> Self {
        self.map_source(|source| source.durable(name))
    }
}

The bound on the source type means these methods do not exist on a builder for another broker. The Out slot vocabulary below uses the same extension shape.

One core setting changes the source type rather than a state slot: start_at(..) wraps the descriptor in StartAt<SubscribeOptions, Position>. On the subscriptions that named a start position your methods fall out of scope, so cover that case with a second impl over the wrapped source. StartAt::map_inner takes the descriptor out of the wrapper and hands the position back untouched, so each method stays one line:

use ruststream::StartAt;
use ruststream::runtime::Fixed;

// The start-position slot is `Fixed` here by construction - `start_at(..)` is what produced the
// wrapper - and the source type is a different one, so this impl and the one above never overlap.
impl<Def, Workers, Failures, Batch, Codec, Position> NatsSubscriber
    for SubscriberBuilder<
        Def,
        StartAt<SubscribeOptions, Position>,
        (Workers, Failures, Fixed, Batch),
        Codec,
    >
where
    Def: Declared,
{
    fn jetstream(self, stream: impl Into<String>) -> Self {
        self.map_source(|source| source.map_inner(|inner| inner.jetstream(stream)))
    }

    fn durable(self, name: impl Into<String>) -> Self {
        self.map_source(|source| source.map_inner(|inner| inner.durable(name)))
    }
}

Publisher settings in your own vocabulary¶

The publish side is built the same way. The mount site names a policy with .out(marker, policy): the Reply marker for what a publish("dest") handler returns, a slot's marker for an Out slot. MapPublisher is the hook over the policy in that position:

use ruststream::runtime::MapPublisher;

pub trait NatsPublish {
    fn stream(self, name: impl Into<String>) -> Self;
    fn expect_last_sequence(self, seq: u64) -> Self;
}

impl<T: MapPublisher<Policy = Publish>> NatsPublish for T {
    fn stream(self, name: impl Into<String>) -> Self {
        self.map_publisher(|policy| policy.stream(name))
    }

    fn expect_last_sequence(self, seq: u64) -> Self {
        self.map_publisher(|policy| policy.expect_last_sequence(seq))
    }
}

In a service it reads like this:

b.include(confirm).out_reply(Publish).stream("ORDERS");
b.include(mirror).out(Audit, Publish).stream("AUDIT").build();

The bound is on the policy, not on the chain, so one impl covers the reply position, every slot, a router and a broker scope alike.

map_publisher replaces the policy with one of the same type, and a different policy type means a different publish mode, which belongs in the .out(marker, policy) call itself. An already-configured value can be passed there directly: .out_reply(Publish::default().stream("ORDERS")).

Per-message settings on the publish builder¶

A call site adjusts a field of your Publisher::Options through a step you add to the publish builder. Nothing wraps the publisher, so the publish still goes through the mount site's own entry, with the codec and the transforms that entry named.

The four pieces are an options type whose every field is optional, a policy that carries the defaults, a live publisher resolving one against the other, and an extension trait over PublishBuilder bounded on the options type. The bound is what keeps your steps off a builder over another broker's publisher:

tests/publish_options.rs

/// The broker's per-message settings. Every field optional: what a call leaves unset keeps what
/// the policy fixed. `Debug` and `PartialEq` are not part of the contract, they are what a test
/// naming this type needs to assert on it.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct PriorityOptions {
    priority: Option<u8>,
    ttl: Option<u8>,
}

/// The publish policy: pure declaration, constructible anywhere, and the place the defaults are
/// configured.
#[derive(Debug, Clone, Copy, Default)]
struct PriorityPublish {
    priority: u8,
    ttl: u8,
}

impl PriorityPublish {
    fn priority(mut self, priority: u8) -> Self {
        self.priority = priority;
        self
    }

    fn ttl(mut self, ttl: u8) -> Self {
        self.ttl = ttl;
        self
    }
}

/// The live publisher: the connection, plus the defaults the policy carried.
struct PriorityPublisher {
    inner: MemoryPublisher,
    default_priority: u8,
    default_ttl: u8,
}

impl PublishPolicy<ConnectedMemoryBroker> for PriorityPublish {
    type Live = PriorityPublisher;

    async fn pair(self, connected: &ConnectedMemoryBroker) -> Result<Self::Live, PairError> {
        Ok(PriorityPublisher {
            inner: Publish.pair(connected).await?,
            default_priority: self.priority,
            default_ttl: self.ttl,
        })
    }
}

impl Publisher for PriorityPublisher {
    // Forwarded to the bus underneath, which keeps what it is handed.
    type Payload = Take;
    type Error = MemoryError;
    type Options = PriorityOptions;

    async fn publish(
        &self,
        msg: OutgoingMessage<'_, BytesMut>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        let priority = options
            .and_then(|options| options.priority)
            .unwrap_or(self.default_priority);
        let ttl = options
            .and_then(|options| options.ttl)
            .unwrap_or(self.default_ttl);
        // A real broker hands the resolved values to its client as the protocol fields they are.
        // The in-memory bus has no such field, so this one puts them where a test can read them
        // back.
        let mut headers = msg.headers().clone();
        headers.insert("priority", priority.to_string());
        headers.insert("ttl", ttl.to_string());
        let stamped = OutgoingMessage::new(msg.name(), msg.payload()).with_headers(headers);
        self.inner.publish(stamped, None).await
    }
}

impl TransactionalPublisher for PriorityPublisher {
    async fn begin_transaction(&self) -> Result<(), Self::Error> {
        self.inner.begin_transaction().await
    }

    async fn commit(&self) -> Result<(), Self::Error> {
        self.inner.commit().await
    }

    async fn abort(&self) -> Result<(), Self::Error> {
        self.inner.abort().await
    }
}

/// The steps the broker puts in its prelude. The bound on the sink's options type is what keeps
/// them off a builder over any other broker's publisher.
trait PriorityPublishSteps {
    /// Sends this one message at `priority`, whatever the mount site's default is.
    #[must_use]
    fn priority(self, priority: u8) -> Self;

    /// Sends this one message with `ttl`, whatever the mount site's default is.
    #[must_use]
    fn ttl(self, ttl: u8) -> Self;
}

impl<Sink, Body, Enc, Hdrs, Dest> PriorityPublishSteps
    for PublishBuilder<Sink, Body, Enc, Hdrs, Dest>
where
    Sink: PublishSink<Options = PriorityOptions>,
{
    fn priority(mut self, priority: u8) -> Self {
        self.options_mut()
            .get_or_insert_with(PriorityOptions::default)
            .priority = Some(priority);
        self
    }

    fn ttl(mut self, ttl: u8) -> Self {
        self.options_mut()
            .get_or_insert_with(PriorityOptions::default)
            .ttl = Some(ttl);
        self
    }
}

The broker half is the same on the macro path and the manual one. Ship the extension trait from your prelude next to the policy aliases.

A step is the only shape a per-message setting takes. Do not put the send in your own trait: a publish that goes through a value of yours is one the slot view no longer sees, and a setting like an ordering key is exactly what a test wants to assert on. Do not carry one as a header either: the setting is a protocol field, and the header map would carry it as bytes your publish has to parse back inside one process.

A value your broker cannot honour is a publish error, never a silent fallback to the default: the caller asked for an ordering it would not get.

Capability traits¶

Implement only the capabilities your broker supports; none are part of the mandatory interface. BatchSubscriber comes closest to one: offer it wherever you can, because every batch handler asks for one and a transport with no batching of its own can still assemble batches on the client.

Trait	For brokers that support
`BatchSubscriber`	receiving messages in batches
`TransactionalPublisher`	begin / commit / abort around publishes on the publisher handle
`OwnedTransactions` / `Transaction`	any number of transactions open at once per handle, each with its own buffer
`RequestReply`	native request-reply
`Partitioned`	a partition key on outgoing messages
`Seekable` / `Seeker`	repositioning a live subscription in a replayable log
`Positioned`	reporting a delivery's own position in the log
`DescribeServer`	reporting a `ServerSpec` for AsyncAPI

Seekable hands out its Seeker handle before stream borrows the subscriber, so a running subscription can be repositioned from outside the dispatch loop.

Positions are broker-owned: you declare the constructors, KafkaPosition-style, on your own type. A position captured from a delivered message through Positioned::position pins the contract: seeking to it redelivers exactly that message. Constructed positions keep the semantics your position type documents.

Document what one seek covers (a consumer instance or a shared group cursor) and reset any ack bookkeeping the reposition invalidates.

To let handler bodies seek, carry the delivery's position and the subscription's seeker as fields of your per-delivery context and publish ContextField keys for them. The in-memory broker's MemoryContext, with its Position and SeekHandle keys, is the model. The batch forms take the seeker from the batch context below, which carries no position.

A DescribeServer description reports the host and port clients connect to. Credentials never appear in it: the document is generated to be published. A broker configured from a URL builds its description with ServerSpec::from_url, which drops the user name and password. Trimming the scheme off the URL and passing the rest on keeps them: that is the bug from_url replaced, and it shipped in more than one broker crate. A broker that configures several addresses joins them from ServerSpec::host_from_url.

These traits are the vocabulary a handler body writes. A body bounds its slot with the capability it needs (Out<impl TransactionalPublisher, Journal>, or where W: TransactionalPublisher on the manual path) and never with a type of yours, and the mount site checks the bound policy's live form against it once, at compile time.

Under each of the four publisher capabilities the arena entry also offers that capability's typed form over the mount site's codec and the marker's dictionary: the publish builder, a transaction scope, an owned transaction, a correlated request. Implementing the trait on your live publisher is all a service needs to reach them.

Batches: `BatchSubscriber`¶

A handler taking &[T] consumes a batch, and its mount site names one number, the batch size. The runtime passes it straight to BatchSubscriber::batches(size). The batch your subscriber yields is the batch the body sees: the runtime never splits or merges one, so a batch never carries more than size messages, and it carries fewer whenever that is all the transport had.

Translate size into whatever your client already speaks: XREADGROUP COUNT, a JetStream pull batch, a Kafka poll limit. Everything else about how a batch forms (a block timeout, a consumer group, a prefetch window) stays your own vocabulary, configured on your subscription source through your settings extension trait. A service then writes b.include(handler.batch(nonzero!(6)).block(Duration::from_secs(5))), the core's word first and yours after it.

Put the capability on every subscriber a mount can reach, not only on the one your own descriptor opens. #[subscriber("topic")] goes through Subscribe, so a &[T] body on that form asks for BatchSubscriber on Subscribe::Subscriber.

A crate that wired the capability onto its descriptor's subscriber alone leaves the string-literal form failing to compile. Where the two are the same type there is nothing to do, and where they differ both need it.

Where the transport delivers one message at a time, implement the capability anyway and assemble the batches on the client with the core's BufferedSubscriber, whose batches honours the size it is given. The size is not yours to choose; the deadline that closes a partial batch is, and it need not be a constant.

Expose that deadline on your subscription descriptor (.max_wait(Duration::from_millis(25))) and hand it to the wrapper as the subscription opens, so a service can tune it per subscription. The 10 ms default is sized for an in-process bus: once a network round trip is in the way it closes most batches at a single delivery, so the broker crates that ship the deadline as a descriptor option settle between 10 and 50 ms.

Everything else about the subscriber passes through the wrapper unchanged:

tests/batch_subscriber.rs

/// What a broker crate writes when its transport has no batches of its own: the subscriber it
/// already has, wrapped in the core's client-side buffer, and `BatchSubscriber` delegated to it.
/// The deadline that closes a partial batch is the broker's own choice; the batch size is not -
/// it arrives per subscription, as the argument of `batches`.
struct TrickleSubscriber(BufferedSubscriber<MemorySubscriber<Retaining>>);

impl TrickleSubscriber {
    fn new(inner: MemorySubscriber<Retaining>) -> Self {
        Self(BufferedSubscriber::new(inner).max_wait(Duration::from_millis(5)))
    }
}

impl Subscriber for TrickleSubscriber {
    type Message = <MemorySubscriber<Retaining> as Subscriber>::Message;
    type Error = <MemorySubscriber<Retaining> as Subscriber>::Error;

    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
        self.0.stream()
    }
}

impl BatchSubscriber for TrickleSubscriber {
    type Batch = Vec<<MemorySubscriber<Retaining> as Subscriber>::Message>;

    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, Self::Error>> + Send + '_ {
        self.0.batches(size)
    }
}

/// Buffering does not move the subscription, so every other capability reaches through the
/// wrapper unchanged - here the seeker, which is what lets a batch subscription open at a
/// position even where the batches are assembled on the client.
impl Seekable for TrickleSubscriber {
    type Seeker = <MemorySubscriber<Retaining> as Seekable>::Seeker;

    fn seeker(&self) -> Self::Seeker {
        self.0.seeker()
    }
}

/// The broker's own subscription descriptor, opening the batching subscriber above.
#[derive(Clone)]
struct Trickle {
    name: &'static str,
}

impl SubscriptionSource<ConnectedMemoryBroker<Retaining>> for Trickle {
    type Subscriber = TrickleSubscriber;
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        self.name
    }

    async fn subscribe(
        self,
        connected: &ConnectedMemoryBroker<Retaining>,
    ) -> Result<TrickleSubscriber, MemoryError> {
        Ok(TrickleSubscriber::new(
            Subscribe::subscribe(connected, self.name).await?,
        ))
    }
}

// A descriptor that addresses its own copies says where they go, and one in-memory subject is
// both ends of the bus, so the name answers for itself with no lookup in between.
impl RedeliveryAddressed<ConnectedMemoryBroker<Retaining>> for Trickle {
    fn redelivery_address(
        &self,
        _connected: &ConnectedMemoryBroker<Retaining>,
    ) -> impl Future<Output = Result<RedeliveryAddress, MemoryError>> + Send {
        ready(Ok(RedeliveryAddress::new(self.name)))
    }
}

Nothing in the mount site says which of the two you did: a service names the batch size and gets batches.

Declining the capability is still a legitimate answer where batching would break a guarantee the transport carries. A ZeroMQ ROUTER is the case in practice: it answers each peer at that peer's own reply-to, while a whole batch reaches its reply wiring with one PublishContext, so the replies for the batch would all go to one peer's address. Say so in your crate's docs: a &[T] body then does not compile on that transport.

The conformance batch suite checks the contract: it opens a subscription at a size smaller than the run and fails a broker whose batches come back larger. It is not part of harness::run_suite; capability suites are yours to call, one per capability you implement.

The prelude your crate ships¶

Your types are named at the mount site, not in the body, and that is what your crate's prelude is for. Ship a prelude module in three layers, in this order:

pub use ruststream::prelude::*; so one glob serves the whole file;
your own surface a service names: the broker, its subscription source, its Config, its error, the ContextField keys a body reads;
your publish policies under the uniform names every broker uses - Publish, and where you have them TransactionalPublish and Request (pub use crate::KafkaTransactionalPublish as TransactionalPublish;). Add the capability traits you implement on your live values as a manifest, so the glob that names the policies also puts their operations in scope.

The core prelude exports nothing under those three names, so a mount site reads the same whichever broker it is on, and swapping brokers swaps the glob. Never alias a policy to a core trait name (Publisher, TransactionalPublisher, OwnedTransactions, RequestReply) or re-export something else under one: a body that globs both preludes has to keep resolving those to the core traits.

The manifest is what your glob adds: the consumer-side traits a body reaches through your broker, Positioned, Seeker, Transaction and the like. The four publisher capabilities are already in the core prelude, so re-exporting them changes nothing.

Leave out a trait whose method would collide with a defaulted core method, in practice Partitioned::partition_key against IncomingMessage::partition_key, and let a service that needs it import it explicitly. BatchSubscriber belongs in no manifest at all: the framework calls it, and no body ever writes it as a bound.

ruststream::memory::prelude is the worked example.

Extending the `Out` slot vocabulary¶

An Out<impl X, Marker> handler parameter accepts any X the live value behind the slot implements; on top of that the core delegates its own capability set (Publisher, TransactionalPublisher, OwnedTransactions, RequestReply). When your live value offers more than that, or is not a publisher at all (a per-partition producer cache, a shard router), declare your own capability trait and implement it for the live value.

What the body holds is not that value but the arena entry, Slot<Marker, W, E, Pipe, Body>, a transparent window onto it. Autoderef carries a method call through the window, but not a trait bound: a helper written as fn issue<L: Lanes>(lanes: &L) rejects the entry with E0277.

Add one blanket impl next to your trait, impl<M, W: Lanes, E, Pipe, Body> Lanes for Slot<M, W, E, Pipe, Body> delegating through the entry's Deref, and helpers and bodies generic over the capability take the entry as it is. The concrete type still never appears in application code:

MacrosManual

tests/out_slots.rs

// A paired value that is NOT a publisher: a lane router in the shape of a broker's
// per-partition producer cache. The capability is broker-defined; the core knows nothing
// about it.
#[derive(Clone)]
struct LaneRouter {
    publisher: MemoryPublisher,
}

/// The broker-defined capability: pick a destination lane for a shard.
trait ShardLanes {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str);
}

impl ShardLanes for LaneRouter {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        let dest = if shard.is_multiple_of(2) {
            "slots.lane.even"
        } else {
            "slots.lane.odd"
        };
        (&self.publisher, dest)
    }
}

// Grafted onto the arena entry once, for every marker, delegating through the entry's
// transparent `Deref`: this is how a broker crate extends the slot vocabulary with its own
// traits. A handler body holds the entry, so without this impl the capability is reachable by
// autoderef for a method call but never satisfies a trait bound.
impl<M, W: ShardLanes, E, Pipe, Body> ShardLanes for Slot<M, W, E, Pipe, Body> {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        (**self).lane(shard)
    }
}

/// The bound the graft buys: a helper generic over the capability, not over the concrete live
/// type, takes the entry a handler body holds.
async fn sent<L: ShardLanes + Sync>(lanes: &L, event: &Event) -> bool {
    let (publisher, dest) = lanes.lane(event.id);
    publisher.message(event).to(dest).publish().await.is_ok()
}

/// The policy half: pure declaration pairing into the router, like a broker's
/// `per_partition()` policy pairs into its producer cache. No `Clone`: resolution consumes it.
struct LanePolicy;

impl PublishPolicy<ConnectedMemoryBroker> for LanePolicy {
    type Live = LaneRouter;

    async fn pair(self, connected: &ConnectedMemoryBroker) -> Result<Self::Live, PairError> {
        Ok(LaneRouter {
            publisher: Publish.pair(connected).await?,
        })
    }
}

/// The handler bounds its slot with the broker-defined capability, not a core one.
#[subscriber("slots.sharded")]
async fn route_shard(event: &Event, Out(lanes): Out<impl ShardLanes>) -> HandlerOutcome {
    if sent(lanes, event).await {
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry()
    }
}

tests/manual_out_slots.rs

// A paired value that is NOT a publisher: a lane router in the shape of a broker's
// per-partition producer cache. The capability is broker-defined; the core knows nothing
// about it.
#[derive(Clone)]
struct LaneRouter {
    publisher: MemoryPublisher,
}

/// The broker-defined capability: pick a destination lane for a shard.
trait ShardLanes {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str);
}

impl ShardLanes for LaneRouter {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        let dest = if shard.is_multiple_of(2) {
            "slots.lane.even"
        } else {
            "slots.lane.odd"
        };
        (&self.publisher, dest)
    }
}

// Grafted onto the arena entry once, for every marker, delegating through the entry's
// transparent `Deref`: this is how a broker crate extends the slot vocabulary with its own
// traits. A body holds the entry, so without this impl the capability is reachable by autoderef
// for a method call but never satisfies a trait bound.
impl<M, W: ShardLanes, E, Pipe, Body> ShardLanes for Slot<M, W, E, Pipe, Body> {
    fn lane(&self, shard: u64) -> (&MemoryPublisher, &'static str) {
        (**self).lane(shard)
    }
}

/// The bound the graft buys: a helper generic over the capability, not over the concrete live
/// type, takes the entry a body holds.
async fn sent<L: ShardLanes + Sync>(lanes: &L, event: &Event) -> bool {
    let (publisher, dest) = lanes.lane(event.id);
    publisher.message(event).to(dest).publish().await.is_ok()
}

/// The policy half: pure declaration pairing into the router, like a broker's
/// `per_partition()` policy pairs into its producer cache. No `Clone`: resolution consumes it.
struct LanePolicy;

impl PublishPolicy<ConnectedMemoryBroker> for LanePolicy {
    type Live = LaneRouter;

    async fn pair(self, connected: &ConnectedMemoryBroker) -> Result<Self::Live, PairError> {
        Ok(LaneRouter {
            publisher: Publish.pair(connected).await?,
        })
    }
}

/// The body leaves the wired live value generic and bounds it with the broker-defined
/// capability, exactly as the attribute's `Out<impl ShardLanes>` does.
struct RouteShard;

struct Lanes;

impl OutSlot for Lanes {
    const NAME: &'static str = "Lanes";
    type Destination = Reads;
}

impl<L> Handle<Event, (), Outs<(L,)>> for RouteShard
where
    L: OutEntry<Lanes, Wire: ShardLanes>,
{
    async fn handle(
        &self,
        event: &Event,
        outs: &Outs<(L,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if sent(outs.get(Lanes), event).await {
            Ok(())
        } else {
            Err(HandlerOutcome::retry())
        }
    }
}

Where the send happens is what shapes the trait, and there are two shapes.

A router-shaped capability hands out a publisher and never sends one itself: the per-partition producer cache above picks the publisher for a shard and returns it. A publish through that publisher passes outside the slot view, so the harness does not attribute it to the slot, no more than it does a settled owned transaction's buffer. Assert it on the broker's publish log instead. That is the attribution boundary and the price of handing out the inner publisher.

A step-shaped capability sets one per-message setting and ends in a single publish: an ordering key, a priority, a QoS. That one is not a capability trait at all: it is a field of your Publisher::Options, adjusted by a step on the publish builder. See per-message settings on the publish builder.

Your crate's prelude¶

Two files import different things, and the split is what keeps a service portable. A handler body imports ruststream::prelude::* and nothing of yours: it bounds an injected slot with the core capability trait it needs (Out<impl Publisher>, Out<impl TransactionalPublisher>, Out<impl OwnedTransactions>, Out<impl RequestReply>), so the body says what it needs of a publisher and never which broker provides it.

A routes file imports your prelude, because mounting is where a broker is named.

The one exception is a per-message setting, whose call site is in the body. A body that adjusts one imports your prelude for the step and names your options type in its bound (Out<impl Publisher<Options = MqttOptions>, Telemetry>). That body is tied to your broker, and its signature says so.

That makes your prelude the one import of yours a service writes, so its shape is part of the contract. The policy aliases (NatsPublish as Publish, KafkaTransactionalPublish as TransactionalPublish, LapinRequest as Request) make a routes file read the same whichever broker it mounts, so switching brokers is a change of import.

Your half of the naming rule: an explicit re-export shadows a glob without a word, so a name you spell like a core trait takes that trait away from every service writing the glob, and the error surfaces in the service's file rather than in yours.

Pin both halves with a probe behind your own glob: the bound a body writes still has to arrive as the core trait, and the mount-site name still has to be your policy.

// in your crate, behind your own prelude glob
use crate::prelude::*;

// A capability bound a body states: the core trait, not something of yours.
fn _p<T: Publisher>() {}

// A mount-site name: your policy, constructible with no connection in sight.
fn _q() {
    let _: Publish = Publish::default();
}

Per-delivery context and `Ctx` keys¶

A broker with native delivery metadata (a partition, an offset, a stream sequence) exposes it as a typed per-delivery context: a #[non_exhaustive] struct the subscriber names, plus ContextField key types. A key binds a single field as a handler parameter through the Ctx<K> extractor. Keys are unit structs, and the delivery path carries no type-map and no heap allocation.

/// Per-delivery context of this broker.
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct MyContext {
    pub partition: i32,
}

/// `Ctx<Partition>` in a handler binds the delivery's partition.
#[derive(Debug, Default, Clone, Copy)]
pub struct Partition;

impl ContextField for Partition {
    type Context = MyContext;
    type Value = i32;
    fn read(self, src: &MyContext) -> i32 {
        src.partition
    }
}

The sketch reads a Copy scalar, where owning and borrowing are the same thing. A position that is not Copy, a Pulsar message id or a Kinesis shard plus its sequence string, is read by borrowing: Field::Value<'a> is generic over the source's lifetime, so the key hands back &'a MessageId and a body reading it with ctx.context(..) copies nothing.

Only ContextField::Value, the value behind the Ctx<K> extractor, has to be owned and 'static, because extractor values bind before the body runs; that key clones what the borrowing one returns. A key usually implements both traits, one shape each.

A broker with no per-delivery fields uses ().

Batch subscriptions get a context of their own, because a batch spans many deliveries. Build a second struct out of what the whole subscription shares (a seek handle, a stream name, a consumer group), implement BuildBatchContext on it, and publish Field keys so a batch body reads it with ctx.context(..). The runtime builds one value per batch from the batch's first delivery.

Per-delivery fields stay out of it: a position belongs to one delivery, so a batch reads it off the elements. Keeping the two structs apart is what makes that a compile-time rule, since a per-delivery context does not implement BuildBatchContext and a batch body therefore cannot name it.

The in-memory broker's MemoryBatchContext is the model: the subscription's seeker sits under the same SeekHandle key its per-delivery context publishes. A broker with nothing subscription-scoped to offer implements nothing and leaves batches on the () default.

Middleware on the async edges¶

Integrations that need async I/O around encode and decode (a schema registry, a wire-format envelope) do not belong in a Codec: the core codec is synchronous and handlers should stay on the default one.

Put them on the async edges instead. Transcode incoming payloads on the subscription's delivery path, before the codec sees them, and frame outgoing ones with a core PublishLayer added app-wide via RustStream::publish_layer. The publish layer is async and can return an error, and Outgoing::payload_mut exists exactly for envelope wrapping.

Protocol bindings¶

The generated AsyncAPI document has room for what only your broker knows: a RabbitMQ queue's durability, a Kafka consumer group, an MQTT QoS. The specification calls those bindings, and your descriptor fills them.

tests/asyncapi.rs

/// A descriptor of the shape a broker crate ships: it reads its own private fields and says
/// what the protocol calls them.
#[derive(Clone)]
struct RabbitQueue {
    name: &'static str,
    durable: bool,
}

impl<C: Subscribe> SubscriptionSource<C> for RabbitQueue {
    type Subscriber = C::Subscriber;
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        self.name
    }

    async fn subscribe(self, connected: &C) -> Result<Self::Subscriber, C::Error> {
        connected.subscribe(self.name).await
    }

    fn channel_bindings(&self) -> Bindings {
        let body = AmqpChannel {
            is: "queue",
            queue: AmqpQueue {
                name: self.name,
                durable: self.durable,
            },
        };
        Binding::new("amqp", "0.3.0", &body)
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }

    fn operation_bindings(&self) -> Bindings {
        Binding::new("amqp", "0.3.0", &AmqpOperation { ack: true })
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }

    fn message_bindings(&self) -> Bindings {
        let body = AmqpMessage {
            message_type: "order",
        };
        Binding::new("amqp", "0.3.0", &body)
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }
}

Binding::new(protocol, version, &body) serializes the body once and writes bindingVersion itself, so you cannot ship a binding without one. The protocol key is checked against the specification's closed list, and an unlisted key comes back as an error rather than reaching a document no tool can read. Bindings is empty by default: a descriptor that says nothing changes no document.

The server level is a field rather than a method, because a server is described once per broker: ServerSpec::new(host, protocol).bindings(..) in your DescribeServer impl.

Three rules bound what belongs in a binding.

The value is computed from the descriptor alone. The document is built before anything connects, so a Kafka topic's real partition count, the topic behind a Pub/Sub subscription and an SQS queue's ARN cannot come from here.

A credential never goes in, for the reason DescribeServer gives. conformance::harness has the check: configure your broker and your descriptor with a known password and run the scan.

tests/conformance_self.rs

/// The in-memory broker has no network address and no binding, so nothing it describes can leak a
/// password. A broker configured from a URL runs this with the password it was configured with.
#[cfg(feature = "asyncapi")]
#[test]
fn memory_broker_describes_without_credentials() {
    harness::describes_without_credentials(
        &MemoryBroker::new(),
        &MemorySource::new("orders"),
        "hunter2",
    );
}

A protocol the specification has no binding for goes in Binding::extension("x-kinesis", &body). The protocol keys are a closed list, so ZeroMQ, Kinesis and a file transport have no lawful key; an x- extension sits at the same level and carries no bindingVersion.

Bindings come from a descriptor, so a subscription opened by bare name carries none: there is nothing bound to it to describe. A broker that wants bindings ships a SubscriptionSource type.

Your publish policies fill the same three names on the other side. A reply, an Out slot and the publisher a dead-lettered delivery leaves through are each a PublishPolicy, and each describes the channel it publishes to.

tests/asyncapi.rs

/// A publish policy of the shape a broker crate ships: an SNS topic is named by its binding's
/// required `name`, and the hook is handed the destination the mount site resolved.
#[derive(Clone, Copy, Default)]
struct TopicPublish;

impl PublishPolicy<ConnectedMemoryBroker> for TopicPublish {
    type Live = MemoryPublisher;

    fn pair(
        self,
        connected: &ConnectedMemoryBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        live(connected)
    }

    fn channel_bindings(&self, channel: &str) -> Bindings {
        let body = SnsChannel {
            name: channel.to_owned(),
        };
        one("sns", "0.1.0", &body)
    }

    fn operation_bindings(&self, channel: &str) -> Bindings {
        let body = SnsOperation {
            topic: SnsChannel {
                name: channel.to_owned(),
            },
        };
        one("sns", "0.1.0", &body)
    }

    fn message_bindings(&self, _channel: &str) -> Bindings {
        let body = SnsMessage {
            message_type: "progress",
        };
        one("sns", "0.1.0", &body)
    }
}

Each hook is handed the destination the mount site resolved. For a reply that is the reply type's own name or the registration's publish("dest") clause, for a slot entry its own name, for a dead-lettered delivery the dead_letter("dlq") declaration. An SNS topic and an SQS queue are named by the required name of their binding, and that name comes from here: a policy holds your broker's settings and never a destination. Where a transform names the destination per delivery the channel reports no address, and the hook is handed the mount site's fallback name instead.

A descriptor's hooks take no such parameter: a subscription source knows the subscription it describes.

The three rules hold unchanged here. One thing does not carry over: a reply has no send operation of its own, so operation_bindings on a reply's policy reaches no document. A slot and a dead-letter destination each have one.

A fourth method belongs to the reply alone. reply_address_location is where a client reads the address of an answer your broker routes through a reply-to header:

tests/asyncapi.rs

/// The reply's own policy: it answers where a client reads the address of an answer.
#[derive(Clone, Copy, Default)]
struct ReplyToPublish;

impl PublishPolicy<ConnectedMemoryBroker> for ReplyToPublish {
    type Live = MemoryPublisher;

    fn pair(
        self,
        connected: &ConnectedMemoryBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        live(connected)
    }

    fn channel_bindings(&self, channel: &str) -> Bindings {
        let body = NatsChannel {
            subject: channel.to_owned(),
        };
        one("nats", "0.1.0", &body)
    }

    fn reply_address_location(&self) -> Option<&'static str> {
        Some("$message.header#/reply-to")
    }
}

The document then reports the reply channel with address: null and puts the expression in the receive operation's reply.address.location. It is read only where the mount site composes a transform that names the destination per delivery; otherwise the reply goes to the declared name, and that is what the document says.

A protocol the specification does not list has no publish-side binding to fill either. The in-memory broker is the example: memory is not a key, so MemoryPublish stays silent rather than inventing one.

The hooks are gated on the core's asyncapi feature. Forward it from your crate:

[features]
asyncapi = ["ruststream/asyncapi"]

and put #[cfg(feature = "asyncapi")] on each method you fill in.

Config and defaults¶

Your crate owns its Config: the core carries no broker-specific config. If a field has no sane default, do not implement Default. The user then sets the value explicitly instead of inheriting a default that breaks later.

Errors¶

Use thiserror and one crate-level error enum, with variants by source. Mark public error enums #[non_exhaustive]. Never use anyhow in a library crate.

Test support¶

Ship an in-process transport implementing TestableBroker on its connected form under a testing feature. Register it with register_testable_broker! for that connected type: the harness connects every broker before recovering its transport. Users can then unit-test handlers against your broker with the TestApp harness.

The transport does core routing only: it dispatches published messages to matching subscribers, and it answers ack and nack the way the real transport answers. Where the real transport acknowledges, the stand-in answers in memory: nack(requeue = true) puts the delivery back. Where it cannot acknowledge at all (ZeroMQ, MQTT QoS 0, Redis pub/sub), the answer stays AckError::Unsupported. A stand-in that claims a settlement its transport never performs is what makes a handler's retry pass in a test and lose the message in production.

Do not simulate broker-specific semantics (durable cursors, redelivery timers, offsets, dead-letter routing) in it; those are verified end to end against a real server.

The reference is the in-memory broker's own implementation (on ConnectedMemoryBroker):

src/memory/mod.rs

// The harness drives the connected form: TestApp connects every registered broker before it
// recovers the in-process transport, and run_suite scenarios receive connected brokers.
#[cfg(feature = "testing")]
impl<Log: LogMode> crate::testing::TestableBroker for ConnectedMemoryBroker<Log> {
    fn install_coordinator(&self, coordinator: Coordinator) {
        self.state.install_coordinator(coordinator);
    }

    fn inject(&self, message: OutgoingMessage<'_>) {
        let (name, payload, headers) = message.into_parts();
        // Injecting into a shut-down bus is a harness bug (both run_suite and TestApp drive
        // the bus strictly before shutdown), so fail loudly instead of losing the message.
        self.state
            .fanout(name, Bytes::copy_from_slice(payload), headers)
            .expect("inject on a shut-down broker: drive the harness before shutdown");
    }

    /// What the broker holds under `name`: everything published there on a discarding broker
    /// (which records for the length of a harness run), and the retained window on a retaining
    /// one, so an assertion never claims more than the broker keeps.
    fn published(&self, name: &str) -> Vec<RawMessage> {
        self.state
            .log
            .lock()
            .expect("memory broker mutex poisoned")
            .name(name)
            .map(|log| log.messages(name))
            .unwrap_or_default()
    }
}

// One registration per log mode: the harness recovers a broker by its concrete type, and the
// two modes are two types.
#[cfg(feature = "testing")]
crate::register_testable_broker!(ConnectedMemoryBroker<Discarding>);
#[cfg(feature = "testing")]
crate::register_testable_broker!(ConnectedMemoryBroker<Retaining>);

The transport calls Coordinator::enqueued on every enqueue into a subscriber and Coordinator::consumed when a delivery is settled or dropped, so the harness can tell when the reaction has settled. It routes delayed redeliveries through Coordinator::schedule_redelivery.

That one type works with both TestApp and the conformance suite. See Testing for the user-facing side, and Conformance to prove the implementation with run_suite and the lifecycle ladder check.

Writing one you can trust¶

A stand-in is the type a service's whole test suite runs against, so every difference between it and the real transport is a green test for behaviour production does not have. The differences that matter are not exotic ones, and each rule below costs about one test.

Run the core's contract suites against the stand-in, not only against a server. The suites are written against the traits and do not care whether a real broker or the stand-in answers them. One #[tokio::test] is enough:

tests/conformance_self.rs

/// The suites that read a broker's publish log back - the routing contract's log check and the
/// seeking capability - need a broker that keeps one.
fn replaying() -> MemoryBroker<Retaining> {
    MemoryBroker::retaining(Retention::Messages(nonzero!(64)))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_broker_passes_conformance_suite() {
    harness::run_suite(replaying).await;
}

Run lifecycle first. It walks new -> connect -> subscribe -> publish -> ack -> shutdown and then asks what a stand-in almost never gets asked: does a publisher created before the shutdown return an error afterwards? A real client answers "not connected". A stand-in whose publish is a channel send has no reason to, and accepts the message instead.

tests/conformance_self.rs

// `make_source` / `make_publisher` must stay closures: their bounds are higher-ranked
// (`Fn(&str) -> _` / `Fn(&B) -> _`), so a bare method path - which binds one concrete lifetime -
// would not type-check.
#[allow(clippy::redundant_closure, clippy::redundant_closure_for_method_calls)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn memory_broker_passes_lifecycle() {
    harness::lifecycle(
        MemoryBroker::new,
        |name| MemorySource::new(name),
        |broker| broker.publisher(),
    )
    .await;
}

Add the capabilities::* suites the same way, one for each capability you implement.

Offer the capability surface the real broker offers. The testing feature is for tests, and a release build turns it off, which is what makes the two directions unequal. Falling short is the expensive direction: a capability the real broker has and the stand-in lacks cannot be mounted in process at all, so the behaviour behind it goes untested. Going over is the cheap one: a transaction or a request-reply that only the stand-in offers does not compile in your own release build, which is annoying and caught at once.

Settle the way the transport settles. The real ack returns AckError::Unsupported where the transport does not acknowledge: a fire-and-forget transport, an at-most-once quality of service. The stand-in returns the same. Answering Ok(()) to keep a suite quiet is how a handler returning HandlerOutcome::retry() passes in process and loses the message in production. The suites accept the honest answer.

Reproduce what the client does; do not fake what the broker does. The split is not about effort, it is about which side the behaviour runs on. Competing consumers, group distribution, correlation and reply routing, and buffering until commit are client-side or routing-level, and an in-process copy of them is exact. Cluster atomicity, fencing, broker-held timeouts and exactly-once are broker-side, and an in-process copy of them is fiction.

Competing consumers is the one to get right, because getting it wrong looks like success. Handing every message of a queue to every subscriber of that queue is a fan-out, not a queue. Two workers sharing one queue then each run the whole stream, and a test that counts what was processed sees the work done and reports no error.

Give every gap a comment naming the assertion it makes unsound. Do not write that the feature is missing; write which test a reader may no longer trust, and what covers it instead:

// No fencing: a second producer claiming the same transactional id is not rejected here, so a
// test cannot assert the first one is fenced out. `capabilities::transactions` against a real
// server is what covers that.

Pin the gap with a test as well. A comment goes stale the first time someone "fixes" the stand-in to route what it deliberately does not route. A test asserting the handler is not reached fails that day and explains itself.

Mount the stand-in with the production wiring. Your own subscription sources and publish policies have to work against it unchanged, so a service tests the routes file it ships. If a user must swap OrdersStream for something else to get a test running, the test no longer covers the mount.

Writing a broker¶

The required traits¶

Broker and ConnectedBroker¶

Subscribe¶

Subscriber¶

IncomingMessage¶

Publisher¶

PublishPolicy¶

Subscription sources¶

Who publishes a retry copy¶

What the registration declares¶

Where a retry copy is published¶

Naming a kind by one string¶

Settings in your own vocabulary¶

Publisher settings in your own vocabulary¶

Per-message settings on the publish builder¶

Capability traits¶

Batches: BatchSubscriber¶

The prelude your crate ships¶

Extending the Out slot vocabulary¶

Your crate's prelude¶

Per-delivery context and Ctx keys¶

Middleware on the async edges¶

Protocol bindings¶

Config and defaults¶

Errors¶

Test support¶

Writing one you can trust¶

`Broker` and `ConnectedBroker`¶

`Subscribe`¶

`Subscriber`¶

`IncomingMessage`¶

`Publisher`¶

`PublishPolicy`¶

Batches: `BatchSubscriber`¶

Extending the `Out` slot vocabulary¶

Per-delivery context and `Ctx` keys¶