Transmit

本指南介绍 AdonisJS 中使用 Transmit 的实时服务器到客户端通信。你将学习如何：

为 Server-Sent Events 安装并配置 Transmit
注册路由并向已连接的客户端广播事件
定义频道并授权访问私有频道
设置客户端库以实时接收事件
使用传输方式在多个服务器实例之间同步事件
为私有频道订阅对客户端进行认证
监听生命周期钩子以监控连接
使用 Japa 测试广播与频道授权

概述

Transmit 是 AdonisJS 原生的 Server-Sent Events（SSE，服务器发送事件）模块。它提供了一个从服务器到客户端的单向通信通道，让你无需 WebSocket 的开销即可推送实时更新。由于 SSE 使用标准 HTTP，它能穿过那些可能会阻止 WebSocket 连接的防火墙和代理。

Transmit 作为一个围绕频道构建的发布/订阅系统运作。服务器向命名的频道广播消息，客户端订阅它们关心的频道。你可以用授权回调来保护频道，以控制谁接收更新，使其既适用于公开广播，也适用于私有的、特定于用户的通知。

对于客户端到服务器的通信，你继续使用标准的 HTTP 请求。Transmit 只负责服务器到客户端的推送。

安装

使用以下命令安装并配置服务器端包：

node ace add @adonisjs/transmit

查看 add 命令执行的步骤

使用检测到的包管理器安装 @adonisjs/transmit 包。

在 adonisrc.ts 文件中注册以下服务提供者。

adonisrc.ts

{
  providers: [
    // ...other providers
    () => import('@adonisjs/transmit/transmit_provider')
  ]
}

创建 config/transmit.ts 文件。

同时在前端应用中安装客户端库：

npm install @adonisjs/transmit-client

配置

配置文件位于 config/transmit.ts。它控制保活行为和多实例同步。

另见：配置桩文件

config/transmit.ts

import { defineConfig } from '@adonisjs/transmit'

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


      pingInterval

Duration | false

控制发送 ping 消息以保持 SSE 连接存活的频率。接受毫秒为单位的数字、像 '30s' 或 '1m' 这样的时长字符串，或为 false 以禁用 ping。

config/transmit.ts

import { defineConfig } from '@adonisjs/transmit'

export default defineConfig({
  pingInterval: '30s',
  transport: null,
})


      transport

object | null

用于跨多个服务器实例同步事件的传输层配置。对于单实例部署设为 null。

配置详情请参阅多实例同步。

注册路由

Transmit 需要三个 HTTP 路由来处理客户端连接、订阅和取消订阅。在你的路由文件中使用 registerRoutes 方法注册它们。

start/routes.ts

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

transmit.registerRoutes()

这会注册以下路由：

路由	方法	用途
`__transmit/events`	GET	建立 SSE 连接
`__transmit/subscribe`	POST	将客户端订阅到一个频道
`__transmit/unsubscribe`	POST	将客户端从一个频道取消订阅

对路由应用中间件

registerRoutes 方法接受一个可选的回调，用于修改每个已注册的路由。这对于应用中间件很有用，例如要求 SSE 连接进行身份认证。

start/routes.ts

import transmit from '@adonisjs/transmit/services/main'
import { middleware } from '#start/kernel'

transmit.registerRoutes((route) => {
  route.middleware(middleware.auth())
})

你可以根据路由模式有条件地应用中间件。

start/routes.ts

import transmit from '@adonisjs/transmit/services/main'
import { middleware } from '#start/kernel'

transmit.registerRoutes((route) => {
  // 仅为 SSE 连接要求身份认证
  if (route.getPattern() === '__transmit/events') {
    route.middleware(middleware.auth())
  }
})

广播事件

导入 transmit 服务并调用 broadcast 方法，将数据发送给某个频道的所有订阅者。

app/controllers/posts_controller.ts

import transmit from '@adonisjs/transmit/services/main'
import type { HttpContext } from '@adonisjs/core/http'

export default class PostsController {
  async store({ request }: HttpContext) {
    const post = await Post.create(request.all())

    // 向所有订阅者广播新帖子
    transmit.broadcast('posts', { id: post.id, title: post.title })

    return post
  }
}

排除特定客户端

使用 broadcastExcept 将消息发送给除一个或多个特定客户端之外的所有订阅者。当发送者不应收到自己的消息时，这会很有用。

app/controllers/messages_controller.ts

import transmit from '@adonisjs/transmit/services/main'
import type { HttpContext } from '@adonisjs/core/http'

export default class MessagesController {
  async store({ request }: HttpContext) {
    const { uid, content } = request.all()

    // 发送给聊天中的所有人，除了发送者
    transmit.broadcastExcept('chats/1/messages', { content }, uid)
  }
}

第三个参数接受单个 UID 字符串或一组要排除的 UID。

频道

频道名称是区分大小写的字符串，支持字母数字字符和正斜杠。使用正斜杠创建与你的应用资源相匹配的层级结构。

// 用于全局通知的公开频道
transmit.broadcast('notifications', { message: 'System update' })

// 特定资源的频道
transmit.broadcast('chats/1/messages', { content: 'Hello!' })

// 特定用户的频道
transmit.broadcast('users/42', { type: 'profile_updated' })

授权频道

默认情况下，任何客户端都可以订阅任何频道。使用 authorize 方法来限制对敏感频道的访问。创建一个 start/transmit.ts 预加载文件来定义你的授权规则。

node ace make:preload transmit

授权回调会接收当前的 HttpContext 以及提取出的频道参数。返回 true 以允许访问，返回 false 以拒绝。

start/transmit.ts

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

// 仅允许用户订阅他们自己的频道
transmit.authorize<{ id: string }>('users/:id', (ctx, { id }) => {
  return ctx.auth.user?.id === +id
})

频道模式使用与 AdonisJS 路由相同的参数语法。参数会在订阅时从频道名称中提取，并传递给授权回调。

start/transmit.ts

import transmit from '@adonisjs/transmit/services/main'
import Chat from '#models/chat'

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

授权回调支持同步和异步逻辑。如果回调抛出错误，则拒绝访问。

Tip

没有 authorize 回调的频道是公开的。任何客户端都可以订阅它们。仅为需要访问控制的频道注册授权。

认证私有频道

授权回调在 subscribe 请求期间运行，而非在建立 SSE 连接时运行。为了让 ctx.auth.user 在你的 authorize 回调中可用，请用你的身份认证中间件保护 __transmit/subscribe 路由。

start/routes.ts

import transmit from '@adonisjs/transmit/services/main'
import { middleware } from '#start/kernel'

transmit.registerRoutes((route) => {
  if (route.getPattern() === '__transmit/subscribe') {
    route.middleware(middleware.auth())
  }
})

使用基于 token 的认证时，在客户端使用 beforeSubscribe 钩子附加 token。该钩子接收出站的 Request，并在每次 subscribe 调用之前运行。

resources/js/app.ts

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

const transmit = new Transmit({
  baseUrl: window.location.origin,
  beforeSubscribe: (request) => {
    request.headers.set('Authorization', `Bearer ${getToken()}`)
  },
})

Note

位于 __transmit/events 的 SSE 连接是由浏览器的原生 EventSource 打开的，它无法发送自定义请求头。对于基于 token 的认证，请改为认证 subscribe 路由。私有频道仍然会受到频道授权的保护，因为客户端在接收事件之前必须通过 authorize 回调。如果你的应用使用基于 Cookie 的会话，认证会自动生效，因为 subscribe 请求包含了 Cookie（credentials: 'include'）。如果你也保护了 unsubscribe 路由，请以相同方式使用 beforeUnsubscribe 钩子。

客户端设置

使用你的 AdonisJS 服务器 URL 创建一个新的 Transmit 实例。客户端在实例化时会自动建立 SSE 连接。

resources/js/app.ts

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

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

订阅频道

使用 subscription 方法创建订阅，然后调用 create 来激活它。使用 onMessage 注册消息处理器。

resources/js/chat.ts

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

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

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

subscription.onMessage((data) => {
  console.log('New message:', data)
})

你可以在单个订阅上注册多个消息处理器。每个处理器都会接收来自服务器的已解析负载。

resources/js/chat.ts

// 注册多个处理器
subscription.onMessage((data) => {
  appendMessageToUI(data)
})

subscription.onMessage((data) => {
  playNotificationSound()
})

// 注册一个只运行一次的处理器
subscription.onMessageOnce((data) => {
  console.log('First message received:', data)
})

移除消息处理器

onMessage 方法会返回一个取消订阅函数，用于停止某个特定处理器接收消息。

resources/js/chat.ts

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

// 之后，停止这个特定的处理器
unsubscribe()

删除订阅

调用 delete 以完全从一个频道取消订阅。

resources/js/chat.ts

await subscription.delete()

监听连接状态

客户端会跟踪其连接状态，并暴露你可以监听的事件。

resources/js/app.ts

transmit.on('connected', () => {
  console.log('SSE connection established')
})

transmit.on('disconnected', () => {
  console.log('SSE connection lost')
})

transmit.on('reconnecting', () => {
  console.log('Attempting to reconnect...')
})

可用的状态事件有 initializing、connected、disconnected 和 reconnecting。

客户端配置选项


      baseUrl

string

你的 AdonisJS 服务器的 URL，包含协议。这是唯一必需的选项。

const transmit = new Transmit({
  baseUrl: 'https://my-app.com',
})


      maxReconnectAttempts

number

连接断开时的最大重连尝试次数。默认为 5。

const transmit = new Transmit({
  baseUrl: window.location.origin,
  maxReconnectAttempts: 10,
})


      uidGenerator

() => string

用于生成客户端唯一标识符的自定义函数。默认为 crypto.randomUUID()。

import { nanoid } from 'nanoid'

const transmit = new Transmit({
  baseUrl: window.location.origin,
  uidGenerator: () => nanoid(),
})


      beforeSubscribe

(request: Request) => void

在每次 subscribe 请求之前调用的钩子。用它来修改请求，例如添加自定义请求头。

const transmit = new Transmit({
  baseUrl: window.location.origin,
  beforeSubscribe: (request) => {
    request.headers.set('X-Custom-Header', 'value')
  },
})


      beforeUnsubscribe

(request: Request) => void

在每次 unsubscribe 请求之前调用的钩子。与 beforeSubscribe 工作方式相同。


      onReconnectAttempt

(attempt: number) => void

每次重连尝试时调用的回调。接收当前的尝试次数。


      onReconnectFailed

() => void

当达到最大重连尝试次数且客户端停止尝试时调用的回调。


      onSubscribeFailed

(response: Response) => void

当 subscribe 请求失败时调用的回调。接收失败请求的 Response 对象。


      onSubscription

(channel: string) => void

当订阅成功创建时调用的回调。


      onUnsubscription

(channel: string) => void

当订阅成功删除时调用的回调。


      eventSourceFactory

(url: string | URL, options: { withCredentials: boolean }) => EventSource

用于创建 EventSource 实例的自定义工厂。对于原生 EventSource 不可用的环境很有用。


      eventTargetFactory

() => EventTarget | null

用于创建状态变更事件所用 EventTarget 的自定义工厂。返回 null 可禁用状态事件。


      httpClientFactory

(baseUrl: string, uid: string) => HttpClient

用于创建 subscribe 和 unsubscribe 请求所用 HTTP 客户端的自定义工厂。

生命周期钩子

服务器端的 transmit 实例会发出你可以监听的生命周期事件，用于监控与调试。

start/transmit.ts

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

transmit.on('connect', ({ uid }) => {
  console.log(`Client ${uid} connected`)
})

transmit.on('disconnect', ({ uid }) => {
  console.log(`Client ${uid} disconnected`)
})

transmit.on('broadcast', ({ channel, payload }) => {
  console.log(`Broadcast on ${channel}:`, payload)
})

transmit.on('subscribe', ({ uid, channel }) => {
  console.log(`Client ${uid} subscribed to ${channel}`)
})

transmit.on('unsubscribe', ({ uid, channel }) => {
  console.log(`Client ${uid} unsubscribed from ${channel}`)
})

connect、disconnect、subscribe 和 unsubscribe 事件回调还会接收一个 context 属性，其中包含该请求的 HttpContext。

测试

服务器端的 transmit 实例暴露了你可以在测试中监听的生命周期钩子。transmit.on 方法会返回一个用于停止监听的函数，这使得收集请求期间发出的广播并对它们进行断言变得很容易。

以下示例使用 Japa 配合 @japa/api-client 插件。

断言广播

在发起请求之前注册一个监听器，执行会触发广播的操作，然后停止监听并对收集到的事件进行断言。

tests/functional/posts.spec.ts

import transmit from '@adonisjs/transmit/services/main'
import User from '#models/user'
import { test } from '@japa/runner'

test.group('Posts', () => {
  test('broadcasts the created post', async ({ client, assert }) => {
    const user = await User.create({
      fullName: 'Jane',
      email: 'jane@example.com',
      password: 'secret',
    })

    const broadcasts: { channel: string; payload: unknown }[] = []
    const stopListening = transmit.on('broadcast', (event) => broadcasts.push(event))

    const response = await client.post('/posts').loginAs(user).json({
      title: 'Hello world',
    })

    stopListening()

    response.assertStatus(201)
    assert.lengthOf(broadcasts, 1)
    assert.equal(broadcasts[0].channel, `users/${user.id}`)
  })
})

Note

在请求之后立即调用 transmit.on 返回的函数，以避免收集到同一进程中运行的其他测试发出的广播。

测试频道授权

你可以通过直接向 __transmit/subscribe 路由发送 subscribe 请求来测试频道授权。该路由会返回：

当客户端未通过认证时返回 401
当通过认证且被授权时返回 204
当通过认证但被 authorize 回调拒绝时返回 400

tests/functional/transmit.spec.ts

import User from '#models/user'
import { test } from '@japa/runner'

test.group('Transmit', () => {
  test('rejects unauthenticated subscriptions', async ({ client }) => {
    const response = await client.post('/__transmit/subscribe').json({
      uid: 'test-client-uid',
      channel: 'users/1',
    })

    response.assertStatus(401)
  })

  test('allows owners to subscribe to their channel', async ({ client }) => {
    const user = await User.create({
      fullName: 'Jane',
      email: 'jane@example.com',
      password: 'secret',
    })

    const response = await client
      .post('/__transmit/subscribe')
      .loginAs(user)
      .json({
        uid: 'test-client-uid',
        channel: `users/${user.id}`,
      })

    response.assertStatus(204)
  })

  test('denies access to another user channel', async ({ client }) => {
    const owner = await User.create({
      fullName: 'Jane',
      email: 'jane@example.com',
      password: 'secret',
    })

    const otherUser = await User.create({
      fullName: 'John',
      email: 'john@example.com',
      password: 'secret',
    })

    const response = await client
      .post('/__transmit/subscribe')
      .loginAs(otherUser)
      .json({
        uid: 'test-client-uid',
        channel: `users/${owner.id}`,
      })

    response.assertStatus(400)
  })
})

获取频道订阅者

使用 getSubscribersFor 方法检索当前订阅了某个频道的所有客户端的 UID。

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

const subscribers = transmit.getSubscribersFor('chats/1/messages')
console.log(`${subscribers.length} clients connected to this chat`)

多实例同步

当在负载均衡器后面运行多个服务器实例时，在一个实例上广播的事件不会到达连接到其他实例的客户端。Transmit 通过传输层解决了这个问题，该传输层使用消息总线在所有实例之间同步事件。

Redis 传输

安装 ioredis 包并配置 Redis 传输。

npm install ioredis

config/transmit.ts

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

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

MQTT 传输

config/transmit.ts

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

export default defineConfig({
  pingInterval: '30s',
  transport: {
    driver: mqtt({
      url: env.get('MQTT_URL'),
    }),
  },
})

传输层通过配置的消息总线将事件广播到所有已连接的实例。默认的广播频道是 'transmit::broadcast'。如果需要，你可以自定义它。

config/transmit.ts

export default defineConfig({
  transport: {
    driver: redis({ /* ... */ }),
    channel: 'my-app::transmit',
  },
})

生产环境注意事项

为 SSE 禁用压缩

Server-Sent Events 要求响应流不被压缩。如果你的反向代理应用了 GZip 压缩，你必须排除 text/event-stream 内容类型。被压缩的 SSE 流会导致连接不稳定和消息缓冲。

Warning

你必须在反向代理中为 text/event-stream 内容类型禁用压缩。否则会导致 SSE 连接中断或消息被无限期缓冲。

对于 Traefik：

traefik.http.middlewares.gzip.compress.excludedcontenttypes=text/event-stream

对于 Nginx：

# 不要在 gzip_types 中包含 text/event-stream
gzip_types text/plain application/json application/javascript text/css;

OpenTelemetry 使用 OpenTelemetry 为你的 AdonisJS 应用添加分布式追踪与可观测性。