HTTP frameworks

RustStream is not an HTTP framework and does not try to become one. When a service exposes a synchronous HTTP API and consumes messages, the HTTP framework (axum, actix-web, or any other tokio-based stack) runs beside the RustStream app in the same process and runtime. This page shows the wiring on axum and the pattern that makes the combination reliable: a transactional outbox.

The full compiled example lives at examples/http_outbox.rs:

cargo run --example http_outbox --features macros,memory,json

Running beside an HTTP server

Both sides come up in main. start() brings the messaging side up in the background - the state producer, broker connects, subscription opens - and resolves once the service is running, so a startup failure surfaces before the HTTP side accepts traffic. The returned RunningApp handle coordinates the two lifetimes: stopping() is an owned future that resolves if the messaging side tears itself down (a fail-fast failure), which plugs straight into axum's with_graceful_shutdown so the process does not keep serving HTTP with a dead consumer; and shutdown() is the explicit graceful teardown - the on_shutdown hooks, a drain of in-flight handlers bounded by the shutdown timeout, broker shutdown - once the HTTP server has stopped. The publisher arrives through a bound token: .bindable() wraps the broker and bind(..) mints the token before the app consumes it, then running.publisher(token) pairs the token once start() has connected the broker - so the sibling task gets a live publisher, never a "not connected" state, and the live form is a plain value, safe to clone into whatever state the HTTP framework carries:

examples/http_outbox.rs

let broker = MemoryBroker::new().bindable();
// A token, not a publisher: minted before registration, paired after start.
let egress = broker.bind(MemoryPublish);
let app = RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
    b.include(fulfil);
});

// The messaging side starts in the background; a startup failure (a broker refusing to
// connect, a subscription failing to open) surfaces here, before HTTP accepts traffic.
let running = app.start().await?;
// The handle existing witnesses that startup connected the broker, so the relay task gets a
// live publisher, never a "not connected" state.
let egress = running.publisher(egress).await?;

let store = Arc::new(Mutex::new(Store::default()));
tokio::spawn(relay_outbox(store.clone(), egress));

let router = Router::new()
    .route("/orders", post(place_order))
    .with_state(store)
    // The health probe is `Clone` and outlives the app handle; /healthz flips to 503 the
    // moment the messaging side fail-fasts, answering through the graceful drain below.
    .route("/healthz", get(healthz).with_state(running.health()));

let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
println!("orders API on http://127.0.0.1:8080/orders");
// The host owns the signals. HTTP stops on Ctrl+C, or when the messaging side tears itself
// down (fail-fast) so the process does not keep serving with a dead consumer; either way the
// messaging side then drains gracefully.
let stopping = running.stopping();
axum::serve(listener, router)
    .with_graceful_shutdown(async move {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            () = stopping => {}
        }
    })
    .await?;
running.shutdown().await?;

A healthz endpoint

start() is the readiness gate; the health probe covers everything after it. RunningApp::health() hands out a cheap, cloneable HealthProbe backed by a watch channel: state() is a lock-free snapshot (Running, ShuttingDown, Stopped, or Failed { reason } carrying the fail-fast diagnostic), and the probe outlives shutdown(), so the route keeps answering with the terminal state. This closes the gap stopping() alone leaves: when the messaging side fail-fasts but a sibling task keeps the process alive, /healthz flips to 503 instead of serving a permanent 200 for a dead consumer:

examples/http_outbox.rs

/// The liveness endpoint: the probe outlives `shutdown`, so this route reports the terminal
/// state for as long as requests still reach it. Here that window is the graceful drain: a
/// fail-fast also resolves the `stopping()` future below and stops the HTTP server. A host
/// that should keep serving after the messaging side dies would omit that select arm.
async fn healthz(State(health): State<HealthProbe>) -> (StatusCode, String) {
    match health.state() {
        HealthState::Running => (StatusCode::OK, "running".to_owned()),
        state => (StatusCode::SERVICE_UNAVAILABLE, format!("{state:?}")),
    }
}

The route carries its own state (get(healthz).with_state(running.health())), so it composes with whatever state the rest of the router holds - the full wiring above registers it beside /orders.

The subscriber side is an ordinary handler; the same service consumes what its HTTP endpoints produce, and any other service subscribed to the broker sees the events too:

examples/http_outbox.rs

/// The same service consumes what its HTTP endpoints produce; any other service subscribed to
/// the broker would see the event too.
#[subscriber("orders.placed")]
async fn fulfil(order: &OrderPlaced) -> HandlerResult {
    println!("fulfilling order {} ({})", order.id, order.item);
    HandlerResult::Ack
}

Publishing straight from a request

The simplest integration puts the publisher into the HTTP framework's state and publishes on the request path, exactly like publishing from inside a handler: encode with the codec, build an OutgoingMessage, and await the publish. The metrics guide's complete server does this to drive its counters.

The trade-off is coupling: a broker outage now fails or stalls HTTP requests, and a crash after the database write but before the publish loses the event (or publishes an event for a write that rolled back, in the opposite order). If the endpoint also writes to a database, that gap is a consistency bug waiting for a deploy window. The fix is the transactional outbox.

Transactional outbox

Instead of publishing on the request path, the endpoint records the event next to the business write, atomically. A relay then moves recorded events to the broker:

examples/http_outbox.rs

/// The integration event the HTTP side hands to the messaging side.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderPlaced {
    id: u64,
    item: String,
}

examples/http_outbox.rs

/// The business records and the pending outbox rows, committed together. With a real database
/// both inserts run in one SQL transaction; the mutex over the pair plays that role here, so an
/// order is never stored without its event, and no event exists without its order.
#[derive(Default)]
struct Store {
    orders: Vec<OrderPlaced>,
    outbox: VecDeque<OrderPlaced>,
}

The endpoint only writes to the store. Recording the order and queueing its event is one atomic step, and no broker I/O can fail or stall the HTTP response:

examples/http_outbox.rs

/// The request path only writes to the store: recording the order and queueing its event is one
/// atomic step, and no broker I/O can fail or stall the HTTP response.
async fn place_order(
    State(store): State<Arc<Mutex<Store>>>,
    Json(order): Json<OrderPlaced>,
) -> &'static str {
    let mut store = store.lock().await;
    store.orders.push(order.clone());
    store.outbox.push_back(order);
    "accepted\n"
}

A background task drains the outbox into the broker. A row is removed only after its publish succeeds, so a broker outage delays events instead of losing them; a crash between the publish and the removal re-publishes the row on restart. Consumers therefore see at-least-once delivery, the usual contract of an outbox, and handle duplicates the same way they handle redeliveries from the broker itself:

examples/http_outbox.rs

/// Drains the outbox into the broker. A row is removed only after its publish succeeds, so a
/// broker outage delays events instead of losing them; a crash between publish and removal
/// re-publishes the row on restart, which is the usual at-least-once contract of an outbox.
async fn relay_outbox(store: Arc<Mutex<Store>>, egress: MemoryPublisher) {
    let mut tick = tokio::time::interval(Duration::from_millis(200));
    loop {
        tick.tick().await;
        loop {
            let Some(event) = store.lock().await.outbox.front().cloned() else {
                break;
            };
            let payload = JsonCodec.encode(&event).expect("serializable");
            let out = OutgoingMessage::new("orders.placed", payload.as_ref());
            if egress.publish(out).await.is_err() {
                // Keep the row and retry on the next tick.
                break;
            }
            store.lock().await.outbox.pop_front();
        }
    }
}

With a real database the Store is a table plus an outbox table written in one SQL transaction, and the relay reads outbox rows in insertion order, publishes, and deletes them. Everything else stays as shown: the broker, the publisher, and the subscriber do not know the outbox exists.

Try it

curl -X POST http://127.0.0.1:8080/orders \
  -H 'content-type: application/json' -d '{"id":1,"item":"book"}'

The response returns as soon as the store commits; the fulfil handler logs the order a moment later, when the relay has published the event.