Struct Router 

pub struct Router<B, Routes = (), C = (), Layers = Identity, Pipe = PublishIdentity> { /* private fields */ }

A statically-typed, lazily-bound group of handler registrations, not attached to any broker.

Build it by chaining (each call consumes the router and returns a new type that carries the added registration), then mount it with include_router. The registration list Routes is an opaque nested tuple, so a builder function returns impl RouterDef<B> instead of naming the type.

The codec parameter C is the codec the include family decodes with. It starts as (), meaning the DefaultCodec; switch it for the rest of the chain with with_codec.

The layer parameter Layers is the router’s own middleware stack, grown with layer. It wraps every handler in the router when the router is mounted, inside the app’s global stack.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{Context, Handle, HandlerOutcome, Router, RouterDef, subscriber};

#[derive(serde::Deserialize, schemars::JsonSchema)]
struct Event {
    id: u64,
}

struct Audit;

impl Handle<Event> for Audit {
    async fn handle(
        &self,
        event: &Event,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let _ = event.id;
        Ok(())
    }
}

fn routes() -> impl RouterDef<MemoryBroker> {
    Router::<MemoryBroker>::new().include(subscriber("events", Audit).build())
}
// later: app.with_broker(broker, |b| b.include_router(routes()));

Implementations

impl<B> Router<B>

where B: Broker + 'static,

pub fn new() -> Router<B>

Creates an empty router decoding with the DefaultCodec.

impl<B, RouteCodec, RouteLayers, RoutePipe> Router<B, (), RouteCodec, RouteLayers, RoutePipe>

where B: Broker + 'static,

pub fn layer<N>( self, layer: N, ) -> Router<B, (), RouteCodec, Stack<N, RouteLayers>, RoutePipe>

Adds a router-scope middleware layer, wrapping every handler in this router when the router is mounted. The first layer added runs outermost within the router; the app’s global layer stack wraps outside it.

The layer must be a BlanketLayer (it applies to handlers whose concrete types the router hides), like the app-global stack, and it is declared before the router’s first registration: after one, .layer(..) rides that registration instead (Router::layer on a registration), exactly as the other steps of the chain ride the position named before them.

impl<B, Routes, RouteCodec, RouteLayers, RoutePipe> Router<B, Routes, RouteCodec, RouteLayers, RoutePipe>

where B: Broker + 'static,

pub fn with_codec<C>( self, codec: C, ) -> Router<B, Routes, C, RouteLayers, RoutePipe>

Sets the codec that subsequent include calls decode with, replacing the default.

Registrations already in the chain keep the codec they were mounted with, so the codec can change mid-chain.

pub fn merge<R2, C2, L2, P2>( self, other: Router<B, R2, C2, L2, P2>, ) -> Router<B, (Router<B, R2, C2, L2, P2>, Routes), RouteCodec, RouteLayers, RoutePipe>

where L2: BlanketLayer,

Appends every registration of other after this router’s own, keeping each router’s codec and layer stack.

The merged router’s handlers stay wrapped by its own layers; this router’s layers wrap around them as well once mounted (scopes nest).

Examples

Two modules’ routers mounted as one group:

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

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    HandlerOutcome::ack()
}

#[subscriber("shipments")]
async fn dispatch(shipment: &Shipment) -> HandlerOutcome {
    HandlerOutcome::ack()
}

fn orders() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    Router::new().include(accept)
}

fn shipping() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    Router::new().include(dispatch)
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("routing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include_router(orders().merge(shipping()));
    })
}

pub fn handle<S, H>( self, subscriber: S, handler: H, meta: HandlerMetadata, ) -> Router<B, (HandleRoute<S, H>, Routes), RouteCodec, RouteLayers, RoutePipe>

where S: Subscriber + Send + 'static, H: Handler<<S as Subscriber>::Message> + 'static,

Attaches handler to an already-created subscriber. Machinery, not the user path: a service mounts definitions with include and the value constructors (subscriber, …).

The subscriber is created up front (before connect), the handler arrives fully wired (a decode adapter included, see typed), and the metadata is the caller’s to assemble. That is the shape the runtime’s own dispatch tests and a broker author holding a hand-built subscriber need, and nothing above it.

impl<B, S, H, Cx, Routes, RouteCodec, RouteLayers, RoutePipe> Router<B, (SubscribeRoute<S, H, Cx>, Routes), RouteCodec, RouteLayers, RoutePipe>

pub fn workers( self, workers: Workers, ) -> Router<B, (SubscribeRoute<S, H, Cx>, Routes), RouteCodec, RouteLayers, RoutePipe>

Sets the concurrency policy of the registration just added (the preceding include call), replacing its default.

The functional-path counterpart of the macro’s workers(..) clause: Workers::pool processes up to n deliveries of this subscriber concurrently, Workers::keyed dispatches over per-key sequential lanes. On an included definition this overrides the attribute’s workers(..) clause and the chained workers setting alike.

Examples

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

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

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "done");
    HandlerOutcome::ack()
}

fn jobs() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    // Four lanes; the jobs of one key stay in order on one lane.
    Router::new()
        .include(work)
        .workers(Workers::keyed(nonzero!(4)))
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include_router(jobs());
    })
}

pub fn layer<N>( self, layer: N, ) -> Router<B, (SubscribeRoute<S, <N as Layer<H>>::Handler, Cx>, Routes), RouteCodec, RouteLayers, RoutePipe>

where N: Layer<H>,

Wraps the registration just added (the preceding include call) with layer, outside its decode step, so the layer sees the raw delivery.

The per-registration counterpart of the app-wide RustStream::layer and the router-wide layer on a fresh router: those two wrap every handler in their scope, so they take a BlanketLayer, which cannot be written for a layer fixed to one message type. This one wraps exactly one registration, whose handler type is still concrete here, so it takes an ordinary Layer - which is what a DynStack over the broker’s own message type is.

Which of the two a .layer(..) call is follows what the chain named before it, like every other step: on a router with no registrations yet it is the router’s own, and after an include it is that registration’s. It repeats: each call wraps what the calls before it produced, the last one outermost.

Examples

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

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

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "done");
    HandlerOutcome::ack()
}

fn jobs() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    // Every handler of this router runs inside the tracing layer's span.
    Router::new().include(work).layer(TracingLayer::default())
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include_router(jobs());
    })
}

impl<B, S, H, Cx, Routes, RouteCodec, RouteLayers, RoutePipe> Router<B, (BatchRoute<S, H, Cx>, Routes), RouteCodec, RouteLayers, RoutePipe>

pub fn workers( self, workers: Workers, ) -> Router<B, (BatchRoute<S, H, Cx>, Routes), RouteCodec, RouteLayers, RoutePipe>

Sets the concurrency policy of the batch registration just added (the preceding batch include call), replacing its default.

Workers::pool keeps up to n batches in flight at once. Keyed lanes order single messages per key and do not apply to batches: a Workers::keyed policy here behaves like a plain pool of the same size.

impl<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe> Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>

where B: Broker + 'static,

pub fn out<M, Policy>( self, marker: M, policy: Policy, ) -> <M as RoutePosition<B, Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>, Policy>>::Out

where M: RoutePosition<B, Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>, Policy>,

Names the publish policy of one position of the registration just added (the preceding include call).

A form that publishes nothing of its own is a finished route by the time include returns, and the one position such a registration still has is the deferred retry - see Retry. Naming it opens a mount chain over that route, so the steps after it are the slot steps and .build() hands the router back.

Name it after that registration’s other settings (workers, layer): those read the route this call wraps.

Examples

use std::time::Duration;

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

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

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "done");
    HandlerOutcome::retry_after(Duration::from_secs(30))
}

fn jobs() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    // The copies a `retry_after` publishes leave through this policy.
    Router::new().include(work).out(Retry, Publish).build()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include_router(jobs());
    })
}

pub fn out_retry<Policy>( self, policy: Policy, ) -> <Retry as RoutePosition<B, Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>, Policy>>::Out

where Retry: RoutePosition<B, Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>, Policy>,

The same call with the marker spelled out: .out(Retry, policy).

It opens the same mount chain, so the slot steps follow it and .build() finishes it.

pub fn max_attempts( self, attempts: NonZero<u32>, ) -> RouterWith<DeclareMount, Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>, (), Declaring<(NoReply, ()), Present>>

where Head: RetryOpen,

Caps the deliveries one message of the registration just added gets, counting the first.

See RouterWith::max_attempts: the declaration is the same on every form, and this is the call for a registration that was a finished route the moment include returned. It opens a mount chain, so .dead_letter(..), .out_retry(..) and .build() follow it.

pub fn dead_letter( self, destination: impl Into<Cow<'static, str>>, ) -> RouterWith<DeclareMount, Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>, (), Declaring<(NoReply, ()), Absent, Present>>

where Head: RetryOpen,

Names where a delivery of the registration just added goes once its attempts run out.

See RouterWith::dead_letter.

impl<B, Routes, C, Layers, Pipe> Router<B, Routes, C, Layers, Pipe>

where B: Broker + 'static, Routes: RouterHandlers,

pub fn handlers(&self) -> Vec<HandlerMetadata>

Returns metadata for every registered handler, in registration order.

impl<B, Routes, RouteCodec, RouteLayers, RoutePipe> Router<B, Routes, RouteCodec, RouteLayers, RoutePipe>

where B: Broker + 'static,

pub fn include<D>( self, def: D, ) -> <<D as Declared>::Form as RouterMount<Router<B, Routes, RouteCodec, RouteLayers, RoutePipe>, <D as Declared>::Settings>>::Out

where D: Declared, <D as Declared>::Form: RouterMount<Router<B, Routes, RouteCodec, RouteLayers, RoutePipe>, <D as Declared>::Settings>,

Mounts a #[subscriber] definition of any form, on the source the definition names: a plain or batch handler grows the router directly, a reply("dest") or Out-taking one hands back a mount chain to finish with .build(), naming a publish policy per position with .out(Reply, policy) / .out(marker, policy) first.

A batch definition’s subscriber must implement BatchSubscriber - natively, or through the Buffered adapter; router and app middleware wrap per-message handlers and do not apply to batch registrations.

Decoding uses the chain’s codec when one was set with with_codec, else the DefaultCodec. The router-level counterpart of BrokerScope::include.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
use ruststream::subscriber;

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    let _ = order.id;
    HandlerOutcome::ack()
}

fn routes() -> impl RouterDef<MemoryBroker> {
    Router::<MemoryBroker>::new().include(handle)
}

Trait Implementations

impl<B, Routes, C, Layers, Pipe> Debug for Router<B, Routes, C, Layers, Pipe>

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

Formats the value using the given formatter.

impl<B> Default for Router<B>

where B: Broker + 'static,

fn default() -> Router<B>

Returns the “default value” for a type.

impl<B, Routes, C, Layers, Pipe, State, RetryPipeline> RouterDef<B, State, RetryPipeline> for Router<B, Routes, C, Layers, Pipe>

where B: Broker + 'static, Routes: RouterDef<B, State, Pipe>, Layers: BlanketLayer + Clone + Send + Sync + 'static,

impl<B, Routes, C, Layers, Pipe> RouterHandlers for Router<B, Routes, C, Layers, Pipe>

where Routes: RouterHandlers,