AsyncAPI

With the asyncapi feature, RustStream generates an AsyncAPI 3.0 document from the application's handlers: each subscriber becomes a channel and a receive operation, and payload types contribute schemas.

ruststream = { version = "0.6", features = ["macros", "memory", "asyncapi"] }

Generating the document

The quickest path is the CLI, which runs your service's generator and prints the document:

ruststream asyncapi gen                  # JSON to stdout
ruststream asyncapi gen -o asyncapi.json
ruststream asyncapi gen --yaml

In code, build the spec from the application with build_spec, then serialize it with to_json or to_yaml:

examples/asyncapi_http.rs

/// Builds the AsyncAPI document and the viewer HTML from the service.
fn document() -> Result<(String, String), serde_json::Error> {
    let spec = build_spec(&service()).to_json()?;
    let viewer = render_viewer_html("/asyncapi.json", &ViewerOptions::default());
    Ok((spec, viewer))
}

#[ruststream::app] wires the asyncapi gen command to build_spec for you, so the CLI and a hand-written call produce the same document.

Payload schemas

A handler's payload type appears as a schema when it derives JsonSchema. RustStream re-exports schemars, so you do not need a direct dependency:

examples/asyncapi_http.rs

/// An order placed by a customer.
#[derive(Debug, Deserialize, Message, JsonSchema)]
struct Order {
    id: u64,
    item: String,
}

A type without JsonSchema still works as a handler payload; it just contributes no schema to the document.

Message names and descriptions

A documented payload type feeds the message component on its own: with the JsonSchema derive, the type's doc comment becomes the message description, and a #[schemars(title = "...")] (or rename) names the component. Without a schema, the component is named after the payload type and the description falls back to the handler's doc comment (which also documents the receive operation).

To control the metadata explicitly - including for types without JsonSchema - implement the Message trait, which takes precedence over the schema; or derive it, which uses the type's name and doc comment:

use ruststream::Message;

/// An order placed by a customer.
#[derive(Message, serde::Deserialize)]
struct Order {
    id: u64,
}
// In the document: components.messages.Order with that description.

A manual impl Message can name the component differently from the Rust type (const NAME: &'static str = "CustomOrder";), which keeps the wire contract stable across renames.

Servers

Record the servers your service connects to so they appear in the document's servers section. Build a ServerSpec directly:

examples/asyncapi_http.rs

fn service() -> RustStream {
    // `with_broker_labeled` records the broker under a label that is both its stable identity and
    // its AsyncAPI server name, deriving the server entry from the broker's own `DescribeServer`
    // spec - here the in-memory broker, which describes itself as an in-process "memory" server
    // with no host. A broker without a `DescribeServer` impl is instead declared explicitly with
    // `.server(name, spec)` alongside a plain `with_broker`.
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // A described external server. Security is the author's statement, not the broker's:
        // the same broker is deployed publicly and internally with different authentication,
        // so the scheme is attached to the spec at registration and brokers never set it.
        .server(
            "kafka",
            ServerSpec::new("kafka.example.com:9093", "kafka")
                .with_security(SecurityScheme::scram_sha512().with_description("SASL over TLS")),
        )
        .with_broker_labeled("in-process", MemoryBroker::new(), |b| b.include(handle))
}

A broker crate may also implement the DescribeServer capability, in which case broker.describe_server() produces the spec for you (the shipped brokers all do), and with_broker_labeled records it automatically under the broker's label.

Server security

Declare how clients authenticate with ServerSpec::with_security; each scheme lands in components.securitySchemes and is referenced from the server's security list:

examples/asyncapi_http.rs

// A described external server. Security is the author's statement, not the broker's:
// the same broker is deployed publicly and internally with different authentication,
// so the scheme is attached to the spec at registration and brokers never set it.
.server(
    "kafka",
    ServerSpec::new("kafka.example.com:9093", "kafka")
        .with_security(SecurityScheme::scram_sha512().with_description("SASL over TLS")),
)

SecurityScheme has constructors for the AsyncAPI 3.0 scheme kinds - user_password, plain, scram_sha256 / scram_sha512, gssapi, api_key, x509, http, http_api_key, open_id_connect, and oauth2 (which takes the flows object as raw JSON) - plus SecurityScheme::custom(json) as the escape hatch for anything they do not model. Without with_security the document carries no security sections at all.

Security is deliberately the service author's statement, not the broker's: the same broker is deployed publicly and internally with different authentication, so DescribeServer never reports it. To secure a server a broker registered automatically (with_broker_labeled), declare it explicitly instead: .server(label, broker.describe_server().with_security(..)) with the same label.

Serving the document

Hosting is intentionally not part of the framework. build_spec and to_json / to_yaml give you the bytes; you mount them in whatever HTTP stack you already run (axum, actix, or any other).

For an interactive viewer, render_viewer_html returns a self-contained HTML page that loads the AsyncAPI React component and points it at your spec URL:

use ruststream::asyncapi::{render_viewer_html, ViewerOptions};

let html = render_viewer_html("/asyncapi.json", &ViewerOptions::default());

Serve that HTML and the spec JSON from two routes in your own server. By default the viewer loads its assets from a CDN; override the base URL with ViewerOptions::with_cdn_base for offline or locked-down deployments (with_title sets the page title).

A complete server

The asyncapi_http example serves the document and the viewer with axum. Run it with cargo run --example asyncapi_http --features macros,memory,asyncapi, then open http://127.0.0.1:8080/.

examples/asyncapi_http.rs

//! Serve a service's AsyncAPI document and an interactive viewer over HTTP with axum.
//!
//! ```text
//! cargo run --example asyncapi_http --features macros,memory,asyncapi
//! ```
//!
//! Then open <http://127.0.0.1:8080/> for the viewer, or fetch the raw document:
//!
//! ```text
//! curl http://127.0.0.1:8080/asyncapi.json
//! ```

use axum::Router;
use axum::http::header::CONTENT_TYPE;
use axum::response::{Html, IntoResponse};
use axum::routing::get;
use ruststream::asyncapi::{ViewerOptions, build_spec, render_viewer_html};
use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerResult, RustStream};
use ruststream::schemars::JsonSchema;
use ruststream::{Message, SecurityScheme, ServerSpec, subscriber};
use serde::Deserialize;

/// An order placed by a customer.
#[derive(Debug, Deserialize, Message, JsonSchema)]
struct Order {
    id: u64,
    item: String,
}

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

fn service() -> RustStream {
    // `with_broker_labeled` records the broker under a label that is both its stable identity and
    // its AsyncAPI server name, deriving the server entry from the broker's own `DescribeServer`
    // spec - here the in-memory broker, which describes itself as an in-process "memory" server
    // with no host. A broker without a `DescribeServer` impl is instead declared explicitly with
    // `.server(name, spec)` alongside a plain `with_broker`.
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // A described external server. Security is the author's statement, not the broker's:
        // the same broker is deployed publicly and internally with different authentication,
        // so the scheme is attached to the spec at registration and brokers never set it.
        .server(
            "kafka",
            ServerSpec::new("kafka.example.com:9093", "kafka")
                .with_security(SecurityScheme::scram_sha512().with_description("SASL over TLS")),
        )
        .with_broker_labeled("in-process", MemoryBroker::new(), |b| b.include(handle))
}

/// Builds the AsyncAPI document and the viewer HTML from the service.
fn document() -> Result<(String, String), serde_json::Error> {
    let spec = build_spec(&service()).to_json()?;
    let viewer = render_viewer_html("/asyncapi.json", &ViewerOptions::default());
    Ok((spec, viewer))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (spec, viewer) = document()?;

    let router = Router::new()
        .route(
            "/",
            get(move || {
                let viewer = viewer.clone();
                async move { Html(viewer) }
            }),
        )
        .route(
            "/asyncapi.json",
            get(move || {
                let spec = spec.clone();
                async move { ([(CONTENT_TYPE, "application/json")], spec).into_response() }
            }),
        );

    let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
    println!("AsyncAPI viewer on http://127.0.0.1:8080/");
    axum::serve(listener, router).await?;
    Ok(())
}