@vutotoite/websocket

Thư viện WebSocket viết bằng TypeScript, hỗ trợ Server và Client (Node.js + trình duyệt), scale theo chiều ngang qua NATS.

📦 Cài đặt

npm install @vutotoite/websocket

# Tùy chọn: chỉ cần khi muốn scale ngang qua NATS
npm install @nats-io/transport-node

nats là optional peer dependency. Nếu chỉ chạy 1 instance (MemoryAdapter), bạn không cần cài.

✨ Tính năng

Server

• ✅ Singleton pattern, tích hợp HTTP server
• ✅ Xác thực (Authentication) tùy chỉnh
• ✅ Quản lý phòng (Room) và client
• ✅ Middleware chain với throw error (dừng ngay khi lỗi)
• ✅ Ping/Pong tự động
• ✅ Adapter pattern: MemoryAdapter (1 node) hoặc NatsAdapter (nhiều node)
• ✅ Scale ngang qua NATS — broadcast/room message hoạt động xuyên node
• ✅ Codec tùy chỉnh (mặc định JSON)

Client

• ✅ Hỗ trợ Node.js và Browser, dùng chung logic qua BaseClient
• ✅ API on/emit đơn giản
• ✅ Tự động kết nối lại, queue tin nhắn khi offline
• ✅ Heartbeat tự động (Node)

🏗️ Kiến trúc & Scale ngang

                    ┌─────────── NATS ───────────┐
                    │      (message bus chung)     │
                    └──────┬───────────────┬───────┘
                           │               │
                  ┌────────▼──────┐ ┌──────▼────────┐
                  │  Node A       │ │  Node B       │
                  │  WsServer     │ │  WsServer     │
                  │  +NatsAdapter │ │  +NatsAdapter │
                  └────┬─────┬────┘ └────┬─────┬────┘
                    client client     client client

Mỗi node giữ kết nối WebSocket của riêng nó. Khi toRooms/toAll/toClients phát message, NatsAdapter publish lên subject NATS phù hợp; node nào quan tâm sẽ nhận và deliver tới client local của mình. Nhờ đó broadcast hoạt động xuyên nhiều instance.

• MemoryAdapter (mặc định): state trong RAM, dùng cho 1 instance.
• NatsAdapter: nhận NatsConnection truyền từ ngoài vào (dependency injection), định tuyến qua core NATS pub/sub.

Định tuyến theo subject

NatsAdapter định tuyến theo loại target để giảm traffic xuyên node (subject mặc định prefix ws):

| Target | Subject publish | Subscription của node | |--------|----------------|----------------------| | toAll() | ws.broadcast | luôn subscribe ws.broadcast | | toClients(id) | ws.client.<id> | wildcard ws.client.*, lọc client local | | toRooms(id) | ws.room.<id> | chỉ subscribe room khi có client local; tự unsubscribe khi client cuối rời |

Điểm quan trọng: node không có client trong một room sẽ không nhận traffic của room đó. Subscription room được tạo/hủy tự động theo join/leave.

Về toClients xuyên node: không có presence registry (clientId → nodeId), nên message client được publish và mọi node nhận qua wildcard rồi lọc local. Đây là tradeoff của thiết kế core NATS thuần. Nếu cần định tuyến client chính xác (chỉ node chứa client nhận), có thể bổ sung presence registry sau.

🚀 Sử dụng

1. Server cơ bản (1 instance — MemoryAdapter)

import { Server } from '@vutotoite/websocket';
import http from 'http';
import express from 'express';

const app = express();
const httpServer = http.createServer(app);

// init() là singleton: lần gọi đầu tạo instance, các lần sau trả lại đúng instance đó.
// noServer: true để tự kiểm soát quá trình HTTP upgrade qua attachServer().
const wsServer = Server.WebsocketServer.init({
  ws: { noServer: true }
  // adapter mặc định = MemoryAdapter (state trong RAM, 1 instance)
});

// Gắn vào HTTP server: lib lắng nghe sự kiện 'upgrade' để bắt tay WebSocket.
wsServer.attachServer(httpServer);

wsServer.connected({
  // connectionHandler chạy MỖI khi có client mới kết nối thành công.
  connectionHandler: (ws, wss) => {
    console.log('Client đã kết nối');

    // onS đăng ký listener cho event 'message' gửi từ client ({ event, data }).
    ws.onS('message', (data) => {
      console.log('Nhận:', data);
      // emitS gửi event về CHÍNH client này.
      ws.emitS('response', { status: 'ok' });
    });
  },
  // errorHandler nhận mọi lỗi: parse message hỏng, hoặc throw trong middleware chain.
  errorHandler: (error, ws) => {
    ws.emitS('error', { message: error.message });
  },
  // closeHandler chạy khi client ngắt kết nối (đã được dọn khỏi room/adapter).
  closeHandler: (ws) => {
    console.log('Client ngắt kết nối');
  }
});

// Lưu ý: vòng đời http.Server do BẠN quản lý — lib không tự listen/close nó.
httpServer.listen(8080, () => console.log('Server chạy trên cổng 8080'));

2. Scale ngang với NATS

Connection NATS được tạo bên ngoài và truyền vào NatsAdapter:

import { Server } from '@vutotoite/websocket';
import { connect } from '@nats-io/transport-node'; // hoặc 'nats'
import http from 'http';

// Bạn TỰ tạo kết nối NATS và quản lý vòng đời của nó (lib không tự connect).
const nats = await connect({ servers: 'nats://localhost:4222' });

const wsServer = Server.WebsocketServer.init({
  ws: { noServer: true },
  // Inject connection vào NatsAdapter -> mọi broadcast định tuyến qua NATS.
  adapter: new Server.NatsAdapter(nats)
});

wsServer.attachServer(http.createServer().listen(8080));

wsServer.connected({
  connectionHandler: (ws, wss) => {
    ws.onS('send_message', (data: { room_id: string; message: string }) => {
      // wss.toRooms(...).emitS(...): NatsAdapter publish lên subject ws.room.<id>.
      // Mọi node CÓ client trong room đó sẽ nhận và deliver tới client local.
      // -> client ở MỌI node (kể cả node khác) đều nhận được new_message.
      wss.toRooms(data.room_id).emitS('new_message', {
        from: ws.getId(),
        message: data.message
      });
    });
  },
  errorHandler: (error, ws) => ws.emitS('error', { message: error.message })
});

Chạy nhiều instance (mỗi instance 1 cổng / 1 container) cùng trỏ về NATS → tự động đồng bộ.

Lưu ý vận hành: dùng load balancer với sticky session (hoặc bất kỳ LB nào, vì state được đồng bộ qua NATS) để client giữ kết nối ổn định. NatsAdapter hiện dùng core pub/sub (không lưu trữ message), phù hợp real-time.

Graceful shutdown

close() đóng có trật tự: ngừng nhận kết nối mới, đóng client local với close code 1001 ("going away"), dọn subscription NATS của node (không ảnh hưởng node khác), rồi chờ ws server đóng hẳn. Không đóng http.Server của bạn — vòng đời đó do bạn quản lý.

// Tự gọi close() khi nhận SIGTERM/SIGINT (K8s, docker stop)
wsServer.enableGracefulShutdown(); // mặc định ['SIGTERM', 'SIGINT']

// Hoặc tự quản lý
process.on('SIGTERM', async () => {
  await wsServer.close();
  httpServer.close(); // đóng http server của bạn nếu cần
  process.exit(0);
});

3. Tổ chức trong NestJS

Mẫu khuyến nghị: bọc thư viện trong một module + service. Connection NATS được inject qua provider; server WebSocket được khởi tạo trong onModuleInit và gắn vào HTTP server của Nest; handler kết nối đăng ký trong một gateway service.

// nats.provider.ts
// Provider tạo & quản lý vòng đời kết nối NATS (lib KHÔNG tự connect).
import { connect, NatsConnection } from 'nats';
import { Provider } from '@nestjs/common';

export const NATS_CONNECTION = 'NATS_CONNECTION';

export const NatsProvider: Provider = {
  provide: NATS_CONNECTION,
  // useFactory chạy 1 lần khi module khởi tạo -> trả về connection dùng chung.
  useFactory: async (): Promise<NatsConnection> =>
    connect({ servers: process.env.NATS_URL ?? 'nats://localhost:4222' }),
};

// websocket.service.ts
// Service bọc WebsocketServer: khởi tạo singleton, cấu hình auth + handler.
import { Inject, Injectable, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { HttpAdapterHost } from '@nestjs/core';
import { NatsConnection } from 'nats';
import { Server } from '@vutotoite/websocket';
import { NATS_CONNECTION } from './nats.provider';

@Injectable()
export class WebsocketService implements OnModuleInit, OnModuleDestroy {
  private wsServer!: Server.WebsocketServer;

  constructor(
    @Inject(NATS_CONNECTION) private readonly nats: NatsConnection,
    // HttpAdapterHost cho phép lấy http.Server thật mà Nest đang chạy.
    private readonly httpAdapterHost: HttpAdapterHost,
  ) {}

  onModuleInit() {
    // init() là singleton: chỉ tạo instance ở lần gọi đầu tiên.
    this.wsServer = Server.WebsocketServer.init({
      ws: { noServer: true },                       // tự kiểm soát HTTP upgrade
      adapter: new Server.NatsAdapter(this.nats),   // inject connection -> scale ngang
    });

    // Lấy http.Server thật của Nest và gắn vào -> lib bắt sự kiện 'upgrade'.
    const httpServer = this.httpAdapterHost.httpAdapter.getHttpServer();
    this.wsServer.attachServer(httpServer);

    // Xác thực: throw -> server trả 401 và từ chối kết nối.
    this.wsServer.setAuth(async (req) => {
      const token = req.headers.authorization?.split(' ')[1];
      if (!token) throw new Error('Unauthorized');
      return await this.verifyToken(token); // -> ws.getAuthData()
    });

    // Tự gọi close() khi nhận SIGTERM/SIGINT (graceful shutdown trong K8s).
    this.wsServer.enableGracefulShutdown();
  }

  // Dọn dẹp khi module bị huỷ (vd hot-reload): đóng có trật tự + drain NATS.
  async onModuleDestroy() {
    await this.wsServer.close();
    await this.nats.drain();
  }

  // Cho phép service khác lấy server để emit (vd từ REST controller).
  getServer(): Server.WebsocketServer {
    return this.wsServer;
  }

  private async verifyToken(token: string): Promise<{ userId: string }> {
    // ... xác thực token, trả về payload
    return { userId: 'demo' };
  }
}

// chat.gateway.ts
// Đăng ký handler kết nối/sự kiện. Tách khỏi service hạ tầng cho gọn.
import { Injectable, OnModuleInit } from '@nestjs/common';
import { WebsocketService } from './websocket.service';

@Injectable()
export class ChatGateway implements OnModuleInit {
  constructor(private readonly ws: WebsocketService) {}

  onModuleInit() {
    this.ws.getServer().connected({
      connectionHandler: (socket, server) => {
        // authData lấy từ setAuth ở trên; gán id + cho vào room riêng của user.
        const auth = socket.getAuthData<{ userId: string }>();
        socket.setId(auth.userId);
        socket.join(auth.userId);

        // Lắng nghe client gửi 'send_message' -> phát tới room (xuyên node qua NATS).
        socket.onS('send_message', (data: { room_id: string; message: string }) => {
          server.toRooms(data.room_id).emitS('new_message', {
            from: socket.getId(),
            message: data.message,
          });
        });
      },
      errorHandler: (error, socket) => socket.emitS('error', { message: error.message }),
    });
  }
}

// websocket.module.ts
import { Module } from '@nestjs/common';
import { NatsProvider } from './nats.provider';
import { WebsocketService } from './websocket.service';
import { ChatGateway } from './chat.gateway';

@Module({
  providers: [NatsProvider, WebsocketService, ChatGateway],
  exports: [WebsocketService], // export để controller/service khác emit được
})
export class WebsocketModule {}

// notification.controller.ts
// Emit từ REST API: tái dùng WebsocketService -> đẩy event tới client cụ thể.
import { Body, Controller, Post } from '@nestjs/common';
import { WebsocketService } from './websocket.service';

@Controller('notify')
export class NotificationController {
  constructor(private readonly ws: WebsocketService) {}

  @Post()
  notify(@Body() body: { userId: string; message: string }) {
    // toClients(userId): tới client ở bất kỳ node nào (định tuyến qua NATS).
    this.ws.getServer().toClients(body.userId).emitS('notification', {
      message: body.message,
    });
    return { success: true };
  }
}

Thứ tự lifecycle: NatsProvider.useFactory (tạo connection) → WebsocketService.onModuleInit (init + attach + auth) → ChatGateway.onModuleInit (đăng ký handler). Vì init() là singleton, mọi service đều thao tác trên cùng một server instance.

4. Xác thực

wsServer.setAuth(async (req, query) => {
  const token = req.headers.authorization?.split(' ')[1];
  if (!token) throw new Error('Unauthorized');

  const user = await verifyToken(token);
  return user; // được truyền vào ws.getAuthData()
});

Throw error trong setAuth → server trả 401 và từ chối kết nối.

5. onInstanceInit (NestJS / module isolation)

// websocket-config.ts
import { Server } from '@vutotoite/websocket';

Server.WebsocketServer.onInstanceInit((wsServer) => {
  wsServer.setAuth(async (req, query) => {
    const token = req.headers.authorization?.split(' ')[1];
    if (!token) throw new Error('Unauthorized');
    return await verifyToken(token);
  });

  wsServer.connected({
    connectionHandler: (ws) => {
      const auth = ws.getAuthData();
      ws.setId(auth.userId);
      ws.join(auth.userId);
    },
    errorHandler: (error, ws) => ws.emitS('error', { message: error.message })
  });
});

// server.ts
import './websocket-config';
import { Server } from '@vutotoite/websocket';

const wsServer = Server.WebsocketServer.init({ ws: { noServer: true } });
wsServer.attachServer(httpServer); // callback đã đăng ký chạy ngay khi init

• Server đã init → callback chạy ngay lập tức.
• Chưa init → callback chạy sau khi init() được gọi.
• Callback chỉ chạy một lần.

6. Query parameters

wsServer.connected({
  connectionHandler: (ws) => {
    // ws://localhost:8080?user_id=123&room=general
    const query = ws.getQuery<{ user_id: string; room: string }>();
    ws.setId(query.user_id);
    ws.join(query.room);
  },
  errorHandler: (error, ws) => ws.emitS('error', { message: error.message })
});

7. Middleware chain với throw error

Middleware chạy tuần tự trái → phải. Khi một middleware throw, chain dừng ngay và errorHandler được gọi.

class WsError extends Error {
  constructor(message: string, public code = 400) {
    super(message);
    this.name = 'WsError';
  }
}

const validateRoom = (data: { room_id?: string }) => {
  if (!data?.room_id) throw new WsError('Room ID bắt buộc', 400);
  if (!wsServer.isExistRoom(data.room_id)) throw new WsError('Phòng không tồn tại', 404);
};

const validateMessage = (data: { message?: string }) => {
  if (!data?.message?.trim()) throw new WsError('Tin nhắn rỗng', 400);
  if (data.message.length > 1000) throw new WsError('Tin nhắn quá dài', 400);
};

wsServer.connected({
  connectionHandler: (ws, wss) => {
    ws.onS(
      'send_message',
      validateRoom,        // dừng chain nếu throw
      validateMessage,
      (data: { room_id: string; message: string }) => {
        wss.toRooms(data.room_id).emitS('new_message', {
          message: data.message,
          from: ws.getId()
        });
      }
    );
  },
  errorHandler: (error, ws) => {
    if (error instanceof WsError) {
      ws.emitS('validation_error', { message: error.message, code: error.code });
    } else {
      console.error('Unexpected:', error);
      ws.emitS('server_error', { message: 'Lỗi không mong muốn' });
    }
  }
});

Mỗi middleware nhận (data, prevResult) — prevResult là giá trị trả về của middleware trước, cho phép truyền dữ liệu qua chain.

8. Quản lý phòng

wsServer.connected({
  connectionHandler: (ws, wss) => {
    ws.onS('join_room', (data: { room_id: string }) => {
      ws.join(data.room_id);
      ws.emitS('joined', { room: data.room_id });
    });

    ws.onS('leave_room', (data: { room_id: string }) => {
      ws.leave(data.room_id);
      ws.emitS('left', { room: data.room_id });
    });
  },
  errorHandler: (error, ws) => ws.emitS('error', { message: error.message })
});

9. Gửi message tới target

wsServer.toAll().emitS('broadcast', { message: 'Hello everyone!' });
wsServer.toClients('user_123', 'user_456').emitS('private', { message: 'Hi' });
wsServer.toRooms('room_1', 'room_2').emitS('announcement', { message: '...' });

// Lọc client LOCAL trên node hiện tại (không xuyên node)
wsServer.filter((c) => c.getVariable('premium') === true)
        .emitS('premium_offer', { discount: 50 });

filter() chỉ duyệt client kết nối tại node hiện tại. Trong môi trường nhiều node, dùng room hoặc client id để định tuyến chính xác xuyên node.

10. Biến tùy chỉnh

ws.setVariable('username', 'John');
ws.setVariable('premium', true);

const username = ws.getVariable<string>('username');

11. Tích hợp Express API

app.post('/api/notify', (req, res) => {
  const { userId, message } = req.body;
  wsServer.toClients(userId).emitS('notification', { message });
  res.json({ success: true });
});

💻 WebSocket Client

Browser

import { Client } from '@vutotoite/websocket';

const ws = Client.WebsocketBrowser.getInstance();

ws.connect('ws://localhost:8080?user_id=123', 3000);

ws.on('__connection_open', () => console.log('Đã kết nối'));
ws.on('__connection_close', (d) => console.log('Mất kết nối', d));
ws.on('__connection_error', (e) => console.error('Lỗi', e));

ws.on('new_message', (data) => console.log('Tin nhắn:', data));

ws.emit('send_message', { room_id: 'general', message: 'Hello!' });

ws.disconnect();

Node.js

import { Client } from '@vutotoite/websocket';

const ws = Client.WebsocketNode.getInstance();

ws.connect('ws://localhost:8080?user_id=456', 3000, {
  headers: { Authorization: 'Bearer token' }
});

ws.on('message', (data) => console.log(data));
ws.emit('chat', { room: 'tech', message: 'Hi from Node' });

ws.setupHeartbeat(30000); // ping mỗi 30s
ws.disconnect(false, 1000, 'Client closing');

Event handlers nâng cao

ws.once('welcome', (data) => console.log(data));

const handler = (d) => console.log(d);
ws.on('message', handler);
ws.off('message', handler);   // hủy 1 handler
ws.off('message');            // hủy tất cả handler của event
ws.clearAllHandlers();
ws.getRegisteredEvents();

Message queue

ws.setMaxQueueSize(200);
ws.getQueueSize();
ws.isConnected();
ws.getReadyState(); // 0 CONNECTING, 1 OPEN, 2 CLOSING, 3 CLOSED

🔧 Codec tùy chỉnh

Mặc định dùng JSON { event, data }. Có thể thay bằng codec riêng:

import { Server } from '@vutotoite/websocket';

const myCodec: Server.Codec = {
  encode: (msg) => JSON.stringify([msg.event, msg.data]),
  decode: (raw) => {
    const [event, data] = JSON.parse(raw);
    return { event, data };
  }
};

Server.WebsocketServer.init({ ws: { noServer: true }, codec: myCodec });

📖 API Reference

WebsocketServer

| Method | Tham số | Mô tả | |--------|---------|-------| | init() | opts: ServerInitOptions | Khởi tạo (singleton) | | getInstance() | - | Lấy instance | | onInstanceInit() | cb: (server) => void | Callback chạy sau init | | attachServer() | httpServer: http.Server | Gắn vào HTTP server | | setAuth() | auth: (req, query) => any | Xác thực; throw để từ chối | | connected() | options: ConnectedOptions | Đăng ký handler kết nối | | toClients() | ...ids: string[] | Target theo client id (xuyên node) | | toRooms() | ...ids: string[] | Target theo room (xuyên node) | | toAll() | - | Target tất cả (xuyên node) | | filter() | cb: (conn) => boolean | Lọc client local | | isExistRoom() | roomId: string | Kiểm tra room (local) | | close() | - | Đóng có trật tự: client 1001, dọn adapter, chờ ws đóng | | enableGracefulShutdown() | signals?: NodeJS.Signals[] | Hook SIGTERM/SIGINT → tự close(); trả hàm hủy |

ServerInitOptions: { ws: ws.ServerOptions; adapter?: Adapter; codec?: Codec }

SocketConnection

| Method | Mô tả | |--------|-------| | setId(id) / getId() | Đặt / lấy id client | | onS(event, ...callbacks) | Lắng nghe với middleware chain | | emitS(event, data?) | Gửi tới chính client này | | join(room) / leave(room) | Tham gia / rời phòng | | getRooms() | Danh sách phòng | | setVariable(k, v) / getVariable(k) | Lưu / lấy biến | | getQuery() / getAuthData() | Lấy query / dữ liệu auth | | ping() / getAlive() | Ping / trạng thái sống | | close(code?, reason?) | Đóng kết nối |

SocketLink

| Method | Mô tả | |--------|-------| | emitS(event, data?) | Gửi tới tập target đã chọn |

Adapter

Implement interface này để tạo backend định tuyến tùy chỉnh. Có sẵn MemoryAdapter và NatsAdapter.

new NatsAdapter(natsConnection, prefix?) — natsConnection là connection NATS đã thiết lập, prefix mặc định "ws".

Client (WebsocketBrowser / WebsocketNode)

| Method | Mô tả | |--------|-------| | getInstance() | Lấy instance (singleton) | | connect(url, reconnectInterval?, options?) | Kết nối (options chỉ Node) | | disconnect(shouldReconnect?, code?, reason?) | Ngắt kết nối | | emit(event, data?) | Gửi event | | on/once/off(event, handler?) | Quản lý handler | | isConnected() / getReadyState() | Trạng thái | | setMaxQueueSize(n) / getQueueSize() | Queue | | clearAllHandlers() / getRegisteredEvents() | Handlers |

Chỉ WebsocketNode: ping(data?), setupHeartbeat(interval?), getWebSocketInstance().

⚠️ Breaking changes (v1 → v2)

• init() đổi signature: init({ ws: ServerOptions, adapter?, codec? }) thay vì init(ServerOptions, callback?).
• WebsocketServer không còn kế thừa ws.WebSocketServer (chuyển sang composition).
• SocketLink chỉ còn emitS (bỏ join/leave/close hàng loạt — không an toàn khi scale ngang).
• Client export đổi tên: WebsocketBrowser / WebsocketNode (thay cho WsBrowserClient / WsNodeClient).
• Middleware chain trong onS nay dừng khi gặp lỗi (trước đây vẫn chạy tiếp).

🧪 Test

npm test

📝 License

MIT

👨‍💻 Tác giả

maxsida [email protected]