RustStream

RustStream subscribes a Rust service to event streams and publishes messages to them. The service is not bound to one message broker. The core is traits and a router runtime. Codecs, AsyncAPI generation, Prometheus metrics, and a conformance harness for broker authors ship with it.

Two architectural commitments shape the framework:

1. A real interface for third-party brokers. The core holds only traits and types, with zero broker dependencies. Each broker is an independent crate. The conformance harness checks the contract.
2. Broker-specific config stays in broker crates. The core carries no broker-specific config or defaults. Each broker crate owns its own Config. An upstream change affects only that crate, not the framework.

MacrosManual

examples/quickstart.rs

//! The landing-page example: a one-handler service with no runtime boilerplate.
//!
//! ```text
//! cargo run --example quickstart --features macros,memory,json -- run
//! ```

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

#[derive(Debug, 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);
    })
}

examples/manual/quickstart.rs

//! The landing-page example written without the `macros` feature: the handler is a named type
//! with an `impl Handle`, mounted with the `subscriber` constructor, and `main` is hand-written.
//!
//! ```text
//! cargo run --example manual_quickstart --no-default-features --features memory,json
//! ```

use std::error::Error;
use std::future::{Future, ready};

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

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

/// The handler: `#[subscriber("orders")]` generates this type and this impl. Every axis of the
/// form - the reply, the injections, the broker context, the application state - is a defaulted
/// parameter of `Handle`, so a plain body names none of them.
struct Receive;

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

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

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

#[ruststream::app] generates main with all the runtime boilerplate. cargo run -- run starts the service, and cargo run -- asyncapi gen prints its AsyncAPI document.

Design principles

• Fully async, tokio-based. The public API has no blocking calls.
• Generic core, no dyn in the contract. The contract is built on associated types and native async fn in trait. The runtime performs type erasure where a service needs it.
• Subscribers are Streams, not callbacks. The Stream itself provides back-pressure. The runtime builds callbacks on top of it.
• Ack consumes self. A second ack is a compile error.
• Capability traits for optional features. BatchSubscriber, TransactionalPublisher, RequestReply, Partitioned, and Seekable are not part of the mandatory interface.

Where to go next

• Installation - features and crate setup.
• Quick start - scaffold a service with cargo generate.
• Tutorial - build a service step by step.
• Testing - test handlers in-process, no server needed.
• HTTP frameworks - run beside axum with a transactional outbox.
• Brokers - the in-memory broker and the broker crates.
• Broker authors - implement the contract and pass conformance.

Scope of this repository

This site documents ruststream, the broker-agnostic core crate. Concrete brokers (NATS, Kafka, RabbitMQ, Redis, MQTT, and more) ship as separate crates. Each of them depends on ruststream from crates.io.

The Rust API reference is published on docs.rs - see API reference.