Server-WebSocket

Serverseitiges WebSocket ist Teil des HTTP-Route-DSL. Die websocket()-Direktive upgradet passende Requests und verbindet jede Verbindung mit einem einzelnen WebsocketServerActor — dem Hub — den du implementierst. Für ausgehende Client-Verbindungen siehe WebsocketClientActor.

import {
  ActorSystem, HttpExtensionId,
  WebsocketServerActor, websocket, type WebsocketConnection,
} from 'actor-ts';
import { match } from 'ts-pattern';

type SetNameMessage = { kind: 'setName'; name: string };
type SayMessage = { kind: 'say'; text: string };
type ClientMessage = SetNameMessage | SayMessage;
type ServerMessage = { kind: 'system'; text: string } | { kind: 'chat'; from: string; text: string };

class ChatRoom extends WebsocketServerActor<ServerMessage, ClientMessage> {
  private readonly names = new Map<string, string>();

  onMessage(message: ClientMessage): void {
    match(message)
      .with({ kind: 'setName' }, (m) => this.onSetName(m))
      .with({ kind: 'say' }, (m) => this.onSay(m))
      .exhaustive();
  }

  private onSetName(m: SetNameMessage): void {
    this.names.set(this.connection.id, m.name);
    this.reply({ kind: 'system', text: `hi ${m.name}` });
  }

  private onSay(m: SayMessage): void {
    this.broadcast({ kind: 'chat', from: this.names.get(this.connection.id) ?? 'anon', text: m.text });
  }

  override onClientDisconnected(c: WebsocketConnection<ServerMessage>): void {
    this.names.delete(c.id);
  }
}

const system = ActorSystem.create('chat');
const chat = system.spawn(ChatRoom, 'chat');

await system.extension(HttpExtensionId)
  .newServerAt('0.0.0.0', 8080)
  .bind(websocket('/ws', chat));

Die Typparameter des Hubs sind aus Sicht des Servers zu lesen: TOut = die Nachrichten, die der Server sendet, TIn = die dekodierten Nachrichten, die er empfängt.

Die websocket()-Direktive

websocket() erzeugt eine Route und komponiert daher mit dem restlichen Route-DSL — path(), concat() und withMiddleware(). Zwei Formen:

import { websocket, path, concat } from 'actor-ts';

// 1. Bare directive — mount it under a path yourself:
path('ws', websocket(chat));

// 2. Path sugar — equivalent to path(p, websocket(target)):
websocket('/ws', chat);

// Mixed with normal HTTP routes on the same server:
concat(
  path('api', apiRoutes),
  websocket('/ws', chat),
);

Optionen

type WebsocketRouteOptions<TOut, TIn> = {
  codec?:            WebsocketCodec<TOut, TIn>;             // default jsonCodec()
  maxFrameBytes?:    number;                          // default 1 MiB
  onOversizeFrame?:  'close' | 'drop';                // default 'close' (1009)
  onInvalidMessage?: 'close' | 'drop' | 'hook';       // default 'close' (1003)
  maxBufferedBytes?: number;                          // default 4 MiB
  onBackpressure?:   'drop' | 'close';                // default 'drop'
  allowedOrigins?:   string[];                        // CSWSH-Schutz — siehe unten
  maxConnections?:   number;                          // Limit gleichzeitiger Verbindungen; Default unbegrenzt
};

const webSocketRouteOptions = WebsocketRouteOptions.create()
  .withMaxFrameBytes(256 * 1024)
  .withOnOversizeFrame('close')       // close with 1009 Message Too Big
  .withOnInvalidMessage('close');
websocket('/ws', chat, webSocketRouteOptions);    // close with 1003 Unsupported Data

Das Frame-Größenlimit wird am rohen Frame vor dem Dekodieren durchgesetzt. onInvalidMessage: 'hook' leitet Dekodier-Fehler an den onInvalidMessage-Hook des Actors weiter, statt zu schließen.

Origin-Allowlist (CSWSH-Schutz)

Browser hängen die Cookies des Nutzers automatisch an einen WebSocket-Upgrade an — eine WS-Route mit ambienter Authentifizierung (Session-Cookie oder IpAllowlist) ist damit anfällig für Cross-Site-WebSocket-Hijacking: eine fremde Seite öffnet new WebSocket('wss://dein-host/ws') und nutzt die Credentials des Opfers.

Mit allowedOrigins wird der Handshake über den Origin-Header des Browsers abgesichert:

const wsOptions = WebsocketRouteOptions.create()
  .withAllowedOrigins(['https://app.example.com']);
websocket('/ws', chat, wsOptions);

Ein Upgrade, dessen Origin vorhanden, aber nicht gelistet ist, wird auf allen drei Backends vor dem Handshake mit 403 abgelehnt. Ein fehlender Origin (Nicht-Browser-Client — natives WebSocket, Server-zu-Server) wird zugelassen, da CSWSH ein reiner Browser-Angriff ist. Der Vergleich ist case-insensitiv.

BearerTokenAuth ist bereits geschützt (Browser können beim WS-Handshake keinen Authorization-Header setzen); allowedOrigins ist v. a. bei Cookie- oder IP-basierter Auth relevant.

Middleware läuft zur Upgrade-Zeit

withMiddleware() komponiert mit websocket(), aber die Middleware läuft einmal, zur Upgrade-Zeit, gegen den HTTP-Upgrade-Request — nicht pro WebSocket-Nachricht. Eine ablehnende Middleware gibt einen normalen HTTP-Fehler zurück und das Upgrade kommt nie zustande, sodass Auth-Middleware wie BearerTokenAuth oder IpAllowlist den Handshake absichert:

import { websocket, withMiddleware } from 'actor-ts';

withMiddleware(BearerTokenAuth({ /* ... */ }), websocket('/ws', chat));
// A bad token → HTTP 401, no upgrade.  A good token → the socket opens.

Die Middleware-Sammlung steht in der HTTP-Übersicht.

WebsocketServerActor

Ein Actor pro Route — der Hub. Er sieht die Events jeder Verbindung, serialisiert:

abstract class WebsocketServerActor<TOut, TIn, TSelf = never> {
  // You implement:
  abstract onMessage(message: TIn): void | Promise<void>;

  // Optional overrides:
  protected onClientConnected(client: WebsocketConnection<TOut>): void;
  protected onClientDisconnected(client: WebsocketConnection<TOut>, info: WebsocketCloseInfo): void;
  protected onInvalidMessage(client: WebsocketConnection<TOut>, error: WebsocketDecodeError): void;
  protected onSelfMessage(message: TSelf): void;
}

Helfer

Innerhalb von onMessage und den Hooks stehen zur Verfügung:

Member	Was er tut
this.connection	Die WebsocketConnection<TOut>, deren Event gerade verarbeitet wird.
this.reply(message)	Sendet message an die aktuelle Verbindung.
this.broadcast(message, filter?)	Sendet an jede Verbindung (optional per Prädikat gefiltert).
this.clients	ReadonlyMap<string, WebsocketConnection<TOut>> aller aktiven Verbindungen.
this.closeAll(code?, reason?)	Schließt jede Verbindung.

onMessage(message: ClientMessage): void {
  this.reply({ kind: 'system', text: 'got it' });         // → the sender
  this.broadcast({ kind: 'chat', from: 'x', text: 'hi' }); // → everyone
  this.broadcast(notice, (c) => c.id !== this.connection.id); // → everyone else
}

Reihenfolge pro Verbindung

Jedes Event einer gegebenen Verbindung wird über den einen Hub-Actor in dieser Reihenfolge serialisiert:

onClientConnected → onMessage* (in frame order) → onClientDisconnected

onClientConnected läuft einmal, dann null oder mehr onMessage-Aufrufe in Frame-Reihenfolge, dann genau ein onClientDisconnected. Da alles auf einem einzelnen Actor läuft, bekommst du die übliche Garantie des Actor-Modells — keine nebenläufige Handler-Ausführung, keine Locks.

`onReceive` ist versiegelt

Überschreibe onReceive an einem WebsocketServerActor nicht — es ist versiegelt. Für jede Verbindung spawnt das Framework einen internen Connection-Actor als Kind des Server-Actors (der Baum ist server → conn-1, conn-2, …); dieses Kind besitzt den Socket, dekodiert eingehende Frames, schreibt ausgehende und wird beim Schließen des Sockets automatisch gestoppt. Du siehst oder verwaltest ihn nie — implementiere stattdessen onMessage und die Hooks.

WebsocketConnection

Jede Verbindung ist eine WebsocketConnection<TOut>, die ActorRef<TOut> erweitert:

interface WebsocketConnection<TOut> extends ActorRef<TOut> {
  readonly id: string;
  readonly remoteAddress?: string;
  readonly upgrade: WebsocketUpgradeInfo;
  readonly isOpen: boolean;
  tell(message: TOut): void;         // encode via codec + send
  sendRaw(frame: WebsocketFrame): void; // bypass the codec
  close(code?: number, reason?: string): void;
}

upgrade trägt den Handshake-Kontext:

type WebsocketUpgradeInfo = {
  path: string;
  params: Record<string, string>;
  query: Record<string, string | string[] | undefined>;
  headers: Record<string, string>;
  remoteAddress?: string;
  subprotocol?: string;
};

WebsocketCloseInfo (an onClientDisconnected übergeben) ist { code: number; reason: string; initiatedBy: 'client' | 'server' | 'error' }.

Kein Deathwatch auf Verbindungen

WebsocketConnection ist eine Ref, aber du kannst sie nicht death-watchen. Um zu erfahren, wann ein Client verschwindet, überschreibe onClientDisconnected — es ist das einzige, geordnete End-of-Life-Signal pro Verbindung.

Codecs

Der Route-Codec dekodiert eingehende Frames in TIn und kodiert TOut-Antworten. Der Default ist jsonCodec():

import { jsonCodec, WebsocketRouteOptions } from 'actor-ts';

const webSocketRouteOptions = WebsocketRouteOptions.create().withCodec(jsonCodec<ServerMessage, ClientMessage>({
    validate: (v: unknown): ClientMessage => ClientMsgSchema.parse(v),
  }));
websocket('/ws', chat, webSocketRouteOptions);

Für Binärprotokolle liefert rawCodec() rohe Frames (TOut = TIn = WebsocketFrame):

import { match } from 'ts-pattern';
import { rawCodec, WebsocketRouteOptions, WebsocketServerActor, type WebsocketFrame } from 'actor-ts';

const webSocketRouteOptions = WebsocketRouteOptions.create().withCodec(rawCodec());
class BinaryHub extends WebsocketServerActor<WebsocketFrame, WebsocketFrame> {
  onMessage(frame: WebsocketFrame): void {
    match(frame)
      .with({ kind: 'binary' }, (f) => this.onBinary(f))
      .otherwise(() => {});
  }

  private onBinary(frame: BinaryFrame): void {
    this.reply({ kind: 'binary', data: process(frame.data) });
  }
}

websocket('/stream', binaryHub, webSocketRouteOptions);

Dekodier-Fehler werfen WebsocketDecodeError und folgen der onInvalidMessage-Policy der Route.

Backends und Runtimes

websocket() funktioniert auf allen drei HTTP-Backends:

Backend	Peer-Dependency	Runtime-Hinweise
Fastify (Default)	@fastify/websocket	Läuft auf Bun und Node; auf Deno bevorzuge Hono.
Express	ws	—
Hono	Bun & Deno: in hono eingebaut. Hono-auf-Node: @hono/node-ws.	Helfer pro Runtime.

npm install @fastify/websocket   # Fastify (default)
npm install ws                   # Express
npm install @hono/node-ws        # Hono on Node only

HOCON-Config

Route-Defaults liegen unter actor-ts.http.websocket:

actor-ts.http.websocket {
  maxFrameBytes    = 1048576
  onOversizeFrame  = close
  onInvalidMessage = close
  maxBufferedBytes = 4194304
  onBackpressure   = drop
}

Präzedenz: Route-Optionen > HOCON > eingebaute Defaults.

Wann einsetzen

Drei gute Einsatzfälle:

1. Echtzeit-UIs — Updates an Browser-Clients pushen.
2. Eigene Messaging-Protokolle — Game-Server, Chat-Backends.
3. Binäres Streaming — nutze rawCodec() und behandle Frames direkt.

Für einseitige Streams vom Server zum Client ist SSE einfacher. Für Request/Reply-RPC ist einfaches HTTP die Norm.

Umbenannt von ServerWebsocketActor

Dies ersetzt den alten ServerWebsocketActor und seine serverWebsocketActorOf- / bunWebsocketHandlers-Helfer. Verbindungen sind keine per-Verbindung-Child-Actors mehr, die du spawnst — der Hub sieht jede Verbindung über einen Actor, und das Framework verwaltet einen internen Session-Actor pro Socket.

Backpressure bei langsamen Clients

const routeOptions = WebsocketRouteOptions.create()
  .withMaxBufferedBytes(4 * 1024 * 1024)
  .withOnBackpressure('drop');
websocket('/ws', chat, routeOptions);

Wenn ein Client nicht mithält, wächst sein Sende-Buffer. Ab maxBufferedBytes verwirft die Verbindung entweder Frames (onBackpressure: 'drop', Default) oder wird geschlossen ('close'). Broadcasts an viele langsame Clients sind der Ort, wo das zubeißt — wähle die Policy, die zu deinem Protokoll passt.

Wohin als Nächstes

• I/O-Übersicht — das große Bild.
• WebsocketClientActor — die ausgehende Client-Hälfte.
• Route-DSL — path, concat und wie websocket() komponiert.
• SSE — Server-Sent Events, die einseitige Alternative.
• HTTP-Übersicht — Backends, Middleware und Server-Binding.