WebSockets: Echtzeit, die das Netz übersteht

import { webSocket } from 'rxjs/webSocket';

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

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

// subscribe() öffnet die Verbindung; jede Nachricht kommt schon geparst an (JSON.parse)
const sub = socket.subscribe({
  next: (msg) => console.log(msg.user, msg.text),
  error: (err) => console.error('Verbindung verloren', err), // der Stream ENDET hier
  complete: () => console.log('vom Server sauber geschlossen'), // ...oder hier
});

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

sub.unsubscribe(); // schließt die Verbindung

/**
 * „Full Jitter“: eine zufällige Wartezeit zwischen 0 und min(cap, base · 2^attempt).
 * Versuch 1 → 0–1 s, 2 → 0–2 s, 3 → 0–4 s ... nie mehr als 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) }, // zurücksetzen, wenn die Leitung wirklich offen ist
});

socket
  .pipe(
    retry({ delay: wait }), // Netzwerkfehler, abnormales Schließen (1006)
    repeat({ delay: wait }), // SAUBERES Schließen (1001, 1012 „service restart“): der Stream endet regulär!
  )
  .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; // 20 s lang gar keine Nachricht = tote Verbindung

/** Eine einzige Verbindung für die ganze App, als Signals bereitgestellt. */
@Service()
export class LiveSocket {
  private readonly destroyRef = inject(DestroyRef);

  readonly status = signal<SocketStatus>('idle');
  readonly online = signal(0);
  readonly messages = signal<readonly ServerMessage[]>([]);
  /** Was gesendet wurde, während die Leitung weg war; es geht beim nächsten „open“ der Reihe nach raus. */
  readonly outbox = signal<readonly ClientMessage[]>([]);

  private attempt = 0;
  // RxJS nutzt einen einzigen Typ für beide Richtungen
  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: Stille → Fehler → 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)); // begrenzt!
  }

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

// Der Browser kann bei einem WebSocket keinen Authorization-Header setzen.
// 1) ein kurzes Ticket per HTTP (dein Auth-Interceptor fügt das Token wie gewohnt hinzu)
// 2) das Ticket ist einmalig und lebt ~30 s, darf also in die URL
// 3) defer() → jeder Reconnect holt ein NEUES 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: nur Server → Browser, schlichtes HTTP, Reconnect eingebaut. */
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 = () => {
      // der Browser verbindet sich selbst neu (und sendet Last-Event-ID); nur CLOSED ist endgültig
      if (source.readyState === EventSource.CLOSED) subscriber.error(new Error('SSE closed'));
    };
    return () => source.close();
  });
}

webSocket<T>(url | config)        subscribe = verbinden · next() = senden · unsubscribe = schließen
vor dem ersten „open“             next() wird gepuffert und beim Verbinden gesendet
...aber nach einem Fehlversuch    wird der Puffer verworfen → eigenen Postausgang führen
Netzwerkfehler / Code 1006        retry({ delay })
sauberes Schließen (1000/1001/1012) repeat({ delay })   der Stream ENDET regulär, ohne Fehler
Backoff                           random(0, min(cap, base · 2^n))   „Full Jitter“
Backoff zurücksetzen              in openObserver, nicht bei der ersten Nachricht
tot, aber nicht geschlossen       Ping alle 15 s + timeout({ each: 20_000 })
SSR / Prerender                   in afterNextRender verbinden, nie im Konstruktor
Auth                              Cookie · kurzes Einmal-Ticket · erste „auth“-Nachricht
nach dem Reconnect                Resync: das Verpasste anfordern (since=lastEventId)
Nachrichten, die ankommen müssen  ID + Ack vom Server, Unbestätigtes erneut senden
nur eine Richtung?                Server-Sent Events (EventSource): Reconnect ist eingebaut