WebSockets : un temps réel qui résiste au réseau

import { webSocket } from 'rxjs/webSocket';

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

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

// subscribe() ouvre la connexion ; chaque message arrive déjà parsé (JSON.parse)
const sub = socket.subscribe({
  next: (msg) => console.log(msg.user, msg.text),
  error: (err) => console.error('connexion perdue', err), // le flux SE TERMINE ici
  complete: () => console.log('fermée proprement par le serveur'), // ...ou ici
});

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

sub.unsubscribe(); // ferme la connexion

/**
 * « Full jitter » : une attente aléatoire entre 0 et min(cap, base · 2^attempt).
 * tentative 1 → 0–1 s, 2 → 0–2 s, 3 → 0–4 s ... jamais plus de 10 s.
 */
export function fullJitter(attempt: number, base = 500, cap = 10_000): number {
  return Math.random() * Math.min(cap, base * 2 ** attempt);
}

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) }, // remise à zéro quand la ligne est vraiment ouverte
});

socket
  .pipe(
    retry({ delay: wait }), // erreur réseau, fermeture anormale (1006)
    repeat({ delay: wait }), // fermeture PROPRE (1001, 1012 « service restart ») : le flux se termine !
  )
  .subscribe((msg) => console.log(msg));

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; // aucun message pendant 20 s = connexion morte

/** Une seule connexion pour toute l’application, exposée sous forme de 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 qui a été envoyé pendant la coupure ; part, dans l’ordre, au prochain « open ». */
  readonly outbox = signal<readonly ClientMessage[]>([]);

  private attempt = 0;
  // RxJS utilise un seul type pour les deux sens
  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 : silence → erreur → 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)); // borné !
  }

  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));
  }
}

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';

// Le navigateur ne peut pas mettre d’en-tête Authorization sur un WebSocket.
// 1) un ticket court via HTTP (votre intercepteur d’auth ajoute le token comme d’habitude)
// 2) le ticket est à usage unique et vit ~30 s, il peut donc aller dans l’URL
// 3) defer() → chaque reconnexion demande un NOUVEAU ticket
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)) }),
  );
}

import { Observable } from 'rxjs';

/** Server-Sent Events : serveur → navigateur uniquement, HTTP simple, reconnexion intégrée. */
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 = () => {
      // le navigateur se reconnecte seul (et envoie Last-Event-ID) ; seul CLOSED est définitif
      if (source.readyState === EventSource.CLOSED) subscriber.error(new Error('SSE closed'));
    };
    return () => source.close();
  });
}

webSocket<T>(url | config)        subscribe = connexion · next() = envoi · unsubscribe = fermeture
avant le premier « open »         next() est mis en tampon et envoyé à la connexion
...mais après un échec            ce tampon est jeté → gardez votre propre file d’envoi
erreur réseau / code 1006         retry({ delay })
fermeture propre (1000/1001/1012) repeat({ delay })   le flux SE TERMINE, sans erreur
backoff                           random(0, min(cap, base · 2^n))   « full jitter »
remettre le backoff à zéro        dans openObserver, pas au premier message
morte mais pas fermée             ping toutes les 15 s + timeout({ each: 20_000 })
SSR / prerender                   connexion dans afterNextRender, jamais dans un constructeur
auth                              cookie · ticket court à usage unique · premier message « auth »
après reconnexion                 resync : demander ce qu’on a manqué (since=lastEventId)
messages qui doivent arriver      id + ack du serveur, renvoyer ce qui n’est pas acquitté
un seul sens ?                    Server-Sent Events (EventSource) : la reconnexion est intégrée