RustStream

https://github.com/powersemmi/ruststream
Admin
An async messaging framework for Rust providing broker-agnostic traits, a router runtime, multiple...
Tokens:499,603
Snippets:2,858
Trust Score:7.9
License:Apache-2.0
Update:4 hours ago
Show doc for...
Tokens:
Raw
### Run the service with cargo run -- run

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/quickstart.md

Starts the service on a tokio runtime. The service runs until Ctrl+C is pressed and needs no external broker. The template enables the logging feature, so logs print to the terminal via tracing.

```bash
cargo run -- run                # or: ruststream run, with the CLI installed
```

--------------------------------

### Run the example and place an order

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/http.md

Commands to run the example with the required features and send a POST /orders request with an x-request-id header and order JSON.

```bash
cargo run --example guide_http --features testing,macros,memory,json,cbor,asyncapi,logging
curl -X POST http://127.0.0.1:8080/orders -H 'content-type: application/json' \
  -H 'x-request-id: req-1' \
  -d '{"id":1,"customer":"ada@example.com","lines":[{"sku":"mug","quantity":12,"price_cents":1200}]}'
```

--------------------------------

### Scaffold a project with cargo generate

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/quickstart.md

Installs cargo-generate and scaffolds a new service from the memory template. The template creates an idiomatic multi-file project with Cargo.toml and src/main.rs, src/orders.rs, and src/routes.rs.

```bash
cargo install cargo-generate
cargo generate --git https://github.com/powersemmi/ruststream templates/memory --name my-service
cd my-service
```

--------------------------------

### Process: main with bindable and publisher (Macros)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/http.md

Sets up the process: bindable takes a publisher token before the shop starts, publisher turns it into a live publisher once the broker is connected. Server stops on Ctrl+C or when stopping reports the shop stopped. After HTTP drains, stops the relay; its last round hands over pending rows. Shop drains with shutdown. MemoryBroker keeps queues in process memory, so unhandled orders are lost on process exit. A hand-written main installs the logger with ruststream::logging::init().

```rust
```rust title="http.rs" reference="crates/ruststream/examples/guide/http.rs#main"
```
```

--------------------------------

### Main application setup (src/main.rs) - Macros API

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Main application setup for the tutorial using the macros API. This block is part of the router organization step and is expected to include the routes module and configure the router.

```rust
```rust title="src/main.rs" reference="crates/ruststream/examples/tutorial/main.rs#main"
```
```

--------------------------------

### Main application setup (src/main.rs) - Manual API

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Main application setup for the tutorial using the manual API. This block is part of the router organization step and is expected to include the routes module and configure the router.

```rust
```rust title="src/main.rs" reference="crates/ruststream/examples/manual/tutorial/main.rs#main"
```
```

--------------------------------

### Metrics example app (Manual)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/metrics.md

The app setup for the metrics example using the manual API. This is the manual variant of the example.

```rust
```rust reference="crates/ruststream/examples/manual/metrics.rs#app"
```
```

--------------------------------

### Build command for MessagePack codec example

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/msgpack.md

Runs the codec_msgpack example with MessagePack as the only codec feature, using the testing, macros, memory, and msgpack features.

```text
cargo test --example codec_msgpack --no-default-features --features testing,macros,memory,msgpack
```

--------------------------------

### Run the otel_export example

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/otel-export.md

Runs the example with the required features. Without a collector on localhost:4317, the exporter retries in the background and the service keeps working.

```text
cargo run --example otel_export --features testing,macros,memory,json,otel -- run
```

--------------------------------

### Process: main with bindable and publisher (Manual)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/http.md

Manual API variant of the process setup. Same behavior as the Macros tab.

```rust
```rust title="http.rs" reference="crates/ruststream/examples/manual/guide/http.rs#main"
```
```

--------------------------------

### Run the service with cargo run --example guide_first

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/service.md

Starts the service with the required features: testing, macros, memory, json, and logging. The logging feature prints service logs to the terminal. The service connects the broker, opens the subscription, and runs until Ctrl+C.

```bash
cargo run --example guide_first --features testing,macros,memory,json,logging -- run
```

--------------------------------

### TestApp::start_live

Source: https://github.com/powersemmi/ruststream

Starts a live test app with a settle deadline.

```APIDOC
## TestApp::start_live

### Description
Starts a live test app with settle deadline `DEFAULT_SETTLE_DEADLINE = 10 s` (override with `start_live_within`). Step budget for quiescence: `DEFAULT_MAX_STEPS = 10_000` dispatched deliveries.

### Modes
- **`start`**: in-process (`InProcess` transition of each broker).
- **`start_live`**: connects for real.
```

--------------------------------

### Build command for JSON codec example

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/json.md

Runs the codec_json example with JSON as the only codec, disabling default features and enabling testing, macros, memory, and json features.

```text
cargo test --example codec_json --no-default-features --features testing,macros,memory,json
```

--------------------------------

### Build command for CBOR codec example

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/cbor.md

Runs the codec_cbor example with CBOR as the only codec feature, disabling default features and enabling testing, macros, memory, and cbor.

```text
cargo test --example codec_cbor --no-default-features --features testing,macros,memory,cbor
```

--------------------------------

### Metrics example app (Macros)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/metrics.md

The app setup for the metrics example using the macros API. This is the macro-based variant of the example.

```rust
```rust reference="crates/ruststream/examples/metrics.rs#app"
```
```

--------------------------------

### Run asyncapi_content_type example with cbor feature

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/asyncapi/reports/content-type.md

Runs the example that demonstrates how the root defaultContentType appears only when all messages agree on the same media type. Requires the cbor feature along with macros, memory, and asyncapi features.

```text
cargo test --example asyncapi_content_type --features macros,memory,asyncapi,cbor
```

--------------------------------

### Build command for custom codec example

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/custom-codec.md

Runs the custom codec example with only the msgpack codec feature enabled, using testing, macros, memory, and msgpack features.

```text
cargo test --example custom_codec --no-default-features --features testing,macros,memory,msgpack
```

--------------------------------

### Install RustStream CLI with cli feature

Source: https://github.com/powersemmi/ruststream/blob/main/README.md

Installs the RustStream CLI from crates.io with the cli feature enabled.

```bash
cargo install ruststream --features cli
```

--------------------------------

### Run metrics example

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/metrics.md

Command to run the metrics example with the required features: testing, macros, memory, json, metrics.

```text
cargo run --example metrics --features testing,macros,memory,json,metrics
```

--------------------------------

### Generate AsyncAPI documentation using ruststream CLI

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/asyncapi/generation/cli-gen.md

Command to generate AsyncAPI documentation and write it to a file using the installed ruststream CLI.

```bash
ruststream asyncapi gen -o asyncapi.json
```

--------------------------------

### start(app)

Source: https://github.com/powersemmi/ruststream

Runs the app's real on_startup, connects each broker through its in-process transition, and spawns every subscription as a task on the test's runtime. Returns a TestApp instance or a TestError.

```APIDOC
## start(app)

### Description
Runs the app's real `on_startup`, connects each broker through its in-process transition, and spawns every subscription as a task on the test's runtime (`threads(n)` becomes `n` runtime workers, so `advance` reaches every timer).

### Method
`start`

### Signature
```rust
pub async fn start<A>(app: A) -> Result<Self, TestError>
where A: App<State = State>;
```

### Parameters
- **app** (A) - Required - The application to start.

### Returns
- `Result<Self, TestError>` - The test harness or an error.
```

--------------------------------

### TestApp::start

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/testing/starting-and-stopping/start.md

Runs the app main builds, unchanged, with every broker connected in process and no server. The harness starts the app in the service's own order: on_startup, then each broker's connect, then the subscriptions, then after_startup.

```APIDOC
## TestApp::start

### Description
Runs the app `main` builds, unchanged, with every broker connected in process and no server. The harness starts the app in the service's own order: `on_startup`, then each broker's connect, then the subscriptions, then `after_startup`.

### Method
start

### Endpoint
TestApp::start

### Parameters
None

### Response
#### Success Response
- **App** - The app instance, started with all brokers connected in process.

#### Error Response
- **TestError::NoTransport** - If a broker lacks an in-process mode, `start` fails with this error naming the broker.
- **TestError::Startup** - If `on_startup` fails, `start` fails with this error before any broker connects.

### Request Example
```rust
// Example from source not provided
```

### Response Example
```rust
// Example from source not provided
```
```

--------------------------------

### Project structure created by cargo generate

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/quickstart.md

Shows the layout of the scaffolded project. main.rs uses #[ruststream::app] to build the service and mount the router, orders.rs contains #[subscriber] handlers (one publishes a reply), and routes.rs collects handlers into a Router.

```text
my-service/
├── Cargo.toml
└── src/
    ├── main.rs      # #[ruststream::app] builds the service and mounts the router
    ├── orders.rs    # handlers as #[subscriber] functions (one publishes a reply)
    └── routes.rs    # collects the handlers into a Router
```

--------------------------------

### Run byte_lanes example without codec feature

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/bytes.md

Builds and tests the byte_lanes example with no codec feature, using testing, macros, and memory features.

```text
cargo test --example byte_lanes --no-default-features --features testing,macros,memory
```

--------------------------------

### Build command for protobuf example without codec feature

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/protobuf.md

Runs the protobuf example with no default features, enabling testing, macros, and memory features. The service builds without a codec feature because the type carries its own bytes via #[wire(prost)].

```text
cargo test --example protobuf --no-default-features --features testing,macros,memory
```

--------------------------------

### start_at

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/replay/start-at.md

Opens a replayable subscription at a specified position: the start of the log, a sequence number, or the tip. It seeks the subscription before its first delivery on every startup and is available as a mount-site setting or attribute clause.

```APIDOC
## start_at

### Description
Opens a replayable subscription at a position: the start of the log, a sequence number, or the tip. It seeks the subscription before its first delivery on every startup.

### Usage
- **Mount-site setting**: `start_at(position)`
- **Attribute clause**: `start_at(position)`

### Parameters
- **position** (MemoryPosition) - Required - The position to start the subscription from. Can be `MemoryPosition::start()` (oldest entry), `sequence(n)` (n-th entry, zero-based), or `end()` (tip).

### Behavior
- Without `start_at`, the subscription opens at the broker's default (in-memory broker defaults to the tip, so earlier entries are not delivered).
- Requires the broker to implement `Seekable` and, for the in-memory broker, to be built with `MemoryBroker::retaining` (does not compile with `MemoryBroker::new()`).

### Example
Opens the subscription at the start of the log and replays what was published before the service started.

```rust
// Example code from crates/ruststream/examples/seek.rs#start_at
```

```

--------------------------------

### Run tests for both forms (guide and manual_guide)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/testing.md

Commands to run the tests for both the macros and manual forms of the shop. The first uses the default features plus testing, memory, json, cbor, and asyncapi; the second disables default features and enables the same set.

```bash
cargo test --example guide --features testing,macros,memory,json,cbor,asyncapi
cargo test --example manual_guide --no-default-features --features testing,memory,json,cbor,asyncapi
```

--------------------------------

### Manual handler with state (quickstart shape)

Source: https://github.com/powersemmi/ruststream

A manual handler with state, no macros — examples/manual/quickstart.rs shape. Uses MemoryBroker and includes a subscriber for the 'orders' source.

```rust
use ruststream::memory::MemoryBroker;
use ruststream::prelude::*;
use serde::Deserialize;

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

struct Audit;

impl Handle<Order> for Audit {
    async fn handle(&self, order: &Order, _outs: &(), _ctx: &mut Context<'_>)
        -> Result<(), HandlerOutcome>
    {
        println!("order {}", order.id);
        Ok(())
    }
}

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

```

--------------------------------

### Current-thread runtime example

Source: https://github.com/powersemmi/ruststream

Shows what `#[ruststream::app(flavor = "current_thread")]` expands to. Uses `run_main_on` with `AppRuntime::CurrentThread` and a `MemoryBroker`.

```rust
use ruststream::memory::MemoryBroker;
use ruststream::runtime::cli::{AppRuntime, run_main_on};
use ruststream::runtime::{App, AppInfo, RustStream};

fn app() -> impl App {
    RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
}

fn main() -> std::process::ExitCode {
    run_main_on(AppRuntime::CurrentThread, app)
}
```

--------------------------------

### Run logging example with RUST_LOG filter

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/logging.md

Command to run the logging example with the console logger, setting the RUST_LOG filter to ruststream=debug,info and enabling the required features.

```text
RUST_LOG=ruststream=debug,info cargo run --example logging --features testing,macros,memory,json,logging -- run
```

--------------------------------

### Run and generate AsyncAPI with ruststream CLI

Source: https://github.com/powersemmi/ruststream/blob/main/README.md

Commands to start the service and print the AsyncAPI document. Use 'cargo run -- run' as an alternative to 'ruststream run'.

```bash
ruststream run                 # start the service (or: cargo run -- run)
ruststream asyncapi gen        # print the AsyncAPI document

```

--------------------------------

### OtelBuilder::init

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/otel-export.md

Builds the exporters and installs the tracer and meter providers as the process's globals. It also installs a tracing subscriber that bridges spans into the tracer, so spans from Propagation are exported without further wiring.

```APIDOC
## OtelBuilder::init

### Description
Builds the exporters and installs the tracer and meter providers as the process's globals. It also installs a `tracing` subscriber that bridges spans into the tracer, so the spans of Propagation are exported with no further wiring.

### Method
init

### Parameters
None explicitly documented.
```

--------------------------------

### start_live_within(app, deadline)

Source: https://github.com/powersemmi/ruststream

Starts the app live with a custom settle deadline. Returns a TestApp instance or a TestError.

```APIDOC
## start_live_within(app, deadline)

### Description
`start_live` with a custom deadline.

### Method
`start_live_within`

### Signature
```rust
pub async fn start_live_within<A>(app: A, deadline: Duration) -> Result<Self, TestError>
where A: App<State = State>;
```

### Parameters
- **app** (A) - Required - The application to start.
- **deadline** (Duration) - Required - The settle deadline.

### Returns
- `Result<Self, TestError>` - The test harness or an error.
```

--------------------------------

### Generate AsyncAPI document with cargo run -- asyncapi gen

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/quickstart.md

Prints the AsyncAPI document as JSON. Output flags (-o, --yaml) and document details are covered in the asyncapi module documentation.

```bash
cargo run -- asyncapi gen
```

--------------------------------

### Example: Policy without a broker, paired at startup

Source: https://github.com/powersemmi/ruststream

Shows a policy without a broker, paired at startup (error-handling form).

```APIDOC
## Example: Policy without a broker, paired at startup

### Description
Shows a policy without a broker, paired at startup (error-handling form).

### Code
```rust
use ruststream::memory::{MemoryBroker, MemoryPublish};
use ruststream::runtime::PublishExt;
use ruststream::{Broker, PublishPolicy};

let policy = MemoryPublish;                          // no broker in sight
let connected = MemoryBroker::new().connect().await?;
let publisher = policy.pair(&connected).await.map_err(|e| e.to_string())?;
publisher.message(&Order(vec![1, 2, 3])).to("orders").publish().await?;
```
```

--------------------------------

### main.rs: Build shop on NatsBroker (Manual API)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/networked-broker.md

Entry point using the manual API to build the shop on a NatsBroker. The broker connects only when the service starts. Requires the ruststream-nats crate and was tested against a NATS 2 server.

```rust
use std::process::ExitCode;

use ruststream::runtime::cli::run_main;
use ruststream_nats::prelude::*;

use crate::state::Shop;

fn app() -> impl App<State = Shop> {
    service::shop(NatsBroker::new("nats://localhost:4222"))
}

fn main() -> ExitCode {
    run_main(app)
}

```

--------------------------------

### SubscriberSettings::start_at

Source: https://github.com/powersemmi/ruststream

Sets the start position for the subscription. Wraps the source in StartAt<Src, P>.

```APIDOC
## SubscriberSettings::start_at

### Description
Sets the start position for the subscription. Wraps the source in StartAt<Src, P>.

### Method
Method

### Signature
```rust
fn start_at<P>(self, position: P) -> <Self::Settings as StartAtStep<P>>::Out
    where Self::Settings: StartAtStep<P>;
```

### Parameters
- `position` (P) - Required - The start position.

### Returns
- The next builder state after setting the start position.
```

--------------------------------

### Router module example with include_router

Source: https://github.com/powersemmi/ruststream

Shows a typical router module structure: a routes() function that builds a Router with includes and settings, and an app() function that creates a RustStream with a MemoryBroker and includes the router.

```rust
fn routes() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    Router::new()
        .include(accept)
        .include(reconcile)               // per-registration steps chain over the last include:
        .max_attempts(nonzero!(5u32))
        .dead_letter("payments.dead")
        .build()
        .include(audit)
}

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

--------------------------------

### Two-phase service with state, hooks, timeout, and broker

Source: https://github.com/powersemmi/ruststream

Standard RustStream app setup using the #[ruststream::app] attribute. Configures startup hook to connect to a PostgreSQL database, sets a 10-second shutdown timeout, adds a tracing layer, and includes a MemoryBroker with the 'accept' handler.

```rust
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| Database::connect("postgres://localhost/orders").await)
        .shutdown_timeout(Duration::from_secs(10))
        .layer(TracingLayer::default())
        .with_broker(MemoryBroker::new(), |b| { b.include(accept); })
}
```

--------------------------------

### Run NATS server and the service

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/networked-broker.md

Starts a NATS 2 server in Docker and runs the service. Core NATS keeps no messages; use JetStream for persistence.

```bash
docker run --rm -p 4222:4222 nats:2-alpine
cargo run -- run

```

--------------------------------

### Building AppInfo for a service

Source: https://github.com/powersemmi/ruststream

Example of constructing an `AppInfo` with contact, license, and tag, then wrapping it in `RustStream`. This consumes all the info model structs.

```rust
use ruststream::runtime::{App, AppInfo, RustStream};
use ruststream::{Contact, License, Tag};

fn app() -> impl App {
    let info = AppInfo::new("payments", "1.4.0")
        .description("everything the payment domain publishes")
        .contact(Contact::new().email("payments@example.com"))
        .license(License::new("Apache-2.0"))
        .tag(Tag::new("payments"));
    RustStream::new(info)
}
```

--------------------------------

### main.rs: Build shop on NatsBroker (Macros API)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/networked-broker.md

Entry point using the macros API to build the shop on a NatsBroker. The broker connects only when the service starts. Requires the ruststream-nats crate and was tested against a NATS 2 server.

```rust
use ruststream_nats::prelude::*;

use crate::state::Shop;

#[ruststream::app]
fn app() -> impl App<State = Shop> {
    service::shop(NatsBroker::new("nats://localhost:4222"))
}

```

--------------------------------

### run_until

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/running/until.md

Starts the service and runs until a provided future resolves, then shuts down gracefully. No signal handlers are installed, and the future starts after the service has started, allowing it to publish to the service and wait for the answer.

```APIDOC
## run_until

### Description
Starts the service and runs until a provided future resolves, then shuts down gracefully. No signal handlers are installed, and the future starts after the service has started, allowing it to publish to the service and wait for the answer.

### Method
run_until

### Endpoint
RustStream::run_until

### Parameters
- **future** (Future) - Required - The future that determines when the service should shut down.

### Response
Returns an error if a subscriber fails fast before the future resolves.

### Example
```rust
// Example from crates/ruststream/examples/app_running.rs#run_until
```
```

--------------------------------

### on_startup

Source: https://github.com/powersemmi/ruststream

Sets a hook that runs once before brokers connect; a failing hook aborts startup.

```APIDOC
## on_startup(hook)

### Description
Sets a hook that runs once before brokers connect; a failing hook aborts startup. State type `()`.

### Parameters
- **hook** (closure) - Required - The hook to run on startup.
```

--------------------------------

### Depend on ruststream-core without alloc

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/installation.md

For devices without heap, leaves alloc off to keep headers and payloads in fixed-size storage.

```toml
[dependencies]
ruststream-core = { version = "0.8", default-features = false }
```

--------------------------------

### RustStream::start

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/running/start.md

Brings the service up by connecting brokers, opening subscriptions, and running startup hooks. Returns a RunningApp that the caller owns, allowing the caller to decide when the service stops.

```APIDOC
## RustStream::start

### Description
Brings the service up and returns a RunningApp, so the caller owns the process and decides when it stops. Connects the brokers, opens the subscriptions, and runs the startup hooks before returning. A startup failure is the error start returns, before the caller accepts its own traffic. start installs no signal handlers; the caller stops the service with shutdown.

### Method
start

### Parameters
None documented.

### Returns
- **RunningApp** - The running application instance, which hands out the service's health and publishers for tasks beside it.

### Errors
- Startup failure - returned as the error from start before the caller accepts its own traffic.

### Notes
- The caller owns the process and decides when it stops.
- The caller stops the service with shutdown.
- The RunningApp also provides access to health and publishers for tasks beside it, as shown in the HTTP page.
```

--------------------------------

### Depend on ruststream-core with alloc feature

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/installation.md

For no_std broker crates with heap, enables alloc to keep headers and payloads on the heap.

```toml
[dependencies]
ruststream-core = { version = "0.8", default-features = false, features = ["alloc"] }
```

--------------------------------

### Disable default features and enable tokio runtime

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/installation.md

For broker crates needing only traits and runtime, disables default JSON codec and enables the tokio runtime feature.

```toml
[dependencies]
ruststream = { version = "0.8", default-features = false, features = ["tokio"] }
```

--------------------------------

### run

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/running/run.md

Starts the service and serves until SIGINT or SIGTERM, then shuts it down gracefully. It connects brokers, opens subscriptions, runs startup hooks, and waits for a signal.

```APIDOC
## run

### Description
Starts the service and serves until SIGINT or SIGTERM, then shuts it down gracefully. It connects the brokers, opens the subscriptions, runs the startup hooks, and waits for a signal. A startup failure is the error it returns. A signal starts the same shutdown a started service gets. A subscriber that fails fast tears the service down, and `run` returns that failure as its error. A subscription whose stream ends while the service runs does the same under `SubscriptionEnd::FailFast`.

### Method
run

### Endpoint
`run`

### Parameters
None

### Response
#### Success Response
No explicit response schema provided.

#### Error Response
- **startup failure** - Error returned if startup fails.
- **subscriber failure** - Error returned if a subscriber fails fast.
- **subscription stream end** - Error returned if a subscription stream ends while the service runs under `SubscriptionEnd::FailFast`.
```

--------------------------------

### Add ruststream and serde dependencies with features

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/installation.md

Adds the ruststream crate with macros, memory, and json features, plus serde for deriving message types. Requires edition 2024 and Rust 1.95 or newer.

```toml
[dependencies]
ruststream = { version = "0.8", features = ["macros", "memory", "json"] }
serde = { version = "1", features = ["derive"] }
```

--------------------------------

### Run the app with cargo run -- run

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Runs the application. The `logging` feature prints the service's log to the terminal: the startup lines, then a line for each order the handler logs with `tracing::info!`.

```bash
cargo run -- run
```

--------------------------------

### Manual RustStream app with froodi

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/state/froodi.md

Implements the same functionality as the macros example but manually: derives JsonSchema instead of Outgoing, implements OutgoingDestination and MessageHeaders by hand, and uses a Handle implementation with explicit FromContext resolution. The app is built with RustStream, MemoryBroker, and publishes replies.

```rust
use froodi::ruststream::ContainerLayer;
use froodi::{Container, DefaultScope, Inject, instance, registry};
use ruststream::memory::prelude::*;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

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

// `#[derive(Outgoing)]` with no `name`, by hand: the call site names the destination.
impl OutgoingDestination for Order {
    type Form = CallerName;
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
pub struct Receipt {
    pub order_id: u64,
    pub total: u64,
}

// `#[outgoing(name = "receipts")]` by hand: the type fixes its destination.
impl OutgoingDestination for Receipt {
    type Form = FixedName;

    const DESTINATION: &'static str = "receipts";
}

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

/// Lives as long as the application.
#[derive(Clone)]
pub struct Config {
    pub price_per_item: u64,
}

/// Built for every delivery from the configuration.
pub struct Pricing {
    per_item: u64,
}

impl Pricing {
    fn total(&self, order: &Order) -> u64 {
        order.items * self.per_item
    }
}

fn container(config: Config) -> Container {
    Container::new(registry! {
        provide(DefaultScope::App, instance(config)),
        scope(DefaultScope::Request) [
            provide(|Inject(config): Inject<Config>| Ok(Pricing {
                per_item: config.price_per_item,
            })),
        ],
    })
}

struct Price;

impl Handle<Order, Receipt> for Price {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> Result<Receipt, HandlerOutcome> {
        // What the attribute emits for `Inject(pricing): Inject<Pricing>`: a failed resolution
        // settles the delivery by its outcome and the body does not run.
        let Inject(pricing) = <Inject<Pricing> as FromContext>::from_context(ctx)
            .await
            .map_err(HandlerOutcome::from)?;
        Ok(Receipt {
            order_id: order.id,
            total: pricing.total(order),
        })
    }
}

pub fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(ContainerLayer::new(container(Config {
            price_per_item: 25,
        })))
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("orders", Price).reply().build())
                .out_reply(Publish)
        })
}

```

--------------------------------

### Define routes module (src/routes.rs) - Manual API

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Defines the routes module for the tutorial using the manual API. The `include` function adds a plain handler directly to the router, while handlers that publish replies use `.out_reply(..)` to name the reply's publish policy and `.build()` to finish registration. Without `.out_reply(..)`, `.build()` uses the broker's default publish policy.

```rust
```rust title="src/routes.rs" reference="crates/ruststream/examples/manual/tutorial/routes.rs#routes"
```
```

--------------------------------

### Define routes module (src/routes.rs) - Macros API

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Defines the routes module for the tutorial using the macros API. The `include` function adds a plain handler directly to the router, while handlers that publish replies use `.out_reply(..)` to name the reply's publish policy and `.build()` to finish registration. Without `.out_reply(..)`, `.build()` uses the broker's default publish policy.

```rust
```rust title="src/routes.rs" reference="crates/ruststream/examples/tutorial/routes.rs#routes"
```
```

--------------------------------

### Minimal service with #[subscriber] macro

Source: https://github.com/powersemmi/ruststream

Shows the minimal setup for a RustStream service using the #[subscriber] macro and MemoryBroker. Requires the memory prelude and serde Deserialize. Run with `cargo run -- run`; `cargo run -- asyncapi gen` prints the AsyncAPI document.

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

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    println!("got order {}", order.id);
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
    })
}
```

--------------------------------

### HealthProbe::subscriptions

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/running/health.md

Returns a SubscriptionCount indicating the number of subscriptions that consume and the number the service started. The count changes only when a subscription starts or stops.

```APIDOC
## HealthProbe::subscriptions

### Description
Returns a `SubscriptionCount` that reports the subscriptions that consume and the subscriptions the service started.

### Method
`subscriptions`

### Return Value
- `SubscriptionCount` - Contains the number of consuming subscriptions and the number of started subscriptions.

### Notes
- A subscription counts once, whatever its number of workers.
- The count changes only when a subscription starts or stops, so a delivery pays nothing for it.
```

--------------------------------

### start_live

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/testing/starting-and-stopping/start-live.md

Connects the brokers of the same app to a running stand, so one test body checks the service against the real broker. It is for a stand of a networked broker (Kafka or NATS) and uses the production startup path except for the broker connection.

```APIDOC
## start_live

### Description
Connects the brokers of the same app to a running stand, so one test body checks the service against the real broker. It is for a stand of a networked broker (Kafka or NATS) and uses the production startup path except for the broker connection.

### Method
start_live

### Endpoint
TestApp::start_live

### Parameters
None

### Response
#### Success Response
- No return value (unit)

#### Error Responses
- `TestError::NotSettled` - If the wait for subscriptions to handle published messages exceeds the timeout (default 10 seconds).
- `TestError::PausedClock` - If the test clock is paused when `start_live` is called.

### Notes
- Each broker connects through its ordinary `connect`.
- The test body is the same as for `TestApp::start`; only the start call differs.
- `MemoryBroker` connects without a server, so the example runs in process in every mode.
- A publish, `settle`, and `advance` wait on the real clock until every subscription has handled what was published to it.
- The wait lasts ten seconds at most; `start_live_within` sets another bound.
- The test's input reaches the broker through its connected form's default publish policy.

### Example
```rust
// See crates/ruststream/examples/testing_starting.rs#start-live
```

```

--------------------------------

### Create the orders-service crate with cargo new

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Initializes a new Rust crate named orders-service and changes into its directory. Run these commands in a terminal before adding dependencies.

```bash
cargo new orders-service
cd orders-service
```

--------------------------------

### MemoryBroker::new

Source: https://github.com/powersemmi/ruststream

Creates a memory broker with default settings.

```APIDOC
## MemoryBroker::new()

### Description
Creates a memory broker with default settings: discarding log (`Log = Discarding`), `Routing::EveryMatch`, no retention.
```

--------------------------------

### Service startup log output

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/build-a-service/service.md

Expected terminal output when running the service, showing lifecycle events: service start, broker connection, subscriber start, and service running. The service runs until interrupted with Ctrl+C.

```text
INFO ruststream::lifecycle starting service service=shop version=0.1.0 brokers=1 subscribers=1
INFO ruststream::lifecycle broker connected broker="ruststream::memory::MemoryBroker"
INFO ruststream::dispatch subscriber started subscriber=orders.placed input="guide_first::Order"
INFO ruststream::lifecycle service running subscribers=1
```

--------------------------------

### logging::Logging::new

Source: https://github.com/powersemmi/ruststream

Creates a logging integration with default settings.

```APIDOC
## logging::Logging::new()

### Description
Creates a logging integration with default filter `"info"`, ANSI auto-detected (`None`), target printed (`true`). `with_default_filter`, `with_ansi`, `with_target` override.
```

--------------------------------

### Example: Custom MessagePack Codec

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/codecs/custom-codec.md

A complete example of a custom codec that writes MessagePack with field names (struct as a map) using the `rmp-serde` crate. It demonstrates implementing the Codec trait and mounting the codec at the scope and reply position.

```APIDOC
## Example: Custom MessagePack Codec

### Description
This example implements a custom codec that writes MessagePack with field names, so a struct travels as a map (the form a peer in a dynamic language reads), unlike the built-in MessagePack codec which writes an array. It uses `rmp-serde`, the crate behind the `msgpack` feature.

### Build Command
```text
cargo test --example custom_codec --no-default-features --features testing,macros,memory,msgpack
```

### Codec Implementation
```rust reference="crates/ruststream/examples/custom_codec.rs#codec"
```

### Mounting
Mount the codec at the scope and at the reply position, like a built-in one.

#### Macros
```rust reference="crates/ruststream/examples/custom_codec.rs#mount"
```

#### Manual
```rust reference="crates/ruststream/examples/manual/custom_codec.rs#mount"
```
```

--------------------------------

### Run the service with ruststream run

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/cli.md

Runs the service via cargo run -- run. Ctrl+C stops it gracefully. The output shows the service starting, connecting the broker, starting subscribers, and stopping on shutdown.

```bash
ruststream run
```

```text
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.02s
     Running `target/debug/my-service run`
2026-10-16T07:49:57.254761Z  INFO ruststream::lifecycle starting service service=my-service version=0.1.0 brokers=1 subscribers=2
2026-10-16T07:49:57.254790Z  INFO ruststream::lifecycle broker connected broker="ruststream::memory::MemoryBroker"
2026-10-16T07:49:57.254844Z  INFO ruststream::dispatch subscriber started subscriber=orders input="my_service::orders::Order"
2026-10-16T07:49:57.254854Z  INFO ruststream::dispatch subscriber started subscriber=cancellations input="my_service::orders::Order"
2026-10-16T07:49:57.254859Z  INFO ruststream::lifecycle service running subscribers=2
2026-10-16T07:50:00.281593Z  INFO ruststream::lifecycle shutdown signal received
2026-10-16T07:50:00.281755Z  INFO ruststream::lifecycle service stopped
```

--------------------------------

### Wire the app manually (src/main.rs)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Shown in the Manual tab of the 'Wire it into an app' section. This is the manual (non-macro) version of the app wiring. The `include` decode uses the default codec, which is `json` when the feature is enabled, otherwise `cbor`, otherwise `msgpack`.

```rust
```rust title="src/main.rs" reference="crates/ruststream/examples/manual/tutorial/first_app.rs#app"
```
```

--------------------------------

### run_main

Source: https://github.com/powersemmi/ruststream

Parses process arguments, builds the Tokio runtime, and dispatches `run` or `asyncapi gen`. Returns an exit code: 0 on success, 1 on error.

```APIDOC
## [Rust] run_main

### Description
Parses process arguments, builds the Tokio runtime, and dispatches `run` or `asyncapi gen`. Returns an exit code: 0 on success, 1 on error.

### Signature
```rust
pub fn run_main<A, F>(build: F) -> ExitCode
where A: App, F: FnOnce() -> A;
```

### Parameters
- `build` (closure) - Required - Builder function that returns an `App` instance

### Returns
- `ExitCode` - `0` on success, `1` on error

### Example
```rust
fn main() -> std::process::ExitCode {
    match ruststream::runtime::cli::run_main(app) {
        code if code == std::process::ExitCode::SUCCESS => code,
        code => { std::process::exit(code.code().unwrap_or(1)); }
    }
}
```

```

--------------------------------

### Wire the app with macros (src/main.rs)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/guides/getting-started/tutorial.md

Shown in the Macros tab of the 'Wire it into an app' section. This is the macro-based version of the app wiring. The `include` macro decodes with the default codec, which is `json` when the feature is enabled, otherwise `cbor`, otherwise `msgpack`.

```rust
```rust title="src/main.rs" reference="crates/ruststream/examples/tutorial/first_app.rs#app"
```
```

--------------------------------

### message(&value)

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/subscribers/publishing/publish-builder.md

Starts a publish with a message value. Returns a PublishBuilder.

```APIDOC
## message(&value)

### Description
Starts a publish with a message value. Returns a PublishBuilder.

### Method
message

### Parameters
- **value** - The message value to publish.

### Returns
PublishBuilder
```

--------------------------------

### MemoryPosition enum and MemorySeeker API

Source: https://github.com/powersemmi/ruststream

MemoryPosition specifies where to start reading a name: Start (oldest retained), Sequence(n) (zero-based n-th message), or End (tip, resume with next publish). MemorySeeker is Clone, per-subscription-instance scope, and returns MemoryError::ShutDown after shutdown. Includes per-delivery context keys and the PARTITION_KEY_HEADER constant.

```rust
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] #[must_use]
pub enum MemoryPosition {
    Start,                  // oldest retained message of the name
    Sequence(usize),        // the zero-based n-th message published to the name
    End,                    // the tip: resume with the next publish
}
impl MemoryPosition {
    pub const fn start() -> Self;
    pub const fn sequence(n: usize) -> Self;
    pub const fn end() -> Self;
}
pub struct MemorySeeker;    // Clone; per-subscription-instance scope; MemoryError::ShutDown after shutdown

// per-delivery context keys (unit structs) + context structs:
pub struct Position;        // reads the delivery's position (Field/ContextField key)
pub struct SeekHandle;      // reads the subscription's seeker
pub struct MemoryContext { ... }       // per-delivery context (Position + SeekHandle keys)
pub struct MemoryBatchContext { ... }  // subscription-scoped (SeekHandle only)
pub const PARTITION_KEY_HEADER: &str = "partition-key";
```

--------------------------------

### Add RustStream plugin in Claude Code

Source: https://github.com/powersemmi/ruststream/blob/main/README.md

Commands to add and install the RustStream plugin in Claude Code.

```text
/plugin marketplace add powersemmi/ruststream
/plugin install ruststream@ruststream
```

--------------------------------

### Logging configuration and initialization (src/logging.rs)

Source: https://github.com/powersemmi/ruststream

Configures and initializes the logging subscriber. Reads the filter from RUST_LOG, falling back to the configured default filter (default 'info'). Writes to stderr with colored levels. The generated CLI 'run' command installs it automatically. Errors include Filter when the fallback directive does not parse and AlreadyInitialized when a global subscriber is already installed.

```rust
pub struct Logging { default_filter: String, ansi: Option<bool>, target: bool }
impl Default for Logging { /* default_filter: "info", ansi: None, target: true */ }
impl Logging {
    pub fn new() -> Self;
    pub fn with_default_filter(mut self, filter: impl Into<String>) -> Self;   // EnvFilter syntax
    pub const fn with_ansi(mut self, ansi: bool) -> Self;   // force colors on/off
    pub const fn with_target(mut self, target: bool) -> Self;   // print event targets, default true
    pub fn try_init(self) -> Result<(), LoggingInitError>;
}
pub fn init() -> Result<(), LoggingInitError>;   // Logging::new().try_init()

#[derive(Debug, Error)] #[non_exhaustive]
pub enum LoggingInitError {
    // Filter: the fallback directive does not parse
    // AlreadyInitialized: a global subscriber is already installed (never replaced)
}
```

--------------------------------

### OtelBuilder::attach

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/observability/otel-export.md

Takes providers built by the caller and installs no globals. This is useful for using your own exporter or an in-memory one in a test.

```APIDOC
## OtelBuilder::attach

### Description
Takes providers built by the caller and installs no globals. This is useful for using your own exporter or an in-memory one in a test.

### Method
attach

### Parameters
- **providers** (object) - Required - The tracer and meter providers built by the caller.
```

--------------------------------

### DescribeServer trait

Source: https://github.com/powersemmi/ruststream

Trait for describing a server, providing a method to get the server specification.

```APIDOC
## DescribeServer trait

### Description
Trait for describing a server, providing a method to get the server specification.

### Method
`fn describe_server(&self) -> ServerSpec`

### Signature
```rust
pub trait DescribeServer: Broker {
    fn describe_server(&self) -> ServerSpec;
}
```

### Returns
- **ServerSpec** - The server specification.
```

--------------------------------

### Cargo.toml dependencies for macros API

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/application/state/froodi.md

Declares froodi with the ruststream feature, ruststream with macros, memory, and json features, and serde with derive. Required for the macros-based example.

```toml
[dependencies]
froodi = { version = "1.0.0-beta.19", features = ["ruststream"] }
ruststream = { version = "0.8", features = ["macros", "memory", "json"] }
serde = { version = "1", features = ["derive"] }

```

--------------------------------

### Dependency line for MemoryBroker

Source: https://github.com/powersemmi/ruststream/blob/main/docs/components/memory/index.md

Add the `memory` feature to the `ruststream` dependency to use the in-process MemoryBroker. The example also enables `macros` and `json` features.

```toml
ruststream = { version = "0.8", features = ["macros", "memory", "json"] }
```
Tokens:
Raw