Transmit

Transmit 是为 AdonisJS 构建的原生、有独特见解的服务器发送事件 (SSE) 模块。这是一种简单而高效的方式，用于向客户端发送实时更新，例如通知、实时聊天消息或任何其他类型的实时数据。

安装

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

node ace add @adonisjs/transmit

你还必须安装 Transmit 客户端包以在客户端监听事件。

npm install @adonisjs/transmit-client

配置

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

另请参阅：Config 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)
})

广播事件 (Broadcasting Events)

你可以使用 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 对象。使用此函数添加自定义头或修改请求对象。

beforeUnsubscribe

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

maxReconnectAttempts

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

onReconnectAttempt

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

onReconnectFailed

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

onSubscribeFailed

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

onSubscription

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

onUnsubscription

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

创建订阅

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

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

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

监听事件

你可以使用接收回调函数的 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 方法删除订阅。该方法返回一个你可以 await 或 void 的 promise。此方法将在服务器上注销订阅。

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