跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

在 HTTP 服务器旁运行，outbox

店面通过 HTTP 接收订单，把订单连同事件一起存储；中继把事件交给 Broker，商店运行在服务器旁边。

服务同时提供 HTTP 时，进程归 HTTP 服务器所有，商店运行在它旁边。 start 在后台启动商店，商店运行起来后才返回。因此，如果 Broker 拒绝连接，进程会在服务器接收请求之前停止。 HTTP 请求不应自己向 Broker 发布：Broker 故障会让请求失败，而存储了订单却没有事件，订单就永远到不了商店。 事务性 outbox 同时解决这两个问题。 请求在一个事务中写入订单和它的事件，中继稍后发布已存储的事件。 Broker 故障只会推迟事件，不会丢失任何事件。如果在发布和删除之间发生崩溃，事件会再发布一次，所以每个订单商店至少会看到一次。

存储​

使用真实数据库时，订单和 outbox 是在一个事务中写入的两张表。 这里由一个保护这两者的互斥锁承担这个角色。 只有真实数据库能让事件在崩溃后保留下来：这里的行存放在内存中，一旦崩溃，尚未发布的行就会丢失。

http.rs
/// The orders the storefront took, and the events the shop still owes the broker. With a real
/// database both writes run in one transaction; the mutex over the pair plays that role here.
#[derive(Default)]
struct Store {
    orders: Vec<Order>,
    outbox: VecDeque<(RequestMeta, Order)>,
}

HTTP 部分和 outbox 都是普通函数，所以两个标签页中的代码相同。

端点​

POST /orders 以 JSON 接收订单，从 x-request-id 消息头读取请求 id。 它只写入存储，所以不等待 Broker 就以 202 Accepted 应答。 超出商店上限的订单以 400 Bad Request 应答，不会被存储。

http.rs
/// Takes an order over HTTP. The request only writes to the store, so no broker I/O can fail or
/// stall the response.
async fn place_order(
    State(store): State<Arc<Mutex<Store>>>,
    headers: HeaderMap,
    Json(order): Json<Order>,
) -> (StatusCode, &'static str) {
    let request_id = headers
        .get("x-request-id")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("unknown")
        .to_owned();
    if total_cents(&order.lines).is_none() {
        return (
            StatusCode::BAD_REQUEST,
            "the order is past the shop's bounds\n",
        );
    }
    {
        let mut store = store.lock().await;
        store.orders.push(order.clone());
        store.outbox.push_back((RequestMeta { request_id }, order));
    }
    (StatusCode::ACCEPTED, "accepted\n")
}

中继​

中继先发布最早的事件，并带上它的消息头契约，发布成功之后才删除它。 发布被拒绝时，这一行留到下一轮处理。 中继一直运行到进程让它停止，然后再执行最后一轮。 发布者可以是任意 Publisher，所以更换 Broker 时中继不变。

http.rs
/// Hands the outbox to the broker, oldest event first. A row leaves the outbox only after its
/// publish succeeded, so a broker outage delays events and loses none.
async fn relay_pending(store: &Mutex<Store>, egress: &impl Publisher) {
    loop {
        let Some((meta, order)) = store.lock().await.outbox.front().cloned() else {
            return;
        };
        let sent = egress
            .message(&order)
            .with_headers(&meta)
            .with_codec(JsonCodec)
            .publish()
            .await;
        if sent.is_err() {
            // The row stays for the next round.
            return;
        }
        store.lock().await.outbox.pop_front();
    }
}

/// Runs the relay every 200 milliseconds until `stop` completes, then runs it once more.
async fn relay_outbox(
    store: Arc<Mutex<Store>>,
    egress: impl Publisher,
    stop: impl Future<Output = ()>,
) {
    let mut rounds = tokio::time::interval(Duration::from_millis(200));
    tokio::pin!(stop);
    loop {
        tokio::select! {
            biased;
            () = &mut stop => break,
            _ = rounds.tick() => relay_pending(&store, &egress).await,
        }
    }
    // The last round runs once no request writes a row any more, so it empties the outbox.
    relay_pending(&store, &egress).await;
}

进程​

bindable 让进程在商店启动之前拿到一个发布者令牌，Broker 连接之后，publisher 把令牌变成可用的发布者。 服务器在 Ctrl+C 时停止，或者在 stopping 报告商店自行停止时停止。 HTTP 处理完所有请求之后，进程停止中继，中继的最后一轮在商店仍在运行时交出剩余的行。 然后商店用 shutdown 处理完剩余的工作。 MemoryBroker 也把队列存放在进程内存中：最后一轮交出、但商店在排空前还没取走的订单会随进程一起丢失，而会保存消息的 Broker 会把它留到下一次启动。 手写的 main 自己用 ruststream::logging::init() 安装日志记录器。

http.rs
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn Error>> {
    ruststream::logging::init()?;
    let broker = MemoryBroker::new().bindable();
    // A token now, a live publisher once the shop has connected the broker.
    let egress = broker.bind(Publish);
    // The shop starts first: a broker that refuses to connect stops the process before HTTP
    // takes a request.
    let running = shop(broker).start().await?;
    let egress = running.publisher(egress).await?;

    let store = Arc::new(Mutex::new(Store::default()));
    let stop_relay = CancellationToken::new();
    let relay = tokio::spawn(relay_outbox(
        Arc::clone(&store),
        egress,
        stop_relay.clone().cancelled_owned(),
    ));
    let router = axum::Router::new()
        .route("/orders", post(place_order))
        .with_state(store);

    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    tracing::info!("storefront on http://127.0.0.1:8080/orders");
    // HTTP stops on Ctrl+C, or when the shop stops on its own; the shop then drains.
    let stopping = running.stopping();
    axum::serve(listener, router)
        .with_graceful_shutdown(async move {
            tokio::select! {
                _ = tokio::signal::ctrl_c() => {}
                () = stopping => {}
            }
        })
        .await?;
    // HTTP has drained, so no request writes a row any more: the relay's last round hands the
    // pending rows over while the shop still runs, and only then does the shop drain.
    stop_relay.cancel();
    relay.await?;
    running.shutdown().await?;
    Ok(())
}

运行它，然后下一个订单：

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}]}'

第 5 章的日志会跟随这个订单走完整个商店。 健康检查端点在同一个服务器上向编排系统报告商店的状态。