Struct TestApp 

pub struct TestApp<State> { /* private fields */ }

A test harness around the service’s production app.

Takes the app main builds, unchanged. start connects each of its brokers through the broker’s InProcess transition instead of connect, so the same connected forms carry the routes with no server; start_live connects them through connect, against a running stand. Either way the harness drives input through the broker, records what handlers saw and what the app published, and exposes per-broker assertions addressed by the production broker type, so one test body runs in both modes and only the start call differs.

Build one with start (runs the app’s real on_startup) or with_state (injects a mirror state for non-broker dependencies). Drive input with broker / broker_named and assert on what happened.

Every subscription loop and every worker of workers(n) runs as a task on the test’s own runtime, whatever runtime the generated main builds, and a threads(n) subscription runs as n such workers rather than on threads of its own, so advance reaches every timer the app arms and settle sees every delivery.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerOutcome, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

#[derive(Outgoing, Serialize, Deserialize, PartialEq, Debug)]
struct Order {
    id: u64,
}

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

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker(MemoryBroker::new(), |b| { b.include(handle); });
let tb = TestApp::start(app).await?;

tb.broker::<MemoryBroker>()
    .message(&Order { id: 1 })
    .to("orders")
    .publish()
    .await?;
tb.broker::<MemoryBroker>()
    .subscriber("orders")
    .assert_called_once()
    .with(&Order { id: 1 })
    .settled(HandlerOutcome::ack());

Implementations

impl<State: Send + Sync + 'static> TestApp<State>

pub async fn start<A>(app: A) -> Result<Self, TestError>

where A: App<State = State>,

Starts the harness in the service’s own startup order: the app’s real on_startup, then each broker’s connect, then the subscriptions, then after_startup. Each broker connects through its InProcess transition, which performs no I/O; the rest of startup is the production one, so an on_startup that fails leaves no broker connected.

Errors

Returns TestError::NoTransport if a broker of the app has no in-process mode, TestError::Startup if a lifecycle hook fails, TestError::Connect if a broker fails to connect in process, or TestError::Subscribe if a subscription fails to open.

The app’s publish pipeline is whatever it was built with: an app-wide publish_layer is already baked into the mounted routes, so the harness drives the same stack production would.

Examples

The publish drives the whole reaction to completion before it returns: the handler, its downstream publishes, any cross-broker cascade. Then the test asserts.

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

#[derive(Debug, Deserialize, Outgoing, Serialize, PartialEq)]
struct Order {
    id: u64,
    quantity: u32,
}

#[derive(Debug, Deserialize, Outgoing, Serialize, PartialEq)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
    accepted: bool,
}

#[subscriber("orders", reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

/// The app `main` runs, and the one the tests hand the harness.
pub fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.0.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(confirm).out_reply(Publish);
    })
}

pub async fn confirms_valid_orders() -> Result<(), Box<dyn std::error::Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<MemoryBroker>()
        .message(&Order { id: 1, quantity: 2 })
        .to("orders")
        .publish()
        .await?;

    tb.broker::<MemoryBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&Order { id: 1, quantity: 2 })
        .settled(HandlerOutcome::ack());
    tb.broker::<MemoryBroker>()
        .published::<Confirmation>("confirmations")
        .assert_called_once()
        .with(&Confirmation {
            id: 1,
            accepted: true,
        });
    Ok(())
}

pub async fn start_live<A>(app: A) -> Result<Self, TestError>

where A: App<State = State>,

Starts the harness against running brokers: each broker of the app connects through its ordinary connect, and the rest of startup is the production one, in its order: the app’s real on_startup runs before any broker connects.

The test body is the one an in-process test runs; only this call differs. What differs underneath is how long things take. A publish, settle and advance wait on the broker’s real clock, up to a deadline of ten seconds (start_live_within sets another), for the subscriptions to handle what was published to them and what fell due.

Errors

Returns TestError::PausedClock on a paused clock, TestError::Startup if a lifecycle hook fails, TestError::Connect if a broker fails to connect, or TestError::Subscribe if a subscription fails to open.

What a live test sees

What changes live is how the harness knows the reaction is over. It cannot read a broker’s queues, so it records what is published through the app’s own wiring: the test’s publishes, and every message a publisher the runtime paired for the app’s wiring hands a broker. That covers replies, Out slot publishes and requests, retry copies, dead letters, and a Bound token’s publishes through any of them. Each is recorded on the broker it goes to: the one the test named, or the one the publisher was paired against. A handler on one broker holding a Bound token for another publishes to the other, and the publish is that broker’s.

A publish, a settle and an advance then wait until every subscription has handled each recorded publish its broker delivers to it, and every redelivery that fell due, and no handler is running. Which subscriptions a publish reaches is the broker’s own routing, which the broker answers through TestableBroker::routes: an equal name for MemoryBroker, every matching subscription for a broker that fans out, the one it picks for a broker that picks one. A subscription of another broker never owes it. Each subscription counts what it handled itself, so two subscriptions reporting one name (two subscriptions on one topic) each owe the publish. The wait is bounded: ten seconds by default, start_live_within sets another, and past it the call fails with TestError::NotSettled naming the subscription it was waiting on. A live advance lets the time pass for real, because the delay is the broker’s own timer, so a live test runs on the running clock, and start_live refuses a paused one.

A publisher the service uses itself is outside this record: one kept in its state, one an after_startup hook is handed or pairs, one from Bound::live or RunningApp::publisher, and one a broker crate builds outside the runtime’s pairing. The harness neither waits for its publishes nor lists them; a test sends its own input through tb.broker::<B>().message(..).

subscriber(..) and settled(..) read the same record in both modes. published::<T>(name) reads the broker’s log in process and the harness’s record live: the test’s publishes onto that broker, and the app’s publishes that went to it, each as the framework handed it over. A test that runs in both modes asserts on those. What only the broker does is asserted in process: a setting its publisher turns into a header or a prefix, and a transaction, which commits on the broker. What a broker moves on its own (a native delayed redelivery, its own dead-lettering) is asserted through the subscription that receives it. Each mode covers what the other cannot: in process the broker’s server-side semantics are the transport’s model of them, and live they are the server’s own.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerOutcome, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

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

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

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

let tb = TestApp::start_live(app()).await?;
tb.broker::<MemoryBroker>().message(&Order { id: 1 }).publish().await?;
tb.broker::<MemoryBroker>().subscriber("orders").assert_called_once();
tb.shutdown().await?;

pub async fn start_live_within<A>( app: A, deadline: Duration, ) -> Result<Self, TestError>

where A: App<State = State>,

start_live with deadline as the longest a publish, a settle or an advance waits for the reaction to settle.

Errors

As start_live.

Examples

use std::time::Duration;

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

// A stand far away: give the reaction half a minute before the test gives up on it.
let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
    .register_broker(MemoryBroker::new());
let tb = TestApp::start_live_within(app, Duration::from_secs(30)).await?;
tb.shutdown().await?;

pub async fn with_state<A, F>(app: A, build: F) -> Result<Self, TestError>

where A: App<State = State>, F: FnOnce(&TestBrokers<'_>) -> State,

Starts the harness with an injected mirror state, instead of running the app’s on_startup. build receives the brokers so it can wire the mirror state’s publishers onto the same transports (tb.broker::<MemoryBroker>().publisher()) and supply fakes for non-broker dependencies. Each broker connects in process first, so the mirror state’s publishers pair against connected brokers.

Errors

Returns TestError::NoTransport if a broker of the app has no in-process mode, TestError::Connect if a broker fails to connect in process, or TestError::Subscribe if a subscription fails to open.

pub fn broker<B>(&self) -> BrokerHandle<'_>

where B: Broker + 'static, B::Connected: DefaultPublish, <<B::Connected as DefaultPublish>::Policy as PublishPolicy<B::Connected>>::Live: Publisher,

Addresses the unique broker of type B, the production type the app was built on.

Live, a test’s input reaches the broker through the connected form’s DefaultPublish policy, which is what the bound asks for.

Panics

Panics if no broker of type B is registered, or more than one is (address by label with broker_named instead).

pub fn broker_named(&self, label: &str) -> BrokerHandle<'_>

Addresses the broker registered under label (see with_broker_labeled).

Panics

Panics if no broker carries label.

pub fn out<M: OutSlot>(&self) -> PublishedAssertions<()>

Asserts on what was published through the Out slot marked M: exactly the messages the handler sent through that injected publisher, with their destinations and headers, across all brokers.

The untyped assertions (assert_called_once, with_raw, …) apply directly; decode the payloads with decoded_as for the typed with form.

Publishes made outside the handler task (a spawned sibling task, a settled owned transaction’s buffer) are not attributed to the slot.

Examples

use ruststream::memory::{MemoryBroker, MemoryPublish};
use ruststream::runtime::{AppInfo, HandlerOutcome, Out, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Deserialized, OutSlot, Outgoing, Publisher, Serialized, subscriber};

#[derive(Deserialized)]
struct Chunk<'a>(&'a [u8]);

// The frame on the way out: a serialized type carries its own bytes, so what the
// handler republishes is byte-for-byte what it received.
#[derive(Outgoing, Serialized)]
struct Frame(Vec<u8>);

#[derive(OutSlot)]
#[publishes(Frame)]
struct Encoded;

#[subscriber("chunks")]
async fn transcode(
    chunk: &Chunk<'_>,
    Out(out): Out<impl Publisher, Encoded>,
) -> HandlerOutcome {
    let frame = Frame(chunk.0.to_vec());
    if out.message(&frame).to("encoded").publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker(MemoryBroker::new(), |b| {
        b.include(transcode).out(Encoded, MemoryPublish).build();
    });
let tb = TestApp::start(app).await?;
tb.broker::<MemoryBroker>()
    .message(&Frame(b"frame".to_vec()))
    .to("chunks")
    .publish()
    .await?;
tb.out::<Encoded>().assert_called_once().with_raw(b"frame");

pub async fn publish<T: Serialize + Sync>( &self, name: &str, value: &T, ) -> Result<(), TestError>

Publishes to the only registered broker, a convenience for single-broker apps.

Errors

Returns TestError::Ambiguous when more than one broker is registered (use broker / broker_named), or any error from BrokerHandle::publish.

pub fn message<'a, T>( &'a self, value: &'a T, ) -> PublishBuilder<InjectSink<'a>, MessageBody<'a, T>, UnnamedCodec, HeadersUnset, T::Form>

where T: OutgoingDestination,

Starts a typed injection on the only registered broker, a convenience for single-broker apps: tb.message(&order).to("orders").publish().await?.

The scoped BrokerHandle::message with the broker chosen for you. An app registering more than one broker has no single target, so the publish reports TestError::Ambiguous and the test addresses a broker with broker / broker_named instead.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerOutcome, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

#[derive(Outgoing, Serialize, Deserialize)]
#[outgoing(name = "orders")]
struct Order {
    id: u32,
}

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

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker(MemoryBroker::new(), |b| { b.include(handle); });
let tb = TestApp::start(app).await?;

tb.message(&Order { id: 7 }).publish().await?;

pub async fn settle(&self) -> Result<(), TestError>

Drives any in-flight reaction to a standstill (handlers run, their publishes cascade) without publishing anything new. A publish through the harness calls this for you; use it after a reaction the test started some other way.

In process it waits for the transports to report nothing in flight. Live, where nothing in this process can read the broker’s queues, it waits until every subscription has handled what the harness saw published to it and every redelivery that fell due, and no handler is running, up to the deadline the harness was started with.

Errors

Returns TestError::NotQuiescent if an in-process reaction does not settle within the step budget, or TestError::NotSettled naming the subscription a live reaction left unsettled at the deadline.

pub async fn advance(&self, by: Duration) -> Result<(), TestError>

Lets by pass, fires every nack_after / retry_after redelivery now due, and drives the resulting reaction to a standstill. Use it to test delayed redeliveries: publish records the immediate NackAfter settlement and returns; advance then delivers the message again. A broker’s native delayed redelivery and the runtime’s deferred re-publish both arrive this way, in either mode; the copy the runtime publishes for a zero delay does not wait for it.

In process it moves the paused clock by by, so an in-process test runs on a paused clock (#[tokio::test(start_paused = true)] or tokio::time::pause; on a running clock tokio::time::advance panics). Live it lets by of real time pass, then settles, waiting for every redelivery that fell due by then.

Errors

Returns TestError::NotQuiescent if an in-process reaction does not settle within the step budget, or TestError::NotSettled naming the subscription a live reaction left unsettled at the deadline.

pub async fn drain(&self)

Waits for the post-settle continuations of everything settled so far to finish, for tests that assert on their side effects. Synchronous handler effects need only settle.

A settlement registers its continuations - the outcome’s and_after and the context’s after(..) / after_settle(..) hooks alike - before it releases the delivery, so an injection that has returned has already put its continuations here for this to find.

pub fn run_result(&self) -> Result<(), RustStreamError>

The result the real run would return: Ok while the service is healthy, or RustStreamError::Dispatch once a fail-fast failure tore it down.

Errors

Returns RustStreamError::Dispatch when a handler panic (or a fail-fast decode failure) triggered shutdown.

pub fn assert_running(&self)

Asserts the service is still running (no fail-fast shutdown was triggered).

Panics

Panics if a fail-fast failure has torn the service down.

pub fn assert_shut_down(&self)

Asserts a fail-fast failure has shut the service down.

Panics

Panics if the service is still running.

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

Shuts the harness down the way RunningApp::shutdown shuts the service down: the on_shutdown hooks, then the dispatch loops, the in-flight handlers and post-settle continuations (bounded by the app’s shutdown timeout), the brokers in reverse registration order, and the after_shutdown hooks last. A failing shutdown hook is logged and the shutdown goes on.

Errors

Returns RustStreamError::Dispatch when a fail-fast failure tore the service down, RustStreamError::Shutdown when a broker fails to shut down, and RustStreamError::Join when a dispatch task panicked.

Trait Implementations

impl<State> Debug for TestApp<State>

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

Formats the value using the given formatter.