WebSocket

Contract

@modularityjs/ws provides a decorator-based WebSocket gateway system with typed message handling, connection lifecycle hooks, and parameter injection.

Gateways

Gateways are classes that handle WebSocket connections on a specific path. The mounted URL is the configured WsConfig.path prefix (default /ws) concatenated with the gateway path — so @WsGateway('/chat') with the default config is served at /ws/chat, not /chat.

import {
  Client,
  Data,
  OnConnect,
  OnDisconnect,
  OnMessage,
  WsGateway,
  WsGatewaysPool,
} from '@modularityjs/ws';
import type { WsClient } from '@modularityjs/ws';

@WsGateway('/chat') // served at '/ws/chat' with default WsConfig.path
class ChatGateway {
  @OnConnect()
  handleConnect(@Client() client: WsClient) {
    console.log(`Client ${client.id} connected`);
  }

  @OnDisconnect()
  handleDisconnect(@Client() client: WsClient) {
    console.log(`Client ${client.id} disconnected`);
  }

  @OnMessage('chat:send')
  handleMessage(@Client() client: WsClient, @Data() data: { text: string }) {
    console.log(`Message from ${client.id}: ${data.text}`);
  }
}

Register gateways via the WsGatewaysPool — this is what decides which gateways are mounted:

@Module({
  name: 'chat',
  imports: [WsModule],
  providers: [ChatGateway],
  pools: [
    {
      pool: WsGatewaysPool,
      key: 'chat-gateway',
      useClass: ChatGateway,
    },
  ],
})
class ChatModule {}

Decorators

Class Decorators

Decorator	Description
@WsGateway(path)	Marks a class as a WebSocket gateway. The mounted URL is WsConfig.path + path (default prefix /ws, so /ws + path).

Method Decorators

Decorator	Description
@OnConnect()	Called when a client connects
@OnDisconnect()	Called when a client disconnects
@OnMessage(type)	Called when a message with the given type arrives

Parameter Decorators

Decorator	Description
@Client()	Injects the WsClient instance
@Data()	Injects the message data payload

WsClient

Each connected client exposes:

interface WsClient {
  readonly id: string;
  send(data: unknown): void;
  close(code?: number, reason?: string): void;
}

WsMessage

Messages exchanged over the WebSocket follow a typed envelope:

interface WsMessage {
  readonly type: string;
  readonly data: unknown;
}

WsServer

The abstract WsServer provides server-wide operations — broadcast, per-client send, rooms, and presence:

abstract class WsServer {
  abstract broadcast(type: string, data: unknown): void;
  abstract getClients(): ReadonlySet<WsClient>;
  abstract close(): Promise<void>;

  // per-client send
  abstract getClient(id: string): WsClient | undefined;
  abstract send(clientId: string, type: string, data: unknown): boolean;

  // rooms
  abstract join(clientId: string, room: string): void;
  abstract leave(clientId: string, room: string): void;
  abstract broadcastToRoom(room: string, type: string, data: unknown): void;

  // presence
  abstract getRoomMembers(room: string): ReadonlySet<WsClient>;
  abstract getRoomsOf(clientId: string): ReadonlySet<string>;
}

Inject WsServer to broadcast messages to all connected clients:

import { Inject, Injectable } from '@modularityjs/di';
import { WsServer } from '@modularityjs/ws';

@Injectable()
class NotificationService {
  constructor(@Inject(WsServer) private readonly ws: WsServer) {}

  notifyAll(message: string): void {
    this.ws.broadcast('notification', { message });
  }

  getOnlineCount(): number {
    return this.ws.getClients().size;
  }
}

Rooms and Presence

Clients join named rooms (typically from an @OnConnect or @OnMessage handler); broadcastToRoom reaches only the members:

@WsGateway('/chat')
class RoomsGateway {
  constructor(@Inject(WsServer) private readonly server: WsServer) {}

  @OnMessage('room:join')
  handleJoin(@Client() client: WsClient, @Data() data: { room: string }) {
    this.server.join(client.id, data.room);
    this.server.broadcastToRoom(data.room, 'room:joined', {
      clientId: client.id,
      members: this.server.getRoomMembers(data.room).size,
    });
  }

  @OnMessage('room:message')
  handleMessage(
    @Client() client: WsClient,
    @Data() data: { room: string; text: string },
  ) {
    this.server.broadcastToRoom(data.room, 'room:message', {
      from: client.id,
      text: data.text,
    });
  }
}

join/leave are no-ops for unknown client ids; room membership is cleaned up automatically on disconnect. getRoomMembers(room) and getRoomsOf(clientId) are the presence surface — member counts, "who's online" lists, per-user room enumeration.

Per-client send is local-only

send(clientId, type, data) returns false when no such client is connected to this instance — per-client sends never fan out across instances. For per-user messaging in a multi-instance deployment, route through a room per user (user:<id>): the user's sockets join it on connect, and broadcastToRoom('user:<id>', ...) reaches them wherever they're connected. Room membership is also per-instance — each instance tracks its own sockets; the fan-out transport carries the message, and every instance delivers to its local members.

Drivers

Fastify (@modularityjs/ws-fastify)

Integrates with the existing Fastify HTTP server via @fastify/websocket. Requires @modularityjs/http-fastify.

import { HttpModule } from '@modularityjs/http';
import { WsModule } from '@modularityjs/ws';
import { HttpFastifyModule } from '@modularityjs/http-fastify';
import { WsFastifyModule } from '@modularityjs/ws-fastify';

const modules = [
  HttpModule.forRoot({ port: 3000 }),
  HttpFastifyModule,
  WsModule,
  WsFastifyModule,
];

The driver registers the @fastify/websocket plugin on the Fastify instance, then mounts the gateways contributed to WsGatewaysPool, wiring each one's routes and message handlers. The pool is what decides: @WsGateway() marks a class and providers binds it in DI, but neither mounts anything on its own, so a gateway you never contribute serves nothing. On shutdown, all client connections are closed gracefully.

Multi-Instance Fan-Out (@modularityjs/ws-redis)

By default, broadcast and broadcastToRoom reach only clients connected to the current instance. The contract declares an optional WsFanoutTransport seam:

abstract class WsFanoutTransport {
  abstract publish(message: WsFanoutMessage): Promise<void>;
  abstract subscribe(
    handler: (message: WsFanoutMessage) => void,
  ): Promise<void>;
  abstract unsubscribe(): Promise<void>;
}

interface WsFanoutMessage {
  readonly room?: string; // absent = every client
  readonly type: string;
  readonly data: unknown;
}

The Fastify driver injects it with @InjectOptional — no transport wired means broadcasts stay process-local, with zero configuration. ws-redis binds a Redis pub/sub transport (one channel, <keyPrefix>ws:fanout, on the shared RedisModule client): every instance publishes its broadcasts and delivers incoming ones to its local clients, tagging messages with a per-instance sender id so its own publishes aren't double-delivered.

import { RedisModule } from '@modularityjs/redis';
import { WsModule } from '@modularityjs/ws';
import { WsFastifyModule } from '@modularityjs/ws-fastify';
import { WsRedisModule } from '@modularityjs/ws-redis';

const modules = [
  // ...HttpModule.forRoot({ port: 3000 }), HttpFastifyModule...
  RedisModule.forRoot({ url: 'redis://localhost:6379' }),
  WsModule,
  WsFastifyModule,
  WsRedisModule,
];

What fans out: broadcast and broadcastToRoom. What stays local: per-client send, join/leave, and the presence reads (getRoomMembers, getRoomsOf) — they reflect this instance's sockets only. Fan-out publish failures never break the local broadcast; they surface as ModularityJsWsFanoutError process warnings.

Configuration

WsModule.forRoot({
  allowedOrigins: ['https://app.example.com'], // required; ['*'] is the explicit allow-all opt-in
  fanout: 'local', // 'transport' when a WsFanoutTransport (e.g. ws-redis) is wired
  path: '/ws', // URL prefix prepended to every gateway path (default '/ws')
  maxClients: 1000, // hard cap on concurrent client sockets (default)
  maxMessageBytes: 65536, // hard cap on a single inbound frame; oversized frames close with 1009 (default)
  idleTimeoutMs: undefined, // close sockets idle longer than this; unset = no timeout
  connectTimeoutMs: 10000, // how long @OnConnect may run before the connection is refused (default)
});

path is a prefix, not a literal route. A gateway declared as @WsGateway('/chat') is served at path + '/chat' — so '/ws/chat' with the default, or '/chat' if path: ''.

maxMessageBytes is capped at 2147483647: the driver stores the limit as a 32-bit int, so a larger value wraps to 0, which means "no limit" — a config that reads as a tightened cap but disables it. Boot rejects it rather than accepting it.

The connect gate

@OnConnect is where a gateway authenticates, so nothing else runs until it settles:

• Frames are not dispatched during connect. The socket is paused, so the pressure lands on TCP instead of the heap, and queued frames dispatch in arrival order once the client is accepted.
• Refusing works both ways. Throwing closes the socket with 1011; the idiomatic client.close(1008, '…') followed by a normal return is equally a refusal — acceptance means the handler settled and the socket is still open. Either way the queued frames are dropped, never dispatched.
• A hung handler cannot hold a slot. connectTimeoutMs (default 10s) closes the socket with 1008 and warns ModularityJsWsConnectTimeout, freeing the maxClients slot.
• idleTimeoutMs starts after the gate. Arming it at open would reap a client whose authentication simply takes longer than the idle window; connectTimeoutMs is what bounds the connect phase. After acceptance, every arriving frame refreshes the idle timer.
• @OnDisconnect never precedes @OnConnect. A client that drops mid-authentication gets its disconnect callback after the connect handler settles — and gets none at all if the connection was refused, since there is no accepted state to tear down.

Usage Example

A chat gateway that broadcasts messages to all connected clients (served at /ws/chat with the default WsConfig.path = '/ws'):

@WsGateway('/chat')
class ChatGateway {
  constructor(@Inject(WsServer) private readonly server: WsServer) {}

  @OnConnect()
  handleConnect(@Client() client: WsClient) {
    this.server.broadcast('system', {
      text: `User ${client.id} joined`,
    });
  }

  @OnMessage('chat:send')
  handleMessage(@Client() client: WsClient, @Data() data: { text: string }) {
    this.server.broadcast('chat:message', {
      from: client.id,
      text: data.text,
    });
  }

  @OnDisconnect()
  handleDisconnect(@Client() client: WsClient) {
    this.server.broadcast('system', {
      text: `User ${client.id} left`,
    });
  }
}

Client-side messages use the { type, data } envelope:

{ "type": "chat:send", "data": { "text": "Hello, world!" } }