Server-Sent Events

SSE（Server-Sent Events）通过一个持续的 HTTP 响应，让服务端单向推送事件到浏览器。NestJS 使用 @Sse() 将 Controller 方法标记为事件流端点，方法必须返回 Observable<MessageEvent>。

方案	通信方向	浏览器客户端	适合场景
HTTP REST	请求—响应	fetch	常规读写
SSE	服务端 → 客户端	原生 EventSource	通知、任务进度、状态更新
Socket.IO	双向事件	socket.io-client	聊天、协作、房间广播

SSE 不是双向协议；客户端写操作仍通过普通 HTTP 请求完成。它复用 HTTP 的认证、Cookie、代理与负载均衡基础设施，浏览器会在连接中断后自动重连。

sequenceDiagram
  participant B as Browser EventSource
  participant C as NotificationsController
  participant S as NotificationsService
  participant W as 任意写入用例
  B->>C: GET /api/notifications/stream
  C->>S: stream(topic)
  W->>S: publish(payload)
  S-->>C: Observable<MessageEvent>
  C-->>B: event: notification\ndata: {...}

完整示例

下面以“按主题订阅通知”为例。内存 Subject 只用于说明流转；生产环境需要将事件持久化或改为消息队列、Redis Streams 等可在多实例间共享的事件源。

1. DTO 与事件服务

// dto/create-notification.dto.ts
import { IsObject, IsString, MaxLength, MinLength } from 'class-validator'

export class CreateNotificationDto {
  @IsString()
  @MinLength(1)
  @MaxLength(64)
  topic!: string

  @IsObject()
  payload!: Record<string, unknown>
}

// notifications.service.ts
import { Injectable } from '@nestjs/common'
import { Observable, Subject, filter, map } from 'rxjs'
import type { MessageEvent } from '@nestjs/common'

type Notification = {
  id: string
  topic: string
  payload: Record<string, unknown>
}

@Injectable()
export class NotificationsService {
  private readonly notifications = new Subject<Notification>()
  private nextId = 1

  publish(topic: string, payload: Record<string, unknown>): void {
    this.notifications.next({
      id: String(this.nextId++),
      topic,
      payload,
    })
  }

  stream(topic?: string): Observable<MessageEvent> {
    return this.notifications.pipe(
      filter((notification) => !topic || notification.topic === topic),
      map((notification) => ({
        id: notification.id,
        type: 'notification',
        data: notification,
      })),
    )
  }
}

MessageEvent 的 data 是事件负载，type 映射为浏览器监听的事件名，id 用于标识事件。内存 Subject 不会保留历史，也无法在进程重启或多实例间恢复事件。

2. Controller：写入与订阅端点

import { Body, Controller, Post, Query, Sse } from '@nestjs/common'
import type { MessageEvent } from '@nestjs/common'
import type { Observable } from 'rxjs'

import { CreateNotificationDto } from './dto/create-notification.dto'
import { NotificationsService } from './notifications.service'

@Controller('notifications')
export class NotificationsController {
  constructor(private readonly notificationsService: NotificationsService) {}

  @Post()
  create(@Body() input: CreateNotificationDto) {
    this.notificationsService.publish(input.topic, input.payload)
    return { accepted: true }
  }

  @Sse('stream')
  stream(@Query('topic') topic?: string): Observable<MessageEvent> {
    return this.notificationsService.stream(topic)
  }
}

入口已使用 app.setGlobalPrefix('api') 时，端点为：

• POST /api/notifications：产生一条通知。
• GET /api/notifications/stream?topic=orders：订阅 orders 主题。

@Sse() 仍是 HTTP Controller 路由，因此会受到全局前缀、Guard、Interceptor 和 CORS 配置影响；这与 Socket.IO Gateway 不同。

3. Module

import { Module } from '@nestjs/common'

import { NotificationsController } from './notifications.controller'
import { NotificationsService } from './notifications.service'

@Module({
  controllers: [NotificationsController],
  providers: [NotificationsService],
})
export class NotificationsModule {}

在 AppModule 的 imports 中加入 NotificationsModule，Nest 才会注册 HTTP 与 SSE 路由。

4. 浏览器客户端

const source = new EventSource('http://localhost:3000/api/notifications/stream?topic=orders')

source.addEventListener('notification', (event) => {
  const notification = JSON.parse(event.data)
  console.log(notification.topic, notification.payload)
})

source.onerror = (error) => {
  console.error('SSE connection failed', error)
}

// 组件卸载或不再订阅时调用
source.close()

服务器的 type: 'notification' 会触发同名事件；未设置 type 时浏览器调用 onmessage。浏览器断线会自动尝试重连，显式 close() 才会停止连接。

用 curl 观察流：

curl -N 'http://localhost:3000/api/notifications/stream?topic=orders'

curl -X POST http://localhost:3000/api/notifications \
  -H 'content-type: application/json' \
  -d '{"topic":"orders","payload":{"orderId":"o_123","status":"paid"}}'

-N 禁用 curl 自身的输出缓冲。收到的事件格式类似：

id: 1
event: notification
data: {"id":"1","topic":"orders","payload":{"orderId":"o_123","status":"paid"}}

连接管理与可靠性

• @Sse() 返回的 Observable 在客户端断开时会被 Nest 自动取消订阅。若持有数据库游标、订阅或计时器，在 RxJS 的 finalize() 中释放资源。
• 低频流应发送心跳，或由反向代理配置足够长的读取超时；否则空闲连接可能被中间层断开。
• SSE 的 id 只能标识事件。需要断线补发时，将事件写入持久化日志，并依据浏览器携带的最后事件 ID 查询缺失事件；内存流无法实现恢复。
• HTTP/1.1 下浏览器对同一域名的并发 SSE 连接数较低，多标签页或多个主题订阅优先复用一条流；HTTP/2 可改善并发流限制。

认证与安全

EventSource 不能像 fetch 一样随意附加 Authorization 请求头。跨域 Cookie 认证可使用 new EventSource(url, { withCredentials: true })，并在服务端配置精确的 CORS 来源与凭据策略。不要在查询参数中放置长期访问令牌；需要 Bearer 认证时，采用同源 HTTP-only Cookie、受控代理，或能自定义请求头的 SSE 客户端实现。

对 SSE 端点仍应使用 Guard 检查身份与主题权限，限制单用户连接数与事件频率，并确保推送负载不包含敏感数据。

参考

• NestJS Server-Sent Events
• MDN：Using server-sent events