Trait Broker 

pub trait Broker:
    Sized
    + Send
    + Sync {
    type Error: Error + Send + Sync + 'static;
    type Connected: ConnectedBroker;

    // Required method
    fn connect(
        self,
    ) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send;

    // Provided method
    fn bindable(self) -> Bindable<Self>
       where Self: 'static { ... }
}

An unconnected broker: configuration captured, no I/O performed yet.

Broker is the entry point of any broker crate (ruststream-nats, ruststream-kafka, …). The lifecycle is a ladder of consuming transitions, so each state is a distinct type and out-of-order calls do not compile:

B::new(config)                      unconnected: sync, I/O-free construction
broker.connect(self)  -> Connected  the live connection, a typed witness
connected.shutdown(self) -> Closed  the terminal witness (may carry diagnostics)

Subscribing is described separately by a SubscriptionSource (or the Subscribe capability for the by-name case), resolved against the Connected form. Publishers are likewise produced by broker-specific constructors.

Send + Sync is required so the runtime can move the broker across tasks.

Writing a broker crate

A broker is a crate of its own that implements the traits below. It depends on the core with the default features off, so it takes the trait surface and the runtime without the bundled JSON codec or any other broker:

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

MemoryBroker is the reference implementation of every trait on this list, and ruststream-nats (docs.rs) is a worked example over a real client.

The required surface:

• Broker and ConnectedBroker: the lifecycle ladder described here.
• Subscribe on the connected form, for subscriptions a name alone identifies, and a SubscriptionSource descriptor for the ones that need settings of the broker’s own.
• Subscriber, the stream of deliveries, and IncomingMessage, one delivery and its settlement.
• Publisher, the live publish handle, constructed from a PublishPolicy; DefaultPublish names the policy a reply and a retry copy use when the mount site names none.

Settings in the broker’s own vocabulary ride the mount chain: SubscriberBuilder for a subscription’s descriptor, MapPublisher for a publish policy, and Publisher::Options for one message.

Optional capabilities, each implemented only where the transport supports it and each checked by its own suite in conformance::capabilities:

Trait	For a broker that supports
BatchSubscriber	receiving messages in batches (offer it wherever you can)
TransactionalPublisher	begin, commit and abort around publishes on the publisher handle
OwnedTransactions / Transaction	many 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
Carries / CarriesBatch	lending the decoded value a delivery already holds
DescribeServer	reporting a ServerSpec for AsyncAPI

A body bounds its Out slot with the capability it needs (Out<impl TransactionalPublisher>) and never with a broker’s type, and the mount site checks the bound policy’s live form against it at compile time. Implementing the trait on the live publisher is all a service needs to reach the typed form of that capability. A live value that offers more than the core’s capabilities extends the vocabulary as OutSlot describes.

Around the traits: a per-delivery context with ContextField keys and a batch context through BuildBatchContext; protocol bindings for the AsyncAPI document on the descriptor and the policies (asyncapi::Bindings); a prelude of the crate’s own (prelude); the in-process mode under a testing feature (InProcess, TestableBroker, register_testable_broker!); and conformance, whose suites prove the crate and whose overview ends with the author checklist. Async work around encode and decode (a schema registry, an envelope) belongs on the async edges, not in a codec, as the codec overview describes.

Lazy startup contract

Implementations MUST be constructible synchronously, without performing I/O: expose a plain new(..) constructor that only captures configuration (addresses, credentials). All network setup happens in connect, which the runtime calls once at startup, after the synchronous #[ruststream::app] builder has run. This is what lets a service be assembled with the app macro regardless of broker. A broker that can only be built by connecting (an async “connect and return the handle” constructor) does not satisfy this contract. Each broker also ships a SubscriptionSource for its subjects, resolved against the connected form. conformance::harness::lifecycle checks the whole ladder: synchronous construction, connect, subscribe through the source, deliver, ack, shutdown, and the post-shutdown behaviour of aliased handles below.

The connected form holds the live client directly, so its operations never check a “maybe connected” state. A broker may also 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, and the cell serves those early handles.

Internal tasks run on the runtime the broker connected on

A task the broker starts on its own behalf (a delayed-redelivery timer, a reply dispatcher, a commit window) runs on the runtime connect ran on. The connected form keeps tokio::runtime::Handle::current() from connect and spawns every such task through it, whichever thread publishes, settles or requests. A handler on a dedicated thread publishes and settles from that thread’s own runtime: a task spawned there waits behind the handler’s computation and stops with that runtime, while the connection it serves lives on. conformance::harness::lifecycle publishes and settles with a delay from a runtime that stops at once, and conformance::capabilities::request_reply requests from one; both then expect the work to complete.

use std::time::Duration;

use tokio::runtime::Handle;
use tokio::sync::mpsc::UnboundedSender;
use tokio::time::sleep;

struct ConnectedExample {
    // Captured in `connect`.
    runtime: Handle,
}

impl ConnectedExample {
    fn requeue_after(&self, back: UnboundedSender<Vec<u8>>, delivery: Vec<u8>, delay: Duration) {
        self.runtime.spawn(async move {
            sleep(delay).await;
            let _ = back.send(delivery);
        });
    }
}

Shutdown is a type, not a flag

ConnectedBroker::shutdown consumes the connected broker, so misuse by the owner of the handle (a publish or subscribe after shutdown) is a compile error, not a runtime one. What remains dynamic is transport reality, not contract bookkeeping: handles aliasing the connection (publishers created before shutdown, clones of a shareable broker) MUST surface an error when used after the connection closed, never a silent success against a dead connection. The conformance lifecycle check verifies that aliased-handle behaviour.

shutdown never blocks or panics: every teardown step that can fail runs there and returns its error. The Closed witness carries teardown diagnostics as plain data, or is (); ClosedMemoryBroker reports how many subscriber registrations the shutdown dropped.

Configuration and errors

The crate owns its Config; the core carries no broker-specific configuration. A field with no sane default gets no Default, so a user sets it explicitly instead of inheriting a value that breaks later.

Errors use thiserror: one crate-level error enum with variants by source, marked #[non_exhaustive]. A library crate never uses anyhow.

Examples

use ruststream::{Broker, ConnectedBroker};

async fn ladder<B: Broker>(broker: B) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let connected = broker.connect().await?;
    let _closed = connected.shutdown().await?;
    Ok(())
}

Required Associated Types

type Error: Error + Send + Sync + 'static

The error type returned by broker-level operations.

type Connected: ConnectedBroker

The connected form of this broker: the typed witness that connect succeeded.

Required Methods

fn connect( self, ) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send

Establishes the connection to the broker, consuming the unconnected form.

The returned Connected value is the only way to reach the connection-bound surface (subscriptions, and for shareable brokers the live handles), so “not connected” is not representable on the owner’s path.

Errors

Returns Self::Error when the broker is unreachable, authentication fails, or the configuration is invalid.

Provided Methods

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

Provided for every broker; implementations do not override it.

Examples

use ruststream::memory::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "orders.accepted")]
struct Accepted {
    id: u64,
}

#[subscriber("orders", reply)]
async fn accept(order: &Order) -> Accepted {
    Accepted { id: order.id }
}

fn app() -> RustStream {
    // Orders arrive on one bus and their answers leave on another.
    let egress = MemoryBroker::new().bindable();
    let to_egress = egress.bind(Publish);
    RustStream::new(AppInfo::new("bridge", "0.1.0"))
        .with_broker_labeled("egress", egress, |_b| {})
        .with_broker_labeled("ingress", MemoryBroker::new(), |b| {
            b.include(accept).out_reply(to_egress);
        })
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl Broker for RedisBroker

type Error = RedisError

type Connected = ConnectedRedisBroker