WebSockets: real time that survives the network

import { webSocket } from 'rxjs/webSocket';

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

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

// subscribe() opens the connection; every message arrives already parsed (JSON.parse)
const sub = socket.subscribe({
  next: (msg) => console.log(msg.user, msg.text),
  error: (err) => console.error('connection lost', err), // the stream ENDS here
  complete: () => console.log('closed cleanly by the server'), // ...or here
});

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

sub.unsubscribe(); // closes the connection

/**
 * "Full jitter": a random wait between 0 and min(cap, base · 2^attempt).
 * attempt 1 → 0–1 s, 2 → 0–2 s, 3 → 0–4 s ... never more than 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) }, // reset when the line is really open
});

socket
  .pipe(
    retry({ delay: wait }), // network error, abnormal close (1006)
    repeat({ delay: wait }), // CLEAN close (1001, 1012 "service restart"): the stream completes!
  )
  .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; // no message at all for 20 s = a dead connection

/** One connection for the whole app, exposed as signals. */
@Service()
export class LiveSocket {
  private readonly destroyRef = inject(DestroyRef);

  readonly status = signal<SocketStatus>('idle');
  readonly online = signal(0);
  readonly messages = signal<readonly ServerMessage[]>([]);
  /** What was sent while the line was down; it leaves, in order, on the next "open". */
  readonly outbox = signal<readonly ClientMessage[]>([]);

  private attempt = 0;
  // RxJS uses a single type for both directions
  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 → error → 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)); // bounded!
  }

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

// The browser cannot set an Authorization header on a WebSocket.
// 1) a short ticket over HTTP (your auth interceptor adds the token as usual)
// 2) the ticket is single-use and lives ~30 s, so it is fine in the URL
// 3) defer() → every reconnect asks for a NEW 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: server → browser only, plain HTTP, reconnect built in. */
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 = () => {
      // the browser reconnects by itself (and sends Last-Event-ID); only CLOSED is final
      if (source.readyState === EventSource.CLOSED) subscriber.error(new Error('SSE closed'));
    };
    return () => source.close();
  });
}

webSocket<T>(url | config)      subscribe = connect · next() = send · unsubscribe = close
before the first "open"         next() is buffered and sent on connect
...but after a failed attempt   that buffer is thrown away → keep your own outbox
network error / code 1006       retry({ delay })
clean close (1000/1001/1012)    repeat({ delay })   the stream COMPLETES, it does not error
backoff                         random(0, min(cap, base · 2^n))   "full jitter"
reset the backoff               in openObserver, not on the first message
dead but not closed             ping every 15 s + timeout({ each: 20_000 })
SSR / prerender                 connect in afterNextRender, never in a constructor
auth                            cookie · short single-use ticket · first "auth" message
after reconnect                 resync: ask for what you missed (since=lastEventId)
must-arrive messages            id + ack from the server, resend what was not acked
one direction only?             Server-Sent Events (EventSource): reconnect is built in