WebSocket & Komunikasi Real-time — Belajar FastAPI

📌 Kapan Memakai WebSocket

Kebutuhan	Pilihan yang tepat
Notifikasi satu arah dari server	Server-Sent Events (SSE) — lebih sederhana, otomatis menyambung ulang
Chat, kolaborasi, game	WebSocket — dua arah
Pembaruan tiap 30 detik atau lebih jarang	Polling biasa — jangan berlebihan
Unggah/unduh berkas besar	HTTP biasa

💡 Jangan langsung melompat ke WebSocket

WebSocket membawa beban nyata: koneksi berumur panjang mempersulit load balancing, rilis bergulir memutus semua koneksi, dan Anda harus menangani penyambungan ulang sendiri. Bila arah datanya hanya server ke klien, SSE hampir selalu pilihan yang lebih baik — ia berjalan di atas HTTP biasa dan peramban menanganinya secara otomatis.

💻 WebSocket Dasar

main.py

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws/{ruangan}")
async def endpoint_ws(websocket: WebSocket, ruangan: str):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[{ruangan}] {data}")
    except WebSocketDisconnect:
        print(f"klien keluar dari ruangan {ruangan}")

🗂️ Connection Manager

manager.py

from collections import defaultdict
from fastapi import WebSocket

class PengelolaKoneksi:
    def __init__(self):
        self.aktif: dict[str, set[WebSocket]] = defaultdict(set)

    async def sambung(self, ws: WebSocket, ruangan: str):
        await ws.accept()
        self.aktif[ruangan].add(ws)

    def putus(self, ws: WebSocket, ruangan: str):
        self.aktif[ruangan].discard(ws)
        if not self.aktif[ruangan]:
            del self.aktif[ruangan]

    async def siarkan(self, pesan: dict, ruangan: str):
        mati = []
        for ws in self.aktif[ruangan]:
            try:
                await ws.send_json(pesan)
            except Exception:
                mati.append(ws)          # koneksi sudah putus
        for ws in mati:
            self.putus(ws, ruangan)

pengelola = PengelolaKoneksi()

⚠️ Bersihkan koneksi mati saat menyiarkan

Koneksi bisa putus tanpa memicu WebSocketDisconnect — misalnya ketika klien kehilangan jaringan secara tiba-tiba. Bila set koneksi tidak pernah dibersihkan, ia terus tumbuh dan setiap siaran membuang waktu mencoba mengirim ke soket yang sudah mati. Kumpulkan yang gagal saat menyiarkan lalu buang, seperti pada kode di atas.

🔐 Autentikasi WebSocket

🚫 Peramban tidak bisa mengirim header Authorization pada WebSocket

API WebSocket di peramban tidak menyediakan cara menambahkan header kustom. Karena itu pola Authorization: Bearer … tidak bisa dipakai langsung.

Tiga pendekatan

# 1. Token lewat query parameter -- sederhana, TAPI token bisa
#    tercatat di log akses proxy. Pakai token berumur sangat pendek.
@app.websocket("/ws")
async def ws(websocket: WebSocket, token: str = Query(...)):
    try:
        pengguna = verifikasi_token(token)
    except InvalidToken:
        await websocket.close(code=1008)   # policy violation
        return
    await websocket.accept()

# 2. Cookie HttpOnly -- paling aman untuk aplikasi web,
#    karena cookie ikut terkirim otomatis saat handshake.
@app.websocket("/ws")
async def ws(websocket: WebSocket):
    sesi = websocket.cookies.get("sesi")
    ...

# 3. Autentikasi pesan pertama setelah koneksi terbuka.
@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    auth = await asyncio.wait_for(websocket.receive_json(), timeout=5.0)
    pengguna = verifikasi_token(auth["token"])

🔀 Masalah Besar: Banyak Worker

uvicorn main:app --workers 4

Worker 1: [klien A, klien B]
Worker 2: [klien C]
Worker 3: [klien D, klien E]
Worker 4: []

Klien A mengirim pesan ke Worker 1.
pengelola.siarkan() hanya menjangkau klien B.
❌ Klien C, D, dan E tidak pernah menerimanya.

🚫 Dictionary di dalam memori tidak dibagi antar proses

Setiap worker uvicorn adalah proses terpisah dengan memorinya sendiri. Ini bug yang sangat sering terjadi: fitur chat bekerja sempurna saat pengembangan dengan satu worker, lalu rusak begitu di-deploy ke produksi dengan banyak worker atau banyak pod.

Solusi: Redis Pub/Sub sebagai jembatan

import asyncio, json
import redis.asyncio as redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = redis.from_url("redis://localhost:6379")
    tugas = asyncio.create_task(dengarkan_redis(app))
    yield
    tugas.cancel()
    await app.state.redis.aclose()

async def dengarkan_redis(app: FastAPI):
    psub = app.state.redis.pubsub()
    await psub.psubscribe("ruangan:*")
    async for pesan in psub.listen():
        if pesan["type"] != "pmessage":
            continue
        ruangan = pesan["channel"].decode().split(":", 1)[1]
        # Kirim ke klien yang terhubung ke worker INI saja
        await pengelola.siarkan(json.loads(pesan["data"]), ruangan)

@app.websocket("/ws/{ruangan}")
async def ws(websocket: WebSocket, ruangan: str):
    await pengelola.sambung(websocket, ruangan)
    try:
        while True:
            data = await websocket.receive_json()
            # Terbitkan ke Redis -- SEMUA worker akan menerimanya
            await app.state.redis.publish(f"ruangan:{ruangan}", json.dumps(data))
    except WebSocketDisconnect:
        pengelola.putus(websocket, ruangan)

📡 Server-Sent Events sebagai Alternatif

sse.py

from fastapi.responses import StreamingResponse

async def aliran_notifikasi(pengguna_id: int):
    psub = redis_klien.pubsub()
    await psub.subscribe(f"notif:{pengguna_id}")
    try:
        async for pesan in psub.listen():
            if pesan["type"] == "message":
                yield f"data: {pesan['data'].decode()}\n\n"
    finally:
        await psub.unsubscribe()

@app.get("/notifikasi")
async def notifikasi(pengguna_id: int):
    return StreamingResponse(
        aliran_notifikasi(pengguna_id),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

💡 Keunggulan SSE yang sering diabaikan

Peramban otomatis menyambung ulang SSE yang terputus, sesuatu yang harus Anda tulis sendiri untuk WebSocket. SSE juga berjalan di atas HTTP biasa sehingga bekerja mulus dengan proxy, CDN, dan autentikasi berbasis cookie. Header X-Accel-Buffering: no penting bila Anda memakai Nginx, karena tanpa itu Nginx akan menahan aliran di buffer.