Request/Response vs. Publish/Subscribe

Studijní materiál pro začínajícího Rust programátora. Vše je ukázáno na reálném kódu z projektu nexial-ventures/site a jeho PR #4 — Stream AI agentic-loop events over a per-user WebSocket.

1. O co jde

Když spolu dvě části systému komunikují (prohlížeč ↔ server, služba ↔ služba), skoro vždy použiješ jeden ze dvou základních komunikačních vzorů:

  • Request/Response (požadavek/odpověď) — jedna strana se zeptá, druhá odpoví. Konverzace tím končí.
  • Publish/Subscribe (pub/sub, publikuj/odebírej) — jedna strana se přihlásí k odběru a pak dostává zprávy kdykoli se něco stane, aniž by se musela znovu ptát.

Klíčová otázka, kterou si u každého vzoru polož: kdo iniciuje přenos dat?

Request/ResponsePublish/Subscribe
Kdo iniciuje přenos datKlient — každý přenos začíná jeho požadavkemPublisher — data „tlačí" (push), jakmile vzniknou
Kardinalita1 : 1 (jeden klient, jedna odpověď)1 : N (jedna událost, N odběratelů)
Vazba (coupling)Těsná — klient musí vědět koho a na co se ptátVolná — publisher neví, kdo (a jestli vůbec někdo) poslouchá
ČasSynchronní z pohledu klienta: pošli → čekej → dostaňAsynchronní: událost přijde, až nastane
Typický nosičHTTP (REST API), gRPC unary, funkce vracející hodnotuWebSocket, MQTT, Kafka, Redis pub/sub, tokio::sync::broadcast
Když se nikdo neptá / neposloucháNic se neděje — data „leží" na serveruUdálost zanikne (pokud ji broker nepersistuje)

2. Request/Response

Klient pošle požadavek a blokuje se (logicky, ne nutně vláknově), dokud nedostane odpověď. HTTP a nad ním postavené REST API je přesně tento vzor: každý GET /api/pages, POST /api/assistant/sessions/1/messages je jedna dvojice požadavek→odpověď a spojení (logicky) končí.

Klient (Vue SPA)Klient (Vue SPA)Server (Axum)Server (Axum)POST /api/assistant/sessions/1/messagesrun_turn() —LLM volání,toolcally...(může běžetdesítkysekund)200 OK + JSON (hotová odpověď)Konec. Další data klient dostane,jen když se znovu zeptá.

Silné stránky: jednoduchost, přirozené mapování na „zavolej funkci, dostaň výsledek", snadné kešování, jasné chybové stavy (HTTP status kódy).

Slabina, kterou řeší PR #4: dokud server odpověď nedokončí, klient nevidí nic. V projektu site běží uvnitř send_message celá agentická smyčka AI asistenta — LLM navrhne tool cally, ty se schvalují a spouští, upravují se stránky… a to celé bylo pro klienta neviditelné, dokud tah neskončil (issue #1 v repu):

"The assistant loop ran fully synchronously and silently — send_message only returned once a turn finished or paused for approval, so nothing about tool calls, approval prompts, or page edits was observable while a turn was in flight."

Jediná možnost, jak v čistém request/response světě dostat průběžné novinky, je polling — ptát se pořád dokola („už je něco nového? a teď? a teď?"). To plýtvá požadavky, zatěžuje server a stejně má latenci v délce polling intervalu.

3. Publish/Subscribe

V pub/sub se role obrací. Odběratel (subscriber) se jednou přihlásí k odběru a pak už jen pasivně přijímá. Vydavatel (publisher) posílá události do kanálu/topicu a vůbec neřeší, kdo poslouchá — o rozeslání všem odběratelům se stará prostředník (bus, broker).

Publisher (loop_driver)Publisher (loop_driver)EventBus (broadcast kanál)EventBus (broadcast kanál)Subscriber A (WS, tab 1)Subscriber A (WS, tab 1)Subscriber B (WS, tab 2)Subscriber B (WS, tab 2)subscribe(user_id)subscribe(user_id)nástroj sespouštípublish(ToolRunning)ToolRunningToolRunningnástrojdoběhlpublish(ToolResult)ToolResultToolResult

Všimni si tří věcí:

  1. Publisher neblokuje a nečeká na odpověď. Odpálí událost a jede dál.
  2. Jedna událost, N příjemců. Dva otevřené taby = dva odběratelé, oba dostanou totéž.
  3. Volná vazba. loop_driver v PR #4 vůbec neví, jestli je nějaký WebSocket otevřený. Když ne, událost prostě zanikne — a to je tam záměr (best-effort mirror).

4. Jak s tím souvisí REST API a WebSocket

Pozor na častou juniorskou záměnu: REST a WebSocket jsou transporty/protokoly, request/response a pub/sub jsou vzory. Souvislost je „typicky nese", ne „rovná se":

  • REST API = request/response nad HTTP. Jinak to ani nejde — HTTP samo je požadavek→odpověď a server nemůže klientovi nic poslat sám od sebe.
  • WebSocket = trvalé obousměrné spojení. Vznikne z HTTP požadavku (Upgrade: websocket handshake — spojení tedy pořád iniciuje klient!), ale pak zůstane otevřené a obě strany mohou kdykoli poslat zprávu. To z něj dělá ideální nosič pro pub/sub směrem k prohlížeči — server může pushovat.
  • WebSocket sám o sobě není pub/sub — je to jen roura. Pub/sub z něj dělá až logika okolo (kanály, odběry, fan-out). V PR #4 tu logiku dodává EventBus.
  • Alternativy pro server→klient push: SSE (Server-Sent Events) — jednosměrný stream nad HTTP, jednodušší než WS; long-polling — nouzovka, emuluje push pomocí dlouho visících požadavků.
Vzor (co)Transport (čím)lze i RPC přes WSRequest/ResponsePublish/SubscribeREST API / HTTPWebSocketServer-Sent EventsMQTT / Kafka / NATS

Kdo kdy iniciuje — shrnutí

FázeRESTWebSocket
Navázání spojeníklient (každý požadavek)klient (jednou, HTTP Upgrade)
Přenos dat klient→serverklientkdokoli, kdykoli
Přenos dat server→klientjen jako odpověď na požadavekserver kdykoli (push)

5. Význam pro real-time aplikace

Real-time UI (průběh AI agenta, chat, notifikace, ceny, telemetrie z IoT) má společné to, že události vznikají na serveru v čase, který klient nezná. S čistým request/response máš jen dvě špatné možnosti — ptát se často (drahé, stejně opožděné), nebo zřídka (ještě opožděnější). Pub/sub přes push kanál dává:

  • Latenci ~0 — událost letí klientovi hned, jak vznikne, žádné čekání na další poll.
  • Škálovatelnost — žádná bouře prázdných pollovacích požadavků; jedno spojení na klienta.
  • Fan-out zdarma — N tabů/zařízení jednoho uživatele dostane totéž bez práce navíc.

A druhá strana mince, kterou PR #4 řeší vědomě: pub/sub kanál je nespolehlivý ze své podstaty (odběratel nemusí existovat, může být pomalý, spojení může spadnout). Proto v PR platí: „send_message stays synchronous — the socket is a best-effort mirror, not a replacement for the HTTP response." Zdroj pravdy zůstává request/response + databáze; WebSocket je jen živé zrcadlo. Tohle je obecně zdravý návrhový princip.

6. Case study: PR #4 krok za krokem

Celé řešení má tři nové kusy: události (events.rs), bus (event_bus.rs) a WS endpoint (routes/api/ws.rs); do stávající smyčky (loop_driver.rs) se jen vložily publish(...) cally na každém přechodu.

Axum server1. HTTP požadavek (request/response)publish(LoopEvent)broadcast (pub/sub)JSON text frame – push2. HTTP odpověď po dokončení tahuVue admin SPAPOST send_messageloop_driver::run_turnEventBus – DashMap user_id → broadcast::SenderGET /api/ws handler

6.1 Události — src/ai/events.rs

Každý „pozorovatelný okamžik" smyčky je varianta enumu. Serde atribut tag = "type" vyrobí JSON, kde klient pozná typ události z pole "type":

#[derive(Clone, Debug, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum LoopEvent {
    AssistantText { session_id: i32, text: Option<String> },
    ToolProposed { session_id: i32, tool_call_id: String, name: String, args: Value },
    ToolAwaitingApproval { session_id: i32, tool_call_id: String, name: String },
    ToolRunning { session_id: i32, tool_call_id: String, name: String },
    ToolResult { session_id: i32, tool_call_id: String, name: String, is_error: bool, output: Value },
    PageChanged { session_id: i32, kind: PageChangedKind, action: PageChangedAction, path: Option<String>, id: Option<i32> },
    TurnDone { session_id: i32 },
    TurnPaused { session_id: i32 },
    TurnError { session_id: i32, message: String },
}

Na drátě pak lítá např. {"type":"tool_running","session_id":7,"tool_call_id":"x","name":"edit_page"}. Všimni si, že každá varianta nese session_id — kanál je per-user, takže klient s více otevřenými sezeními filtruje na své straně.

Proč Clone? Broadcast kanál doručuje každému odběrateli kopii zprávy — bez Clone by to nešlo.

6.2 Bus — src/ai/event_bus.rs

Celý pub/sub prostředník má ~45 řádků. Jeden tokio::sync::broadcast kanál na uživatele, uložený v DashMap (konkurentní HashMap — nepotřebuješ Mutex<HashMap>):

const CHANNEL_CAPACITY: usize = 256;

pub struct EventBus {
    channels: DashMap<i32, broadcast::Sender<LoopEvent>>,
}

impl EventBus {
    fn sender_for(&self, user_id: i32) -> broadcast::Sender<LoopEvent> {
        self.channels
            .entry(user_id)
            .or_insert_with(|| broadcast::channel(CHANNEL_CAPACITY).0)
            .clone()
    }

    /// Best-effort publish. `send` vrací Err jen když právě není žádný
    /// odběratel — což je běžný stav, když není otevřený WebSocket.
    pub fn publish(&self, user_id: i32, event: LoopEvent) {
        let _ = self.sender_for(user_id).send(event);
    }

    pub fn subscribe(&self, user_id: i32) -> broadcast::Receiver<LoopEvent> {
        self.sender_for(user_id).subscribe()
    }
}

Poznámky pro juniora:

  • let _ = ...send(...) — vědomé zahození chyby. broadcast::Sender::send selže jen tehdy, když nikdo neposlouchá, a to je tady OK (události se nepersistují ani nepřehrávají). Kdyby to byl kritický kanál, tohle by byl bug; tady je to dokumentované rozhodnutí.
  • Kanál se vytváří lazy — až první publish nebo subscribe pro daného uživatele.
  • EventBus žije v AppState jako Arc<EventBus> — sdílený stav mezi všemi handlery a tasky, klasický Axum vzor.

6.3 Publisher — src/ai/loop_driver.rs

Do existující smyčky se na každém přechodu jen přidal publish. Ukázka (spuštění povoleného nástroje):

Effect::Allow => {
    event_bus.publish(user_id, LoopEvent::ToolRunning {
        session_id,
        tool_call_id: tc.id.clone(),
        name: tc.name.clone(),
    });
    match tool_registry.dispatch(&tc.name, tc.args.clone(), &dispatch_ctx).await {
        Ok(v) => (v, false),
        Err(e) => (json!({ "text": e.to_string() }), true),
    }
}

Smyčka se nijak nezpomalila ani nezkomplikovala — publish je neblokující a nemůže selhat. Přesně takhle má vypadat volná vazba: publisher o odběratelích nic neví.

6.4 Subscriber — src/routes/api/ws.rs

WebSocket handler v Axum: HTTP požadavek se upgraduje na WS a pak běží smyčka, která přeposílá události jako JSON text frames:

async fn ws_handler(
    ws: WebSocketUpgrade,
    State(state): State<AppState>,
    Extension(user_id): Extension<i32>,
) -> impl IntoResponse {
    ws.on_upgrade(move |socket| handle_socket(socket, state, user_id))
}

async fn handle_socket(mut socket: WebSocket, state: AppState, user_id: i32) {
    let mut rx = state.event_bus.subscribe(user_id);
    loop {
        tokio::select! {
            recv = rx.recv() => match recv {
                Ok(event) => {
                    let Ok(text) = serde_json::to_string(&event) else { continue };
                    if socket.send(Message::Text(text.into())).await.is_err() { break; }
                }
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    tracing::warn!(user_id, skipped = n, "ws receiver lagged");
                    continue;
                }
                Err(broadcast::error::RecvError::Closed) => break,
            },
            msg = socket.recv() => match msg {
                Some(Ok(Message::Close(_))) | None => break,
                Some(Ok(_)) => continue,
                Some(Err(_)) => break,
            },
        }
    }
}

Tři důležité detaily:

  1. tokio::select! čeká na dvě věci najednou: novou událost z busu nebo zprávu/zavření od klienta. Kterákoli přijde dřív, ta se obslouží. Bez toho bys nepoznal, že klient odešel, dokud bys mu nezkusil něco poslat.
  2. RecvError::Lagged(n) — broadcast kanál má kapacitu 256; když odběratel nestíhá, nejstarší zprávy se zahodí a receiver dostane Lagged. Tady se to jen zaloguje a jede dál — další ukázka best-effort návrhu. (Srovnej: Kafka by zprávy držela, mpsc kanál by brzdil odesílatele — každý nástroj dělá jiný trade-off.)
  3. Endpoint je za require_login_api middlewarem a odebírá jen kanál přihlášeného uživatele — izolace dat mezi uživateli je vyřešená už volbou klíče v DashMap.

6.5 Co si z návrhu odnést

  • Oba vzory se doplňují, nenahrazují. send_message zůstal synchronní request/response (spolehlivý, zdroj pravdy v DB); WebSocket je doplňkový pub/sub mirror pro UX.
  • Vrstvení: vzor (pub/sub) ↔ mechanismus uvnitř procesu (broadcast kanál) ↔ transport ke klientovi (WebSocket). Tři nezávislé volby.
  • Vědomé trade-offy jsou v komentářích — proč je publish best-effort, proč se Lagged jen loguje. Piš to tak taky.

7. Cvičení

  1. Otevři si PR #4 a najdi všechna místa, kde se volá event_bus.publish. Ke každému přiřaď variantu LoopEvent a rozmysli, co v UI klient uvidí.
  2. Co by se stalo, kdyby LoopEvent neimplementoval Clone? Zkus si to představit na úrovni typů broadcast::Sender.
  3. Proč je kanál per-user a ne jeden globální, nebo per-session? Jaké výhody/nevýhody by měly druhé dvě varianty?
  4. Napiš si minimální verzi sám: Axum server s GET /ws, tokio::sync::broadcast kanálem a endpointem POST /publish, který přijme text a rozešle ho všem připojeným WS klientům. (~60 řádků, viz axum examples níže.)
  5. Rozšíření: jak bys řešil, že klient po výpadku spojení přišel o události? (Nápověda: sekvenční čísla + dohrání z DB přes request/response — přesně proto je dobře, že DB zůstala zdrojem pravdy.)

8. Reference

Vzory

Protokoly

Rust / Tokio / Axum

Studovaný kód