Belajar Axum - WebSocket & SSE

#Pendahuluan

Sejauh ini komunikasi kita satu arah: klien meminta, server menjawab. Episode ini membuka arah kedua: server push. Axum mendukung dua mekanisme realtime secara native — WebSocket (dua arah, full-duplex) dan Server-Sent Events / SSE (satu arah, dari server ke klien) — keduanya ada di inti framework, bukan ekstensi.

Mengapa ini penting? Chat, notifikasi, dashboard live, kolaborasi dokumen — semua fitur modern ini butuh push. Dan Axum 0.8.9 bahkan menambahkan dukungan subprotocol WebSocket, menjadikannya pilihan serius untuk aplikasi realtime di 2026. Kita akan membangun broadcast server sederhana yang bisa menyalurkan pesan antar banyak klien.

#WebSocket: Konsep dan Setup

WebSocket adalah protokol yang mempertahankan koneksi TCP terbuka, memungkinkan kedua pihak kirim pesan kapan saja. Di Axum, endpoint WebSocket adalah handler biasa yang mengekstrak WebSocketUpgrade:

use axum::{
    extract::ws::{WebSocket, WebSocketUpgrade},
    response::Response,
};
 
async fn ws_handler(ws: WebSocketUpgrade) -> Response {
    ws.on_upgrade(handle_socket)
}
 
async fn handle_socket(mut socket: WebSocket) {
    while let Some(msg) = socket.recv().await {
        if let Ok(msg) = msg {
            if msg.is_text() {
                let text = msg.to_text().unwrap().to_string();
                // echo balik ke pengirim
                let _ = socket.send(axum::extract::ws::Message::Text(text.into())).await;
            }
        }
    }
}

Anatomi:

1. Handler menerima WebSocketUpgrade sebagai extractor — ini menandakan route ini siap untuk upgrade HTTP ke WebSocket.
2. ws.on_upgrade(handler) — mengupgrade koneksi dan menjalankan handle_socket sebagai task terpisah.
3. handle_socket membaca pesan dalam loop; recv() mengembalikan None saat koneksi ditutup.

Pasang route:

use axum::{routing::get, Router};
 
fn app() -> Router {
    Router::new().route("/ws", get(ws_handler))
}

Uji dengan tool CLI websocat atau wscat:

npx wscat -c ws://localhost:3000/ws

Connected (press CTRL+C to quit)
> halo
< halo

#WebSocket Subprotocol (Baru di Axum 0.8.9)

Sebelum 0.8.9, kalian harus membungkus sendiri subprotocol. Axum 0.8.9 (14 April 2026) menambahkan dukungan native: klien menyebut protokol yang didukung di header Sec-WebSocket-Protocol, server memilih salah satu. Ini penting untuk protokol seperti JSON-RPC over WebSocket atau MQTT:

use axum::extract::{
    ws::{WebSocket, WebSocketUpgrade},
    TypedHeader,
};
use axum::http::header;
use headers::HeaderMapExt;
 
async fn ws_handler(
    ws: WebSocketUpgrade,
    protocols: Option<TypedHeader<headers::Origin>>, // placeholder, contoh header
) -> axum::response::Response {
    ws.subprotocol("chat").on_upgrade(handle_socket)
}

Pada Axum 0.8.9, WebSocketUpgrade::subprotocol(...) memilih subprotocol yang dinegosiasikan dengan klien — mismatch protokol ditolak otomatis. Klien (misal JavaScript) mengirim subprotocol lewat konstruktor new WebSocket(url, "chat"). Pastikan versi axum kalian benar-benar 0.8.9+ jika ingin memakai fitur ini:

cargo tree -i axum

#Broadcast Server: Membagikan Pesan ke Semua Klien

Endpoint echo di atas hanya menghubungkan satu klien dengan dirinya sendiri. Untuk chat/notifikasi, kita butuh broadcast: satu pesan masuk, semua klien terima. Pola standarnya tokio::sync::broadcast:

use std::sync::Arc;
use tokio::sync::broadcast;
 
#[derive(Clone)]
struct AppState {
    tx: broadcast::Sender<String>,
}
 
#[derive(Clone)]
struct AppStateBuilder {
    tx: broadcast::Sender<String>,
}
 
fn new_state() -> AppState {
    let (tx, _rx) = broadcast::channel(64);
    AppState { tx }
}
 
async fn ws_handler(
    State(state): State<AppState>,
    ws: WebSocketUpgrade,
) -> axum::response::Response {
    ws.on_upgrade(move |socket| handle_socket(state, socket))
}
 
async fn handle_socket(state: AppState, mut socket: WebSocket) {
    let mut rx = state.tx.subscribe();
 
    loop {
        tokio::select! {
            incoming = socket.recv() => {
                match incoming {
                    Some(Ok(msg)) if msg.is_text() => {
                        let text = msg.to_text().unwrap().to_string();
                        let _ = state.tx.send(text);   // broadcast ke semua
                    }
                    Some(Ok(_)) => {}
                    _ => break,                        // koneksi tertutup
                }
            }
            broadcasted = rx.recv() => {
                match broadcasted {
                    Ok(text) => {
                        let _ = socket.send(axum::extract::ws::Message::Text(text.into())).await;
                    }
                    Err(_) => break,                   // sender dibuang
                }
            }
        }
    }
}

Kunci pola ini ada dua:

1. broadcast::channel(64) — antrian broadcast dengan buffer 64 pesan. Semua subscriber (klien) menerima salinan; jika subscriber lambat dan buffer penuh, ia di-drop dan harus re-subscribe.
2. tokio::select! — menunggu dua kejadian sekaligus: pesan dari soket (klien kirim) atau pesan dari channel (klien lain kirim). Tak satu pun memblokir yang lain.

Ini adalah pola inti realtime Axum — dari chat sampai dashboard. Kita pakai lagi di episode 22 dengan worker task.

#Server-Sent Events (SSE)

SSE adalah alternatif satu arah yang lebih sederhana: server mengirim event terus-menerus ke klien lewat koneksi HTTP biasa (tidak ada upgrade protokol). Kelebihan dibanding WebSocket: otomatis re-connect, tidak perlu library khusus di klien (EventSource native), dan berjalan di atas HTTP standar.

Di Axum, SSE memakai Event stream:

use axum::{
    response::sse::{Event, Sse},
};
use tokio_stream::{wrappers::BroadcastStream, StreamExt};
 
async fn sse_handler(State(state): State<AppState>) -> Sse<impl futures_core::Stream<Item = Result<Event, axum::Error>>> {
    let stream = BroadcastStream::new(state.tx.subscribe()).map(|item| {
        let text = item.unwrap_or_default();
        Ok(Event::default().data(text))
    });
    Sse::new(stream).keep_alive(
        axum::response::sse::KeepAlive::new().interval(std::time::Duration::from_secs(15)),
    )
}

Kelebihan pola ini: satu channel broadcast bisa menyuplai WebSocket dan SSE sekaligus. Klien SSE cukup:

const es = new EventSource("/events");
es.onmessage = (e) => console.log("event:", e.data);

Dan keep_alive mengirim komentar kosong tiap 15 detik — mencegah koneksi diputus proxy yang menganggap idle (masalah yang dibahas lagi di episode 20).

#Memilih: WebSocket atau SSE?

Aturan praktis: sebut SSE dulu — jauh lebih sederhana. WebSocket hanya kalau klien perlu mengirim data realtime dengan frekuensi tinggi (chat, kolaborasi).

#Batasan dan Pitfall

Beberapa hal yang perlu diwaspadai saat men-deploy realtime:

• Proxy & load balancer — WebSocket butuh proxy yang mendukung upgrade (nginx dengan Upgrade header, episode 20). SSE rentan diputus jika proxy punya idle timeout — itulah gunanya KeepAlive.
• Skalabilitas horizontal — broadcast bekerja per-instance. Jika aplikasi di-scale ke banyak instance, channel antar instance butuh transport eksternal (Redis pub/sub, NATS). Kita singgung di episode 22 dan 26.
• Koneksi idle — atur timeout dan tutup koneksi yang tidak aktif untuk mencegah pemborosan resource.
• Batas pesan — ukuran maksimal pesan WebSocket diatur protokol; jangan kirim payload raksasa lewat soket.

#Penutup

Pada episode 12 ini aplikasi kalian punya kemampuan realtime:

• WebSocket via WebSocketUpgrade + on_upgrade, dengan subprotocol native di Axum 0.8.9.
• Broadcast channel tokio::sync::broadcast untuk chat/notifikasi multi-klien.
• tokio::select! untuk menunggu dua sumber event tanpa blokir.
• SSE dengan Sse + Event + KeepAlive.
• Panduan memilih WebSocket vs SSE dan pitfall deployment.

Di episode 13 selanjutnya kita amankan akses: authentication & authorization — JWT, session, basic auth, dan middleware RBAC untuk endpoint yang dilindungi. Sampai jumpa di episode 13!