HTTP

An HTTP server owns the process and the service runs beside it: started before the server, stopped after it.

The service starts with start before the server accepts traffic, so a startup failure stops the process early. publisher pairs a bound token with its connected broker, for a task beside the service that publishes. The example's relay task publishes the orders an HTTP request stored in a transactional outbox. The server stops on Ctrl+C, or when stopping reports a fail-fast failure of the service. The service then drains with shutdown.

Example

Macros
Manual

examples/http_outbox.rs
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let broker = MemoryBroker::new().bindable();
    // A token, not a publisher: minted before registration, paired after start.
    let egress = broker.bind(Publish);
    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?;
    tracing::info!("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?;
    Ok(())
}

examples/manual/http_outbox.rs
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let broker = MemoryBroker::new().bindable();
    // A token, not a publisher: minted before registration, paired after start.
    let egress = broker.bind(Publish);
    let app = RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // The attribute reads the description off the handler's doc comment; on the value path
        // `describe` states it.
        b.include(
            subscriber(OrderPlaced::DESTINATION, Fulfil)
                .describe(
                    "The same service consumes what its HTTP endpoints produce; any other \
                     service subscribed to the broker would see the event too.",
                )
                .build(),
        );
    });

    // 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?;
    tracing::info!("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?;
    Ok(())
}

Example​

Example