---
title: "Stream"
description: "订阅流经 evlog 的广泛事件——可通过 createStreamDrain 在进程内订阅，或通过本地 SSE 流服务器经网络订阅。"
canonical_url: "https://evlog.zhcndoc.com/extend/stream"
---
# Stream

> 订阅流经 evlog 的广泛事件——可通过 createStreamDrain 在进程内订阅，或通过本地 SSE 流服务器经网络订阅。

evlog 提供了一个 **stream 原语**，因此任何本地消费者都可以订阅广泛事件，而无需重新实现一个 drain。它分为两层，并且相互构建：

- **进程内总线**：`createStreamDrain()`，规范的发布/订阅机制。同步监听器、异步迭代器、环形缓冲区重放。
- **网络桥接**：`startStreamServer()`，一个可选的 HTTP 迷你服务器，通过 Server-Sent Events 将进程内总线暴露给浏览器、CLI 或外部开发工具。

<callout icon="i-lucide-shield">

**专为仅限本地而设计。** 两层都存在于单个 Node / Bun / Deno 进程内。它们可在 `pnpm dev`、长期运行的自托管服务器，以及 VM 和容器（Fly、Railway、Coolify……）上工作。它们**无法**在无服务器平台（Vercel Functions、Cloudflare Workers、AWS Lambda）上工作，因为每次调用都是一个隔离进程。在这些平台上，请使用真正的 broker（Redis Streams、NATS、Pub/Sub）来实现跨实例扇出。

</callout>

## 进程内总线

<stream-bus>



</stream-bus>

`createStreamDrain()` 只是一个 drain。将它注册到 evlog drain hook，并在事件发出时订阅事件：无需 HTTP、无需序列化、无需额外跳转。

<prompt :actions="["copy","cursor","claude"]" description="在进程内订阅广泛事件" icon="i-lucide-radio-tower">

在 evlog 的 stream drain 之上建立一个进程内订阅者。

- 从 `evlog/stream` 导入 `createStreamDrain`，并在应用启动时调用一次
- 将返回的 `drain` 注册到适用于我的框架的 evlog drain hook（Nitro：`nitroApp.hooks.hook('evlog:drain', stream.drain)`；Next/standalone：传递给 `initLogger({ drain })`）
- 使用 `stream.subscribe((event) => ...)` 注册同步监听器，或使用 `for await (const event of stream.events())` 进行异步迭代
- 在打开实时迭代器之前，使用 `stream.recent()`（环形缓冲区的快照）为后加入的订阅者准备历史数据
- 调整 `buffer` 以设置重放历史记录，并调整 `perSubscriberQueue` 以处理慢速消费者的背压
- 在无服务器平台上跳过。该 stream 位于进程内，因此隔离的调用之间不会共享它

文档：[https://www.evlog.dev/extend/stream](https://www.evlog.dev/extend/stream)

</prompt>

```ts
import { createStreamDrain } from 'evlog/stream'

const stream = createStreamDrain({ buffer: 200 })

nitroApp.hooks.hook('evlog:drain', stream.drain)

const off = stream.subscribe((event) => {
  if (event.level === 'error') notify(event)
})
off()

for (const past of stream.recent()) {
  bootstrap(past)
}
for await (const event of stream.events()) {
  bootstrap(event)
}
```

### 选项

<table>
<thead>
  <tr>
    <th>
      选项
    </th>
    
    <th>
      类型
    </th>
    
    <th>
      描述
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        buffer
      </code>
    </td>
    
    <td>
      <code>
        number
      </code>
    </td>
    
    <td>
      <code>
        recent()
      </code>
      
       快照的环形缓冲区大小。设为 <code>
        0
      </code>
      
       可禁用。默认：<code>
        500
      </code>
      
      。
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        perSubscriberQueue
      </code>
    </td>
    
    <td>
      <code>
        number
      </code>
    </td>
    
    <td>
      每个 <code>
        events()
      </code>
      
       异步迭代器在最旧事件被丢弃前可排队的最大事件数。drain 本身永远不会被阻塞。默认：<code>
        1000
      </code>
      
      。
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        filter
      </code>
    </td>
    
    <td>
      <code>
        (event) => boolean
      </code>
    </td>
    
    <td>
      可选谓词，在每个被 drain 的事件上运行——返回 <code>
        false
      </code>
      
       可完全跳过该事件（既不缓冲也不投递）。
    </td>
  </tr>
</tbody>
</table>

`stream.drain(events)` 可接受单个事件或批量事件。`stream.recent()` 返回已缓冲事件的快照（最旧在前，最新在后），可用于临时检查或为 UI 面板填充初始数据。

## 网络桥接：stream server

<sse-wire>



</sse-wire>

`startStreamServer()` 会在与你的应用相同的进程中启动一个小型 `node:http` 服务器，使用其自身的临时端口，并通过 Server-Sent Events 暴露进程内总线。任何消费者（浏览器标签页、CLI、Tauri/Electron 开发工具）都可以订阅，而不会触及你的应用 API。

<callout icon="i-lucide-shield">

**严格选择启用。** 除非你显式设置选项，否则不会启动任何内容。开发环境中不会自动启用，只有在你要求时服务器才会启动。

</callout>

<prompt :actions="["copy","cursor","claude"]" description="通过 SSE 暴露 evlog stream" icon="i-lucide-radio">

开启本地 stream server，这样我就可以从浏览器标签页、CLI 或 Tauri/Electron 开发工具订阅广泛事件。

- 检测我的框架并显式选择启用（Nuxt：在 `nuxt.config.ts` 中设置 `evlog.stream: true`；Next.js：在 `instrumentation.ts` 中设置 `defineStreamedInstrumentation({ stream: true })`；Hono/Express/Fastify/Elysia/standalone：在启动时调用一次 `startStreamServer()`，并将返回的 `drain` 注册到 evlog drain hook）
- 默认永远不要在生产环境中启用；通过 `process.env.NODE_ENV !== 'production'` 或功能标志控制
- 对于共享开发环境，设置 `token: process.env.EVLOG_STREAM_TOKEN`，并让消费者在每个请求中以 `Authorization: Bearer <token>` 的形式发送它
- 从 `.evlog/stream.url`（或 Nuxt 中的 `/api/_evlog/stream-info`）发现 URL，永远不要硬编码端口，因为端口是临时的
- 在无服务器平台上跳过。服务器位于进程内

文档：[https://www.evlog.dev/extend/stream](https://www.evlog.dev/extend/stream)

</prompt>

### 启动内容

当你启用后，evlog 会调用 `startStreamServer()`，并：

1. 选择一个临时可用端口（或你指定的端口）
2. 在 `127.0.0.1`（或你的主机）上启动一个小型 `node:http` 服务器
3. 将发现的 URL 写入 `.evlog/stream.url`
4. 在服务器控制台打印一行横幅
5. 将进程内 stream drain 挂接到 evlog 管道中
6. 为 SSE 流暴露 `GET /`，为握手元数据暴露 `GET /info`

在关闭时（SIGINT / SIGTERM / process exit），服务器会清理监听器并删除 URL 文件。

### 按框架启用

<framework-tabs>

```ts [Nuxt]nuxt.config.ts
export default defineNuxtConfig({
  modules: ['evlog/nuxt'],
  evlog: {
    stream: true,
    // 或：stream: { port: 4444, token: process.env.EVLOG_STREAM_TOKEN }
  },
})
```

```ts [Next.js]instrumentation.ts
import { defineStreamedInstrumentation } from 'evlog/next/stream'

export const { register } = defineStreamedInstrumentation({
  stream: true,
})
```

```ts [Hono / Express / Fastify / standalone]
import { startStreamServer } from 'evlog/stream'

if (process.env.NODE_ENV !== 'production' && process.env.EVLOG_STREAM === '1') {
  const { drain } = await startStreamServer()
  // 将 `drain` 接入你框架的 evlog drain hook
}
```

</framework-tabs>

Hono / Express / Fastify 集成不需要“功能 PR”，因为 `startStreamServer()` 与你的框架**正交**。你只需启动一次，并像连接其他 drain 一样，将它的 `drain` 连接到 evlog 管道。

### 发现

这个迷你服务器运行在随机端口上，因此任何消费者都必须先发现它。

```ts [.evlog/stream.url]
http://127.0.0.1:53942
```

直接从磁盘读取：

```ts
import { readFile } from 'node:fs/promises'

const url = (await readFile('.evlog/stream.url', 'utf-8')).trim()
```

或者，对于 Nuxt 应用中同源的浏览器标签页，访问发现路由：

```ts
const { url } = await fetch('/api/_evlog/stream-info').then(r => r.json())
```

### 线协议格式

每条 SSE 消息的形状都是 `{ evlog: '1', type, data }`：

<table>
<thead>
  <tr>
    <th>
      <code>
        type
      </code>
    </th>
    
    <th>
      何时
    </th>
    
    <th>
      <code>
        data
      </code>
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        hello
      </code>
    </td>
    
    <td>
      连接后的第一帧
    </td>
    
    <td>
      <code>
        { evlogVersion, bufferSize, heartbeatMs }
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        replay
      </code>
    </td>
    
    <td>
      紧随 <code>
        hello
      </code>
      
       之后，当消费者传入 <code>
        ?since=<iso>
      </code>
      
       时重放缓冲事件
    </td>
    
    <td>
      <code>
        WideEvent
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        event
      </code>
    </td>
    
    <td>
      之后每个发出的事件
    </td>
    
    <td>
      <code>
        WideEvent
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        ping
      </code>
    </td>
    
    <td>
      心跳（默认每 15 秒一次，可通过 <code>
        heartbeatMs
      </code>
      
       配置）
    </td>
    
    <td>
      <code>
        { ts: number }
      </code>
    </td>
  </tr>
</tbody>
</table>

```ts [Browser consumer]
const { url } = await fetch('/api/_evlog/stream-info').then(r => r.json())
const es = new EventSource(url)

es.onmessage = (msg) => {
  const { type, data } = JSON.parse(msg.data)
  if (type === 'event') events.push(data)
}
```

### Auth + CORS

<table>
<thead>
  <tr>
    <th>
      选项
    </th>
    
    <th>
      行为
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        token
      </code>
    </td>
    
    <td>
      设置后，服务器要求每个请求都带有 <code>
        Authorization: Bearer <token>
      </code>
      
      ，否则返回 401。
    </td>
  </tr>
  
  <tr>
    <td>
      （未设置 token）
    </td>
    
    <td>
      仅当 <code>
        Host
      </code>
      
       标头为 <code>
        localhost
      </code>
      
      、<code>
        127.0.0.1
      </code>
      
       或 <code>
        [::1]
      </code>
      
      ，且存在 <code>
        Origin
      </code>
      
       标头时其主机名也与其中之一完全匹配，才接受连接。<code>
        https://localhost.example.com
      </code>
      
       以及通过解析到 <code>
        127.0.0.1
      </code>
      
       的域名访问的页面都会收到 403。
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        host
      </code>
    </td>
    
    <td>
      默认 <code>
        127.0.0.1
      </code>
      
      ——永远无法从局域网访问。只有在设置了 token 时才应覆盖为 <code>
        0.0.0.0
      </code>
      
      。
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        heartbeatMs
      </code>
    </td>
    
    <td>
      心跳间隔（默认 <code>
        15000
      </code>
      
      ）。
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        buffer
      </code>
    </td>
    
    <td>
      保存在底层默认 stream 上的环形缓冲区——通过 <code>
        ?since=<iso>
      </code>
      
       为后加入的客户端重放。默认 <code>
        500
      </code>
      
      。
    </td>
  </tr>
</tbody>
</table>

## 进一步了解

- [FS reader](https://evlog.zhcndoc.com/extend/fs-reader)：重放或跟踪历史 NDJSON 文件（跨进程，重启后仍然存在）
- [Consumer recipes](https://evlog.zhcndoc.com/extend/consumer-recipes)：构建一个最小开发工具，通过 curl + jq 管道传输，重放历史记录后进入实时模式


## Sitemap

See the full [sitemap](https://evlog.zhcndoc.com/sitemap.md) for all pages.