Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Рядом с HTTP-сервером, outbox

Витрина принимает заказы по HTTP и сохраняет их вместе с событиями. Ретранслятор передаёт события брокеру, а магазин работает рядом с сервером.

Когда сервис также обслуживает HTTP, процессом управляет HTTP-сервер, а магазин работает рядом с ним. start запускает магазин в фоне и возвращает управление, когда магазин работает. Поэтому если брокер отказывает в подключении, процесс останавливается раньше, чем сервер примет первый запрос. HTTP-запрос не должен сам публиковать сообщения в брокер. Иначе сбой брокера приведёт к ошибке запроса, а заказ, сохранённый без события, никогда не дойдёт до магазина. Транзакционный outbox решает обе проблемы. Запрос записывает заказ и его событие в одной транзакции, а ретранслятор позже публикует сохранённые события. Сбой брокера только откладывает события, и ни одно из них не теряется. Если процесс аварийно завершится между публикацией и удалением события, событие будет опубликовано снова, поэтому магазин получает каждый заказ хотя бы один раз.

Хранилище​

В настоящей базе данных заказы и 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. Она только записывает в хранилище, поэтому отвечает 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, поэтому ретранслятор не меняется при смене брокера.

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 позволяет процессу получить токен издателя до запуска магазина, а publisher превращает токен в работающего издателя после подключения брокера. Сервер останавливается по Ctrl+C или когда stopping сообщает, что магазин остановился сам. Когда HTTP-сервер завершил запросы, процесс останавливает ретранслятор, и его последний проход передаёт брокеру оставшиеся строки, пока магазин ещё работает. Затем магазин дорабатывает начатые доставки и останавливается через shutdown. MemoryBroker тоже хранит очереди в памяти процесса: заказ, который последний проход передал, а магазин не успел забрать до остановки, пропадёт вместе с процессом. Брокер, который хранит сообщения, сбережёт его до следующего запуска. 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 показывает путь заказа через магазин. Конечная точка проверки состояния того же сервера сообщает оркестратору о состоянии магазина.