Transmit

Transmit 是一个为 AdonisJS 构建的、原生且有明确设计取向（opinionated）的 Server-Sent-Event（SSE）模块。它是一种简单高效的方式，用于向客户端发送实时更新，例如通知、实时聊天消息或任何其他类型的实时数据。

安装

使用以下命令安装并配置该包：

node ace add @adonisjs/transmit

你还需要安装 Transmit 客户端包，以便在客户端监听事件。

npm install @adonisjs/transmit-client

配置

transmit 包的配置存储在 config/transmit.ts 文件中。

另见：配置 stub

import { defineConfig } from '@adonisjs/transmit'

export default defineConfig({
  pingInterval: false,
  transport: null,
})

pingInterval

用于向客户端发送 ping 消息的间隔。该值以毫秒为单位，或使用字符串 Duration 格式（即 10s）。设置为 false 以禁用 ping 消息。

transport

Transmit 支持跨多个服务器或实例同步事件。你可以通过引用所需的传输层来启用此功能（目前仅支持 redis）。设置为 null 以禁用同步。

Copy code to clipboard

import env from '#start/env'
import { defineConfig } from '@adonisjs/transmit'
import { redis } from '@adonisjs/transmit/transports'

export default defineConfig({
  transport: {
    driver: redis({
      host: env.get('REDIS_HOST'),
      port: env.get('REDIS_PORT'),
      password: env.get('REDIS_PASSWORD'),
      keyPrefix: 'transmit',
    })
  }
})

使用 redis 传输时，请确保已安装 ioredis。

注册路由

你必须注册 transmit 路由，以允许客户端连接到服务器。路由需手动注册。

import transmit from '@adonisjs/transmit/services/main'

transmit.registerRoutes()

你也可以通过手动绑定控制器来逐个注册每个路由。

const EventStreamController = () => import('@adonisjs/transmit/controllers/event_stream_controller')
const SubscribeController = () => import('@adonisjs/transmit/controllers/subscribe_controller')
const UnsubscribeController = () => import('@adonisjs/transmit/controllers/unsubscribe_controller')

router.get('/__transmit/events', [EventStreamController])
router.post('/__transmit/subscribe', [SubscribeController])
router.post('/__transmit/unsubscribe', [UnsubscribeController])

如果你想修改路由定义，例如使用 Rate Limiter 和 auth 中间件来防止某些 transmit 路由被滥用，你可以更改路由定义，或向 transmit.registerRoutes 方法传递一个回调。

import transmit from '@adonisjs/transmit/services/main'

transmit.registerRoutes((route) => {
  // Ensure you are authenticated to register your client
  if (route.getPattern() === '__transmit/events') {
    route.middleware(middleware.auth())
    return
  }

  // Add a throttle middleware to other transmit routes
  route.use(throttle)
})

频道（Channels）

频道用于对事件进行分组。例如，你可以为通知设置一个频道，为聊天消息设置另一个频道，等等。 它们会在客户端订阅时即时创建。

频道名称

频道名称用于标识频道。它们区分大小写，且必须是字符串。除了 / 之外，你不能在频道名称中使用任何特殊字符或空格。以下是一些有效频道名称的示例：

import transmit from '@adonisjs/transmit/services/main'

transmit.broadcast('global', { message: 'Hello' })
transmit.broadcast('chats/1/messages', { message: 'Hello' })
transmit.broadcast('users/1', { message: 'Hello' })

频道授权

你可以使用 authorize 方法授权或拒绝对频道的连接。该方法接收频道名称和 HttpContext。它必须返回一个布尔值。

import transmit from '@adonisjs/transmit/services/main'
import Chat from '#models/chat'
import type { HttpContext } from '@adonisjs/core/http'

transmit.authorize<{ id: string }>('users/:id', (ctx: HttpContext, { id }) => {
  return ctx.auth.user?.id === +id
})

transmit.authorize<{ id: string }>('chats/:id/messages', async (ctx: HttpContext, { id }) => {
  const chat = await Chat.findOrFail(+id)
  
  return ctx.bouncer.allows('accessChat', chat)
})

广播事件

你可以使用 broadcast 方法向频道广播事件。该方法接收频道名称和要发送的数据。

import transmit from '@adonisjs/transmit/services/main'

transmit.broadcast('global', { message: 'Hello' })

你也可以使用 broadcastExcept 方法向除某个客户端之外的任意频道广播事件。该方法接收频道名称、要发送的数据，以及你想要忽略的 UID。

transmit.broadcastExcept('global', { message: 'Hello' }, 'uid-of-sender')

跨多个服务器或实例同步

默认情况下，广播事件仅在 HTTP 请求的上下文中有效。不过，如果你在配置中注册了 transport，你就可以使用 transmit 服务从后台广播事件。

传输层负责跨多个服务器或实例同步事件。它的工作方式是使用一个 Message Bus（消息总线）将任何事件（如广播的事件、订阅和取消订阅）广播给所有已连接的服务器或实例。

负责你客户端连接的那个服务器或实例将接收该事件，并将其广播给客户端。

Transmit 客户端

你可以使用 @adonisjs/transmit-client 包在客户端监听事件。该包提供了一个 Transmit 类。客户端默认使用 EventSource API 连接到服务器。

import { Transmit } from '@adonisjs/transmit-client'

export const transmit = new Transmit({
  baseUrl: window.location.origin
})

配置 Transmit 实例

Transmit 类接受一个具有以下属性的对象：

baseUrl

服务器的基础 URL。该 URL 必须包含协议（http 或 https）和域名。

uidGenerator

一个为客户端生成唯一标识符的函数。该函数必须返回一个字符串。它默认为 crypto.randomUUID。

eventSourceFactory

一个创建新 EventSource 实例的函数。它默认为 WebAPI 的 EventSource。如果你想在 Node.js、React Native 或任何其他不支持 EventSource API 的环境中使用该客户端，你需要提供自定义实现。

eventTargetFactory

一个创建新 EventTarget 实例的函数。它默认为 WebAPI 的 EventTarget。如果你想在 Node.js、React Native 或任何其他不支持 EventTarget API 的环境中使用该客户端，你需要提供自定义实现。返回 null 以禁用 EventTarget API。

httpClientFactory

一个创建新 HttpClient 实例的函数。它主要用于测试目的。

beforeSubscribe

一个在订阅频道之前调用的函数。它接收频道名称和发送到服务器的 Request 对象。使用此函数来添加自定义请求头或修改 request 对象。

beforeUnsubscribe

一个在取消订阅频道之前调用的函数。它接收频道名称和发送到服务器的 Request 对象。使用此函数来添加自定义请求头或修改 request 对象。

maxReconnectAttempts

最大重连尝试次数。它默认为 5。

onReconnectAttempt

一个在每次重连尝试之前调用的函数，接收迄今为止已进行的尝试次数。使用此函数来添加自定义逻辑。

onReconnectFailed

一个在重连尝试失败时调用的函数。使用此函数来添加自定义逻辑。

onSubscribeFailed

一个在订阅失败时调用的函数。它接收 Response 对象。使用此函数来添加自定义逻辑。

onSubscription

一个在订阅成功时调用的函数。它接收频道名称。使用此函数来添加自定义逻辑。

onUnsubscription

一个在取消订阅成功时调用的函数。它接收频道名称。使用此函数来添加自定义逻辑。

创建订阅

你可以使用 subscription 方法创建对某个频道的订阅。该方法接收频道名称。

const subscription = transmit.subscription('chats/1/messages')
await subscription.create()

create 方法在服务器上注册该订阅。它返回一个 promise，你可以 await 或 void 它。

监听事件

你可以使用接收回调函数的 onMessage 方法监听订阅上的事件。你可以多次调用 onMessage 方法以添加不同的回调。

subscription.onMessage((data) => {
  console.log(data)
})

你也可以使用接收回调函数的 onMessageOnce 方法只监听一次频道。

subscription.onMessageOnce(() => {
  console.log('I will be called only once')
})

停止监听事件

onMessage 和 onMessageOnce 方法返回一个函数，你可以调用它来停止监听特定的回调。

const stopListening = subscription.onMessage((data) => {
  console.log(data)
})

// Stop listening
stopListening()

删除订阅

你可以使用 delete 方法删除订阅。该方法返回一个 promise，你可以 await 或 void 它。此方法将在服务器上注销该订阅。

await subscription.delete()

避免 GZip 干扰

在部署使用 @adonisjs/transmit 的应用时，务必确保 GZip 压缩不会干扰 Server-Sent Events（SSE）所使用的 text/event-stream 内容类型。对 text/event-stream 应用压缩可能导致连接问题，进而引发频繁断连或 SSE 失败。

如果你的部署使用了反向代理（如 Traefik 或 Nginx）或其他应用 GZip 的中间件，请确保为 text/event-stream 内容类型禁用压缩。

Traefik 配置示例

traefik.http.middlewares.gzip.compress=true
traefik.http.middlewares.gzip.compress.excludedcontenttypes=text/event-stream
traefik.http.routers.my-router.middlewares=gzip