WebSockets: timp real care rezistă rețelei

HTTP e o întrebare urmată de un răspuns. Un WebSocket e o linie care rămâne deschisă, iar serverul poate vorbi primul: un mesaj nou în chat, un preț care s-a schimbat, un build care s-a terminat. Deschiderea liniei e o singură linie de cod. Munca adevărată e să o ții în viață în tren, pe un laptop care intră în sleep sau cât serverul face un deploy.

Pagina construiește un socket tipizat care se reconectează singur și e expus ca signals: starea conexiunii, mesajele, un outbox pentru ce trimiți cât ești offline și o reconectare care nu dărâmă serverul din nou în clipa în care revine.

De ce contează

În producție, socket-urile mor tot timpul: Wi-Fi-ul schimbă rețeaua, proxy-urile închid conexiunile inactive după un minut, deploy-urile repornesc serverele. Un socket fără reconectare arată perfect în demo, apoi încetează în tăcere să se actualizeze în mâinile utilizatorului — fără nicio eroare pe ecran.

Minimul: webSocket() din RxJS

RxJS are un subject pentru WebSocket: subscribe() deschide conexiunea, next() trimite, unsubscribe() o închide. Implicit, mesajele sunt JSON în ambele direcții (serializer și deserializer schimbă asta).

import { webSocket } from 'rxjs/webSocket';

interface ChatMessage {
  type: 'chat';
  user: string;
  text: string;
}

const socket = webSocket<ChatMessage>('wss://api.example.com/live');

// subscribe() deschide conexiunea; fiecare mesaj vine deja parsat (JSON.parse)
const sub = socket.subscribe({
  next: (msg) => console.log(msg.user, msg.text),
  error: (err) => console.error('conexiune pierdută', err), // stream-ul SE TERMINĂ aici
  complete: () => console.log('închisă curat de server'), // ...sau aici
});

socket.next({ type: 'chat', user: 'ana', text: 'hi' }); // trimis ca JSON.stringify(...)

sub.unsubscribe(); // închide conexiunea

Două detalii contează mai târziu. Stream-ul se termină când cade conexiunea: cu error dacă a fost o cădere anormală, cu complete dacă serverul a închis-o curat. Iar ce trimiți înainte să fie deschisă conexiunea se păstrează și pleacă la conectare.

O conexiune care revine: backoff cu jitter

Când cade linia, cea mai proastă variantă e să reîncerci imediat, în buclă. Dacă serverul tocmai a repornit, toți clienții îl lovesc în aceeași secundă — thundering herd — și serverul cade din nou. Rețeta obișnuită: aștepți mai mult după fiecare eșec (backoff exponențial), pui un plafon și adaugi întâmplare (jitter), ca clienții să se împrăștie.

/**
 * „Full jitter”: o așteptare aleatoare între 0 și min(cap, base · 2^attempt).
 * încercarea 1 → 0–1 s, 2 → 0–2 s, 3 → 0–4 s ... niciodată peste 10 s.
 */
export function fullJitter(attempt: number, base = 500, cap = 10_000): number {
  return Math.random() * Math.min(cap, base * 2 ** attempt);
}

Live200 de clienți se reconectează după un restart al serverului

Server căzut timp de10 s

încercare eșuatăreconectat

vârf după restart: 200
încercări în total: 2200
reconectați: 200 / 200
ultimul revine la: 10.2 s

Intervalul fix de 1 s ciocănește serverul cât timp e căzut. Exponențialul fără jitter trimite mai puține cereri, dar în valuri sincronizate: toți cei care au eșuat în același moment reîncearcă în același moment. Full jitter împrăștie revenirile pe câteva secunde, așa că serverul primește un firicel în loc de un zid; prețul e că cei mai ghinioniști clienți revin cu câteva secunde mai târziu. Plafonul ține așteptarea în limite rezonabile. Pentru un socket, e aproape întotdeauna schimbul potrivit.

retry() nu ajunge: închiderile curate

retry() reacționează doar la erori. Un server care repornește politicos închide conexiunile cu codul 1001 sau 1012 („service restart”) și wasClean: true — iar atunci stream-ul se încheie, nu dă eroare. Fără repeat(), aplicația rămâne deconectată exact în timpul deploy-urilor. Resetează contorul de încercări în openObserver, când linia e chiar deschisă, nu la primul mesaj.

import { repeat, retry, timer } from 'rxjs';
import { webSocket } from 'rxjs/webSocket';
import { fullJitter } from './backoff';
import type { ServerMessage } from './live-socket';

let attempt = 0;
const wait = () => timer(fullJitter(++attempt));

const socket = webSocket<ServerMessage>({
  url: 'wss://api.example.com/live',
  openObserver: { next: () => (attempt = 0) }, // resetăm când linia e chiar deschisă
});

socket
  .pipe(
    retry({ delay: wait }), // eroare de rețea, închidere anormală (1006)
    repeat({ delay: wait }), // închidere CURATĂ (1001, 1012 „service restart”): stream-ul se încheie!
  )
  .subscribe((msg) => console.log(msg));

Bufferul pe care nu te poți baza

WebSocketSubject păstrează ce trimiți înainte să se deschidă socket-ul, dar înlocuiește bufferul la fiecare încercare eșuată (_resetState() în sursa RxJS). Mesajele trimise cât rețeaua e căzută se pierd imediat ce o încercare de reconectare eșuează. Dacă contează, ține-ți propriul outbox și golește-l în openObserver.

Socket-ul ca serviciu cu signals

Componentele n-ar trebui să știe de backoff sau de JSON. Un serviciu ține o singură conexiune pentru toată aplicația, expune starea și mesajele ca signals și oferă un singur send(). Cât linia e căzută, mesajele așteaptă într-un outbox limitat și pleacă în ordine după reconectare.

import { DestroyRef, Service, inject, signal } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval, repeat, retry, timeout, timer } from 'rxjs';
import { webSocket } from 'rxjs/webSocket';
import { fullJitter } from './backoff';

export type ServerMessage =
  | { type: 'chat'; user: string; text: string }
  | { type: 'presence'; online: number }
  | { type: 'pong' };
export type ClientMessage = { type: 'chat'; text: string } | { type: 'ping' };

export type SocketStatus = 'idle' | 'connecting' | 'open' | 'reconnecting';

const PING_EVERY = 15_000;
const DEAD_AFTER = 20_000; // niciun mesaj timp de 20 s = conexiune moartă

/** O singură conexiune pentru toată aplicația, expusă ca signals. */
@Service()
export class LiveSocket {
  private readonly destroyRef = inject(DestroyRef);

  readonly status = signal<SocketStatus>('idle');
  readonly online = signal(0);
  readonly messages = signal<readonly ServerMessage[]>([]);
  /** Ce s-a trimis cât linia a fost căzută; pleacă, în ordine, la următorul „open”. */
  readonly outbox = signal<readonly ClientMessage[]>([]);

  private attempt = 0;
  // RxJS folosește un singur tip pentru ambele direcții
  private readonly socket = webSocket<ServerMessage | ClientMessage>({
    url: 'wss://api.example.com/live',
    openObserver: { next: () => this.opened() },
    closeObserver: { next: () => this.status.set('reconnecting') },
  });

  connect(): void {
    if (this.status() !== 'idle') return;
    this.status.set('connecting');
    const wait = () => timer(fullJitter(++this.attempt));

    this.socket
      .pipe(
        timeout({ each: DEAD_AFTER }), // heartbeat: tăcere → eroare → retry
        retry({ delay: wait }),
        repeat({ delay: wait }),
        takeUntilDestroyed(this.destroyRef),
      )
      .subscribe((msg) => this.receive(msg as ServerMessage));

    interval(PING_EVERY)
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe(() => {
        if (this.status() === 'open') this.socket.next({ type: 'ping' });
      });
  }

  send(msg: ClientMessage): void {
    if (this.status() === 'open') this.socket.next(msg);
    else this.outbox.update((list) => [...list, msg].slice(-100)); // limitat!
  }

  private opened(): void {
    this.status.set('open');
    this.attempt = 0;
    const pending = this.outbox();
    this.outbox.set([]);
    pending.forEach((msg) => this.socket.next(msg));
  }

  private receive(msg: ServerMessage): void {
    if (msg.type === 'presence') this.online.set(msg.online);
    else if (msg.type === 'chat') this.messages.update((list) => [...list, msg].slice(-200));
  }
}

LiveO conexiune live: taie rețeaua, repornește serverul

neconectat0 online

Evenimentele conexiunii apar aici.

Demo-ul rulează codul de mai sus, cu intervale mai scurte. Doar WebSocketCtor e înlocuit: în loc de un WebSocket adevărat primește un server în memorie, care poate reporni, poate pierde rețeaua sau poate îngheța o conexiune.

Heartbeat: o conexiune moartă care pare vie

O conexiune poate muri fără niciun eveniment de închidere: se închide capacul laptopului, un NAT uită maparea, rețeaua mobilă schimbă celula. TCP poate să nu observe minute întregi, iar readyState spune în continuare OPEN. Leacul e un heartbeat: clientul trimite un ping la interval fix, iar timeout({ each }) tratează orice tăcere mai lungă decât limita ca pe o linie moartă. Eroarea ajunge la retry(), care deschide o conexiune nouă.

Ce trimiți pe o linie moartă-dar-deschisă se pierde fără nicio eroare: send() reușește, datele nu ajung niciodată. Pentru mesajele care trebuie să ajungă (o comandă, o plată), dă-i fiecăruia un id, așteaptă un ack de la server și retrimite după reconectare ce n-a fost confirmat. Serverul ignoră duplicatele după id.

SSR, autentificare, taburi și resync

SSR și prerender. Codul rulează și pe server, unde nu există niciun socket care merită deschis. Conectează-te în afterNextRender (sau în spatele lui isPlatformBrowser) și lasă randarea de pe server să arate o stare goală. Vezi cod compatibil cu serverul.
Autentificare. Browserul nu te lasă să pui un header Authorization pe un WebSocket. Variante: un cookie de sesiune (același site), un ticket scurt, de unică folosință, luat prin HTTP și pus în URL, sau un prim mesaj auth. Nu pune un token cu viață lungă în URL — URL-urile ajung în logurile serverului. Token-urile și refresh-ul sunt în Autentificare: token-uri și refresh.
Mai multe taburi. Fiecare tab își deschide propria conexiune. Dacă asta e o problemă (limite de conexiuni, notificări duble), un singur tab poate deține socket-ul și le poate da celorlalte mesajele prin BroadcastChannel sau printr-un SharedWorker.
Resync după reconectare. Cât ai lipsit, serverul a trimis lucruri pe care nu le-ai primit. După fiecare open, cere ce ai pierdut (since=lastEventId) sau reîncarcă starea prin HTTP. Modificările făcute între timp sunt subiectul paginii despre actualizări optimiste.

import { HttpClient } from '@angular/common/http';
import { defer, retry, switchMap, timer } from 'rxjs';
import { webSocket } from 'rxjs/webSocket';
import { fullJitter } from './backoff';
import type { ServerMessage } from './live-socket';

// Browserul nu poate pune un header Authorization pe un WebSocket.
// 1) un ticket scurt prin HTTP (interceptorul de auth adaugă token-ul ca de obicei)
// 2) ticket-ul e de unică folosință și trăiește ~30 s, deci poate sta în URL
// 3) defer() → fiecare reconectare cere un ticket NOU
export function authenticatedSocket(http: HttpClient) {
  return defer(() => http.post<{ ticket: string }>('/api/ws-ticket', {})).pipe(
    switchMap(({ ticket }) => webSocket<ServerMessage>(`wss://api.example.com/live?ticket=${ticket}`)),
    retry({ delay: (_, n) => timer(fullJitter(n)) }),
  );
}

WebSocket sau Server-Sent Events?

Dacă vorbește doar serverul — notificări, un feed, progresul unui job, text trimis token cu token — Server-Sent Events sunt mai simple: HTTP obișnuit, reconectare inclusă în browser și id-ul ultimului eveniment trimis automat înapoi.

	WebSocket	Server-Sent Events
Direcție	în ambele sensuri	server → browser
Protocol	propriu (`ws://` / `wss://`, după un upgrade HTTP)	HTTP obișnuit (`text/event-stream`)
Reconectare	o scrii tu	inclusă, cu `Last-Event-ID`
Date	text și binar	doar text
Folosire tipică	chat, editare colaborativă, jocuri, tranzacționare în ambele sensuri	notificări, feed-uri live, progresul unui job, streaming de text AI

import { Observable } from 'rxjs';

/** Server-Sent Events: doar server → browser, HTTP simplu, reconectare inclusă. */
export function serverEvents<T>(url: string): Observable<T> {
  return new Observable<T>((subscriber) => {
    const source = new EventSource(url, { withCredentials: true });
    source.onmessage = (e) => subscriber.next(JSON.parse(e.data) as T);
    source.onerror = () => {
      // browserul se reconectează singur (și trimite Last-Event-ID); doar CLOSED e definitiv
      if (source.readyState === EventSource.CLOSED) subscriber.error(new Error('SSE closed'));
    };
    return () => source.close();
  });
}

Când DA, când NU

WebSocket

Ambele părți vorbesc des: chat, prezență, editare colaborativă.
Latența contează, iar mesajele sunt mici și dese.
Ai nevoie de date binare sau de propriul protocol peste conexiune.

Altceva

Vorbește doar serverul: Server-Sent Events sunt mai simple. Cereri din când în când: HttpClient simplu.
Date care se schimbă la câteva minute: polling la 30–60 s e mai ieftin și mult mai simplu de întreținut.
Actualizări care trebuie să meargă offline și să se sincronizeze mai târziu: asta e o problemă offline-first, nu una de socket.

Capcane frecvente

Un socket per componentă. Cinci widget-uri, cinci conexiuni la același server. Ține conexiunea într-un serviciu root.
Retry fără jitter sau fără plafon. Mii de clienți revin în valuri sincronizate, sau un client așteaptă 17 minute după o pană lungă.
Doar retry(), fără repeat(). Închiderile curate (la fiecare deploy) termină stream-ul pentru totdeauna.
Încrederea în bufferul integrat. Mesajele trimise offline dispar după prima încercare eșuată. Folosește propriul outbox, limitat.
Fără heartbeat. Starea spune „conectat”, dar n-a mai venit nimic de zece minute.
Conectare în SSR, sau un JWT lung în URL. Primul deschide socket-uri de pe server; al doilea lasă token-uri în loguri.

Fișă rapidă

webSocket<T>(url | config)        subscribe = conectare · next() = trimitere · unsubscribe = închidere
înainte de primul „open”          next() e pus în buffer și trimis la conectare
...dar după o încercare eșuată    bufferul e aruncat → ține-ți propriul outbox
eroare de rețea / codul 1006      retry({ delay })
închidere curată (1000/1001/1012) repeat({ delay })   stream-ul SE ÎNCHEIE, nu dă eroare
backoff                           random(0, min(cap, base · 2^n))   „full jitter”
resetarea backoff-ului            în openObserver, nu la primul mesaj
moartă, dar neînchisă             ping la 15 s + timeout({ each: 20_000 })
SSR / prerender                   conectare în afterNextRender, niciodată în constructor
auth                              cookie · ticket scurt, de unică folosință · primul mesaj „auth”
după reconectare                  resync: cere ce ai pierdut (since=lastEventId)
mesaje care trebuie să ajungă     id + ack de la server, retrimite ce nu a fost confirmat
doar într-o direcție?             Server-Sent Events (EventSource): reconectarea e inclusă