Server-Sent Events

Server-Sent Events (SSE) là một công nghệ server push cho phép client nhận update tự động từ server qua kết nối HTTP. Mỗi notification được gửi dưới dạng một block text kết thúc bởi hai dấu newline liên tiếp (tìm hiểu thêm tại đây).

Cách dùng (Usage)

Để bật Server-Sent Events cho một route (route được đăng ký trong một controller class), gắn decorator @Sse() lên method handler.

@Sse('sse')
sse(): Observable<MessageEvent> {
  return interval(1000).pipe(map((_) => ({ data: { hello: 'world' } })));
}

Trong ví dụ trên, chúng ta định nghĩa một route tên sse, cho phép chúng ta phát (propagate) update real-time. Các event này có thể được lắng nghe bằng EventSource API.

Method sse trả về một Observable phát ra (emit) nhiều MessageEvent (trong ví dụ này, nó emit một MessageEvent mới mỗi giây). Object MessageEvent nên tuân theo interface sau để khớp với specification:

export interface MessageEvent {
  data: string | object;
  id?: string;
  type?: string;
  retry?: number;
}

Với những cài đặt trên, giờ ta có thể tạo một instance của class EventSource trong ứng dụng phía client, truyền route /sse (khớp với endpoint ta đã truyền vào decorator @Sse() bên trên) làm tham số cho constructor.

Instance EventSource mở một kết nối bền vững (persistent) đến HTTP server, server này gửi event ở dạng text/event-stream. Kết nối vẫn mở cho đến khi bị đóng bằng cách gọi EventSource.close().

Sau khi kết nối được mở, các message đến từ server sẽ được chuyển tới code của bạn dưới dạng event. Nếu message đến có field event, event được trigger sẽ chính là giá trị của field event đó. Nếu không có field event, một event message chung sẽ được fire (nguồn).

const eventSource = new EventSource('/sse');
eventSource.onmessage = ({ data }) => {
  console.log('New message', JSON.parse(data));
};

Client ngắt kết nối (Client disconnection)

Khi client đóng kết nối SSE (ví dụ, gọi eventSource.close()), NestJS tự động unsubscribe khỏi Observable được trả về, việc này sẽ dừng event stream và dọn dẹp (clean up) mọi resource liên quan — bao gồm cả interval timer trong ví dụ trên.

Để chạy logic dọn dẹp (teardown) tùy chỉnh khi client ngắt kết nối, dùng operator finalize:

@Sse('sse')
sse(): Observable<MessageEvent> {
  return interval(1000).pipe(
    map((_) => ({ data: { hello: 'world' } })),
    finalize(() => console.log('Client disconnected')),
  );
}

Một handler @Sse() có thể là bất đồng bộ (asynchronous) — trả về Promise<Observable> thay vì trả Observable trực tiếp. Điều này thường gặp khi stream cần bước setup tốn kém trước khi event đầu tiên có thể được tạo ra: mở một database cursor, lấy (acquire) một model session, hoặc xác thực (authorize) với một service downstream.

@Sse('stream')
async stream(): Promise<Observable<MessageEvent>> {
  const session = await createSession();

  return new Observable(subscriber => {
    // ...tạo event từ `session`
  });
}

Ở đây có một lỗ hổng (gap). Nếu client ngắt kết nối trong khi promise vẫn đang resolve, Observable được trả về sẽ không bao giờ được subscribe — Nest sẽ không khởi động một producer mà consumer của nó đã biến mất. Đó là hành vi đúng đối với stream, nhưng nó có nghĩa là logic teardown của Observable sẽ không bao giờ chạy, và bất cứ thứ gì được allocate trong lúc setup (session ở trên) sẽ bị leak (rò rỉ).

Để lấp lỗ hổng này, inject AbortSignal của request bằng decorator @SseSignal():

import { MessageEvent, Sse, SseSignal } from '@nestjs/common';
import { EMPTY, Observable } from 'rxjs';

@Sse('stream')
async stream(@SseSignal() signal: AbortSignal): Promise<Observable<MessageEvent>> {
  const session = await createSession();

  if (signal.aborted) {
    // Client đã ngắt kết nối trong lúc setup. Observable bên dưới sẽ không bao giờ
    // được subscribe, nên giải phóng resource ngay ở đây.
    await session.close();
    return EMPTY;
  }

  return new Observable(subscriber => {
    const stream = session.start();
    stream.on('data', data => subscriber.next({ data }));

    return () => {
      stream.stop();
      session.close();
    };
  });
}

Thời gian sống của signal (Signal lifetime)

Signal đại diện cho thời gian sống (lifetime) của response SSE, không chỉ riêng kết nối. Nó bị abort ngay khi stream kết thúc vì bất kỳ lý do gì:

• client ngắt kết nối;
• Observable hoàn tất (completed);
• Observable bị lỗi (errored).

Điều này khiến signal trở thành một cleanup hook duy nhất cho toàn bộ request. Thay vì lặp lại logic teardown ở cả một abort listener và teardown function riêng của Observable, bạn có thể gắn resource vào signal một lần và để chúng được giải phóng ở mọi đường thoát (exit path):

@Sse('stream')
async stream(@SseSignal() signal: AbortSignal): Promise<Observable<MessageEvent>> {
  // Signal bị abort khi response kết thúc, nên fetch sẽ bị hủy bất kể
  // client ngắt kết nối hay stream đơn giản là đã hoàn tất.
  const upstream = await fetch(UPSTREAM_URL, { signal });

  return new Observable(subscriber => {
    // ...
  });
}

Vì signal cũng bị abort khi hoàn tất bình thường, signal.aborted chỉ có ý nghĩa như một cách kiểm tra “client đã rời đi chưa?” trong lúc setup — trước khi Observable được trả về. Tại thời điểm đó stream chưa thể hoàn tất được, nên một signal đã bị abort chỉ có thể mang một nghĩa duy nhất là client đã ngắt kết nối.

Bên trong một producer, signal cũng là một cách tiện lợi để kết thúc stream khi client rời đi:

return new Observable<MessageEvent>(subscriber => {
  const timer = setInterval(() => subscriber.next({ data: 'tick' }), 1000);
  const onAbort = () => subscriber.complete();

  signal.addEventListener('abort', onAbort, { once: true });

  return () => {
    clearInterval(timer);
    signal.removeEventListener('abort', onAbort);
  };
});

Ví dụ (Example)

Một ví dụ hoạt động thực tế có sẵn tại đây.