Struct RustStream 

pub struct RustStream<Layers = Identity, State = (), Pipeline = PublishIdentity, Phase = Wired> { /* private fields */ }

The top-level application object.

RustStream binds one or more brokers, the handlers attached to each, and the service lifecycle into a single runnable unit. Handlers are registered through with_broker, which hands a scope bound to that broker; nothing connects or subscribes until run. Brokers are held type-erased (only their lifecycle), so a single service can mix broker types.

The type parameter Layers is the global middleware stack applied to every per-message handler registered on a broker scope, whether directly or via include_router (batch handlers are the exception: a per-message layer cannot wrap a whole-batch handler); it defaults to the no-op Identity and grows as layer is called. Add all layers before with_broker, since a layer only applies to handlers registered after it.

The type parameter Phase tracks the builder phase at compile time: new starts in Setup, where the state (on_startup), the middleware stack (layer) and the publish pipeline (publish_layer) are still configurable; the first with_broker moves it to Wired, where those methods no longer exist - so a configuration call that would silently not apply to already-registered handlers does not compile. Phase defaults to Wired, the built service; Setup lives inside a builder chain and is rarely written out.

Examples

use ruststream::memory::prelude::*;
use ruststream::runtime::layers::TracingLayer;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

/// What `#[ruststream::app]` runs: a span around every delivery, one broker, one handler.
pub fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(TracingLayer::default())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
        })
}

Implementations

impl<Layers, State, Pipeline, Phase> RustStream<Layers, State, Pipeline, Phase>

where Layers: Send, State: Send + Sync + 'static,

pub async fn run(self) -> Result<(), RustStreamError>

Runs the service until an interrupt (SIGINT / SIGTERM) is received, then shuts down gracefully.

Errors

Returns RustStreamError if a broker fails to connect, a subscription fails to open, a dispatch task panics, or a broker fails to shut down.

pub async fn run_until<F>(self, shutdown: F) -> Result<(), RustStreamError>

where F: Future<Output = ()> + Send,

Runs the service until shutdown resolves, then shuts down gracefully.

Use this instead of run to drive shutdown from a caller-owned future (a name, a timeout, a test signal) rather than from process signals.

Errors

Returns RustStreamError if a broker fails to connect, a subscription fails to open, a dispatch task panics, or a broker fails to shut down.

pub async fn start(self) -> Result<RunningApp, RustStreamError>

Starts the service in the background and hands back a RunningApp handle.

Performs the same startup sequence as run - the on_startup state producer, broker connects, subscription opens, after_startup hooks - and resolves once the service is running, so a startup failure surfaces here, before the caller starts accepting its own traffic. Installs no signal handlers: the caller decides what stops the service, by calling RunningApp::shutdown.

Use this to run the service beside another foreground server (an HTTP framework) in the same process; when the service is the whole process, run / run_until stay the simpler form.

Errors

Returns RustStreamError if the state producer or an after_startup hook fails, a broker fails to connect, or a subscription fails to open. Brokers that had already connected are shut down (best effort, failures logged) before the error is returned, so a failed startup does not leak live connections.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

let app = RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new());
let running = app.start().await?;
// ... serve HTTP in the foreground ...
running.shutdown().await

impl RustStream<Identity, (), PublishIdentity, Setup>

pub fn new(info: AppInfo) -> RustStream<Identity, (), PublishIdentity, Setup>

Creates an empty service with the given metadata, no global middleware, and the unit application state (). Produce a typed state with on_startup.

impl<Layers, State, Pipeline> RustStream<Layers, State, Pipeline, Setup>

pub fn layer<N>( self, layer: N, ) -> RustStream<Stack<N, Layers>, State, Pipeline, Setup>

Adds a global middleware layer, applied to every handler registered later.

The first layer added runs outermost. Only available before the first with_broker: a layer added later could not wrap the handlers already registered, so that ordering does not compile.

pub fn on_startup<F, Fut, State2, E>( self, hook: F, ) -> RustStream<Layers, State2, Pipeline, Setup>

where F: FnOnce(State) -> Fut + Send + 'static, Fut: Future<Output = Result<State2, E>> + Send, State: Send + 'static, State2: Send + Sync + 'static, E: Error + Send + Sync + 'static,

Produces the typed application state at startup, transitioning the app’s state type from the previous State to State2.

The hook runs once before brokers connect; its future can await (open a database pool, connect a client), and the produced State2 is shared with every handler (read via Context::state) and the read-only lifecycle hooks. A failing hook aborts startup. The initial state is () (from new), so the first call’s hook receives ().

Only available before the first with_broker: the call fixes the app’s state type, and handlers registered against a different state type could not be carried across - so that ordering does not compile.

Panics

Panics if a lifecycle hook (after_startup, on_shutdown, after_shutdown) was registered first: hooks close over the state type and cannot be carried across the state change. Register hooks after on_startup.

Examples

A pool built at startup is the state the handlers read, and is closed at the end:

use std::sync::Arc;
use std::time::Duration;

use ruststream::memory::prelude::*;
use serde::Deserialize;

/// A stand-in for a connection pool.
struct Database;

impl Database {
    async fn connect(url: &str) -> Result<Self, std::io::Error> {
        let _ = url;
        Ok(Self)
    }

    async fn close(&self) {}
}

#[subscriber("orders")]
async fn handle(order: &Order, ctx: &mut Context<'_, (), Database>) -> HandlerOutcome {
    let _db: &Database = ctx.state();
    HandlerOutcome::ack()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| Database::connect("postgres://localhost/orders").await)
        .after_shutdown(async move |db: Arc<Database>| {
            db.close().await;
            Ok::<_, std::io::Error>(())
        })
        .shutdown_timeout(Duration::from_secs(10))
        .with_broker(MemoryBroker::new(), |b| {
            b.include(handle);
        })
}

pub fn publish_layer<M>( self, middleware: M, ) -> RustStream<Layers, State, PublishStack<M, Pipeline>, Setup>

where M: PublishLayer + Clone + 'static,

Adds an outgoing publish middleware, run on every publish a handler makes before it reaches the broker (a Confluent / Avro envelope, publish metrics, dead-letter): a reply(..) form’s reply and every message that leaves through an injected Out slot alike, each under the transforms its own mount site named. It composes into the pipeline type parameter, so the last one added wraps the rest and runs outermost (unlike the consume-side layer, where the first added is outermost); the middleware must be Clone (the pipeline is cloned into each publishing handler and each slot). Only available before the first with_broker: a middleware added later could not wrap the publishers already handed out, so that ordering does not compile.

A handler mounted through include_router is the one exception on the slot side: a router’s routes are typed before the app that mounts them, so its slots carry their own .out(marker, policy).transform(..) steps and not this chain (its replies still travel it).

impl<Layers, State, Pipeline, Phase> RustStream<Layers, State, Pipeline, Phase>

pub fn after_startup<F, Fut, E>( self, hook: F, ) -> RustStream<Layers, State, Pipeline, Phase>

where State: Send + Sync + 'static, F: FnOnce(Arc<State>) -> Fut + Send + 'static, Fut: Future<Output = Result<(), E>> + Send, E: Error + Send + Sync + 'static,

Adds a hook run after brokers connect and handlers are spawned (for example, to publish an initial message or signal readiness). A failing hook aborts startup.

pub fn on_shutdown<F, Fut, E>( self, hook: F, ) -> RustStream<Layers, State, Pipeline, Phase>

where State: Send + Sync + 'static, F: FnOnce(Arc<State>) -> Fut + Send + 'static, Fut: Future<Output = Result<(), E>> + Send, E: Error + Send + Sync + 'static,

Adds a hook run when shutdown begins, while brokers are still connected. Errors are logged.

pub fn after_shutdown<F, Fut, E>( self, hook: F, ) -> RustStream<Layers, State, Pipeline, Phase>

where State: Send + Sync + 'static, F: FnOnce(Arc<State>) -> Fut + Send + 'static, Fut: Future<Output = Result<(), E>> + Send, E: Error + Send + Sync + 'static,

Adds a hook run after brokers have shut down (for final async resource teardown). Errors are logged.

pub fn shutdown_timeout( self, timeout: Duration, ) -> RustStream<Layers, State, Pipeline, Phase>

Sets how long run waits for in-flight handlers to finish after shutdown is triggered. After the timeout, the remaining handler tasks are aborted. The same bound then applies to draining post-settle and_after continuations; on timeout they are abandoned (they are at-most-once side effects, so this loses follow-up work, never a settlement). Defaults to waiting indefinitely.

pub fn on_subscription_end( self, on_end: SubscriptionEnd, ) -> RustStream<Layers, State, Pipeline, Phase>

Sets what a subscription’s stream ending while the service runs does.

The default, SubscriptionEnd::Log, logs the end and lets the rest of the service run on; HealthProbe::subscriptions then counts one subscription fewer. SubscriptionEnd::FailFast tears the service down the way a fail-fast handler failure does. An end during shutdown is never reported.

Examples

use ruststream::memory::prelude::*;
use ruststream::runtime::SubscriptionEnd;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

pub fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_subscription_end(SubscriptionEnd::FailFast)
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
        })
}

pub fn register_broker<R>( self, broker: R, ) -> RustStream<Layers, State, Pipeline, Phase>

where R: BrokerRegistration,

Registers a broker for lifecycle management only (connect / shutdown), without attaching subscribers. Use for publish-only brokers.

pub fn server( self, name: impl Into<String>, spec: ServerSpec, ) -> RustStream<Layers, State, Pipeline, Phase>

Records an AsyncAPI server (one per broker the service connects to).

Build the ServerSpec directly, or get it from a broker that implements DescribeServer: app.server("nats", broker.describe_server()). build_spec emits these in the document’s servers section.

For a self-describing broker, prefer with_broker_labeled, which derives this entry from the broker under its label in one step; the in-memory broker describes itself too. Use this method for a broker that does not implement DescribeServer, or to override a labeled broker’s own spec.

pub fn with_broker<R, F>( self, broker: R, build: F, ) -> RustStream<Layers, State, Pipeline>

where R: BrokerRegistration, Layers: Clone, Pipeline: Clone, State: Send + Sync + 'static, F: FnOnce(&mut BrokerScope<<R as BrokerRegistration>::Broker, Layers, (), State, Pipeline>),

Registers a broker and the handlers attached to it.

build receives a BrokerScope typed to this broker; use it to attach handlers. The broker is then held for lifecycle management. Call this once per broker.

The scope carries no codec of its own, so macro handlers mounted with b.include(handle) decode with the DefaultCodec. To decode with another codec scope-wide, use with_broker_codec.

The first call moves the builder to the Wired phase: the state, the middleware stack, and the publish pipeline are fixed from here on, so a configuration call that could not apply to this broker’s handlers does not compile.

pub fn with_broker_codec<R, C, F>( self, broker: R, codec: C, build: F, ) -> RustStream<Layers, State, Pipeline>

where R: BrokerRegistration, C: Codec + Clone + 'static, Layers: Clone, Pipeline: Clone, State: Send + Sync + 'static, F: FnOnce(&mut BrokerScope<<R as BrokerRegistration>::Broker, Layers, C, State, Pipeline>),

Registers a broker with a scope-wide codec, replacing the DefaultCodec a codec-less scope decodes with.

build receives a BrokerScope whose include family reads the same as on a codec-less scope (b.include(handle)) but decodes with codec. A Router mounted in the scope can name a codec of its own with with_codec, which wins for its handlers.

Examples

use ruststream::codec::{CborCodec, JsonCodec};
use ruststream::memory::prelude::*;
use serde::Deserialize;

fn app() -> RustStream {
    RustStream::new(AppInfo::new("codecs", "0.1.0"))
        // per scope: every handler mounted here decodes with CBOR
        .with_broker_codec(MemoryBroker::new(), CborCodec, |b| {
            b.include(handle);
            // per handler: this mounting decodes with JSON
            b.include_router(Router::new().with_codec(JsonCodec).include(audit));
        })
}

pub fn with_broker_labeled<R, F>( self, label: impl Into<String>, broker: R, build: F, ) -> RustStream<Layers, State, Pipeline>

where R: BrokerRegistration, <R as BrokerRegistration>::Broker: DescribeServer, Layers: Clone, Pipeline: Clone, State: Send + Sync + 'static, F: FnOnce(&mut BrokerScope<<R as BrokerRegistration>::Broker, Layers, (), State, Pipeline>),

Registers a self-describing broker under label, along with the handlers attached to it.

The label is the broker’s stable identity in the service and the name of its AsyncAPI server: the broker’s describe_server coordinates are recorded in the servers map under label, so the document stays in sync with the brokers actually mounted, with no separate server call. An explicit server(label, spec) entry for the same label takes precedence.

Like with_broker, the scope decodes with the DefaultCodec; use with_broker_labeled_codec to name another.

Examples

use ruststream::runtime::{AppInfo, RustStream};
use ruststream::{Broker, DescribeServer, ServerSpec};

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker_labeled("ingress", NatsBroker::new("nats://localhost"), |_b| {});
// The AsyncAPI `servers` map now carries "ingress" with the broker's host / protocol.
app.run().await

pub fn with_broker_labeled_codec<R, C, F>( self, label: impl Into<String>, broker: R, codec: C, build: F, ) -> RustStream<Layers, State, Pipeline>

where R: BrokerRegistration, <R as BrokerRegistration>::Broker: DescribeServer, C: Codec + Clone + 'static, Layers: Clone, Pipeline: Clone, State: Send + Sync + 'static, F: FnOnce(&mut BrokerScope<<R as BrokerRegistration>::Broker, Layers, C, State, Pipeline>),

Registers a self-describing broker under label with a default codec.

Combines with_broker_labeled (the label is the broker’s identity and AsyncAPI server name) with with_broker_codec (the scope’s macro handlers decode with codec).

pub fn handlers(&self) -> &[HandlerMetadata]

Returns metadata for every registered handler, in registration order. Input to the AsyncAPI generator.

pub fn info(&self) -> &AppInfo

Returns the service metadata.

pub fn servers(&self) -> &BTreeMap<String, ServerSpec>

Returns the registered AsyncAPI servers, keyed by name. Input to the AsyncAPI generator.

Trait Implementations

impl<Layers, State, Pipeline, Phase> App for RustStream<Layers, State, Pipeline, Phase>

where Layers: Send, State: Send + Sync + 'static, Pipeline: Send,

fn run(self) -> impl Future<Output = Result<(), RustStreamError>> + Send

Runs the service until an interrupt (SIGINT / SIGTERM), then shuts down gracefully.

fn run_until<F>( self, shutdown: F, ) -> impl Future<Output = Result<(), RustStreamError>> + Send

where F: Future<Output = ()> + Send,

Runs the service until shutdown resolves, then shuts down gracefully.

fn start( self, ) -> impl Future<Output = Result<RunningApp, RustStreamError>> + Send

Starts the service in the background and hands back a RunningApp handle: the side-by-side form for hosting the service next to another foreground server. See RustStream::start.

fn info(&self) -> &AppInfo

The service metadata (title, version, description): the AsyncAPI info object.

fn servers(&self) -> &BTreeMap<String, ServerSpec>

The registered AsyncAPI servers, keyed by name.

fn handlers(&self) -> &[HandlerMetadata]

Metadata for every registered handler, in registration order: input to the AsyncAPI generator.

impl<Layers, State, Pipeline, Phase> Debug for RustStream<Layers, State, Pipeline, Phase>

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.