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.

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.

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.

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<DB: QueueDatabase, D: Dialect + 'static> Broker for SqlxBroker<DB, D>

type Error = SqlxBrokerError

type Connected = ConnectedSqlxBroker<DB, D>