# 服务器发送事件（SSE） Server-Sent Events 通过纯 HTTP 单向流式传输事件 —— 从服务器到客户端。路由处理器用 `http/sse` 打开一个 `text/event-stream` 响应，响应头会立即写入并刷新，连接随即脱离常规的请求/响应生命周期：事件以 HTTP/1.1 分块传输帧的形式追加，直到流关闭或客户端断开。浏览器会自动重连，因此既没有轮询循环，也没有 WebSocket 握手。 ## `http/sse*` 标准库函数 | 函数 | 返回值 | 用途 | |---|---|---| | `[http/sse $res ?opts]` | `%SseStream{}` | 打开 `text/event-stream` 响应 | | `[http/sse-send $stream $data ?opts]` | `true` / `%Err` | 发送一个事件（`:event` 名称、`:id` 标识） | | `[http/sse-comment $stream $text]` | `true` / `%Err` | 发送注释行 —— 手动保活 | | `[http/sse-close $stream]` | `true` / `false` | 将终止块入队并关闭 | ## 打开流

def events {|req res|
    set stream [http/sse $res %{ :retry 3000 :keep-alive 15000 }]
    [http/sse-send $stream "hello"]
    [http/sse-close $stream]
    nil
}

`http/sse` 选项： | 选项 | 含义 | |---|---| | `:retry` | 初始重连延迟（毫秒），作为 `retry:` 字段发送 | | `:keep-alive` | 原生定时器上空闲时的 `: ping` 注释间隔（毫秒） | | `:status` | 状态码覆盖（默认 200） | | `:headers` | 额外的响应头；`[http/header ...]` 优先 | | `:on-close` | 拆除时触发 `{|stream| ...}`（见下文） | 该响应带有 `Content-Type: text/event-stream`、`Cache-Control: no-cache, no-store`、`Connection: keep-alive` 以及 `Transfer-Encoding: chunked`；`X-Accel-Buffering: no` 可防止 nginx 之类的反向代理缓冲该流。`Content-Length` 永远不会被发送。通过 `[http/header $res ...]` 设置的响应头会替换匹配的默认头；`:headers` 选项只会添加行，因此要覆盖 `Content-Type` 之类的默认头时应使用 `[http/header ...]`。当 `$res` 不是 `%Response{}`、响应已被写入（`http/respond`、`http/json`、`http/html`、`http/redirect`）、已经是 SSE 流，或连接已在流式传输时，`http/sse` 返回 `%Err`。客户端在首次刷新时断开也会让它返回 `%Err`。 ## 发送事件 `http/sse-send` 写入并刷新一个事件。仅数据：

[http/sse-send $stream "hello"]

带命名事件和 id：

[http/sse-send $stream "payload" %{ :event "tick" :id "42" }]

多行数据会逐行拆分为一条条 `data:` 行。`:id` 让浏览器在重连后通过 `Last-Event-ID` 请求头续传。`http/sse-send` 返回 `true`，流关闭后则返回 `%Err`（客户端消失、慢消费者拆除、显式关闭）。 `http/sse-comment` 发送注释行（`: text`），可用作应用层保活。`http/sse-close` 将终止块入队、将其排空并关闭连接；流已关闭时返回 `false`。 ## 处理器挂起处理器以协程方式运行，因此 SSE 处理器可以在事件之间挂起。`[sleep]` 和 `[await ...]` 只让出当前处理器 —— 其他连接继续运行：

def ticks {|req res|
    set stream [http/sse $res]
    for i in @{ 1 2 3 } {
        [http/sse-send $stream [str "tick-" $i] %{ :event "tick" }]
        [sleep 1000]
    }
    [http/sse-close $stream]
    nil
}

恢复后写入的事件会到达同一个流。 ## 扇出模式（发布/订阅，集群安全）常见做法是把活动流保存在应用级映射中，并从具名的发布/订阅处理器推送。客户端离开时由 `:on-close` 移除对应条目：

def clients [ref %{}]        # request id -> %SseStream{}

def on-tick {|ev|
    [values [deref $clients]] |> each {|s| [http/sse-send $s $ev~data] }
}

def events {|req res|
    set id $req~id
    set stream [http/sse $res %{ :keep-alive 15000
                                 :on-close {|s| [swap! $clients {|m| [dissoc $m $id]}]} }]
    [swap! $clients {|m| [assoc $m $id $stream]}]
    [pub/subscribe "ticks" "on-tick"]
    nil
}

`$req~id`（请求 id）是客户端映射中便利的稳定键。集群模式下每个工作进程都会订阅；来自任意运行时的发布通过共享 broker 到达其他工作进程，而每个工作进程只写自己的套接字。 ## 客户端浏览器用 `EventSource` 消费该流： ```javascript const es = new EventSource("/events"); es.onmessage = (e) => console.log(e.data); es.addEventListener("tick", (e) => console.log("tick:", e.data)); es.onerror = () => { // readyState === EventSource.CLOSED means the server closed the stream }; ```` 命名事件（`:event`）分派到 `addEventListener`；未命名事件触发 `onmessage`。经过 `retry` 间隔后会自动重连。若要支持断点续传，请在服务端读取 `last-event-id` 请求头（`[get $req~headers "last-event-id"]`），并从自己的事件日志重放。在 shell 中，`curl -N http://localhost:8080/events` 会在帧到达时打印它们。 ## 需要遵守的语义 - 分块、单向、以关闭定界。 `http/sse-close` 写入 `0\r\n\r\n` 终止块并关闭套接字；SSE 连接永远不会被归还给 keep-alive。 - 流是原生资源。 丢弃最后一个引用后，GC 终结器会拆除连接（关闭 fd、释放待处理缓冲区）。GC 拆除时不会触发 `:on-close` —— 只有显式关闭、客户端断开和处理器失败才会回调进 VM。 - 慢客户端会被切断。 每个连接的待发送输出上限为 1 MiB；超出上限即关闭流（且 `http/sse-send` 返回 `%Err`），而不是无限制地缓冲。 - 重连是客户端的职责。 `EventSource` 会自动重试；若需要重放，请使用 `:id` + `Last-Event-ID`。 - 集群重载会丢弃活动流，与任何进行中的请求一样。客户端会重连并落到新一代工作进程上。集成测试覆盖见 `test/test_sse.dft`（分帧、增量投递、心跳、断开/`:on-close`、GC 拆除、校验）。 ## 快速参考 | 形式 | 用途 | |---|---| | `[http/sse $res ?opts]` | 打开流（`:retry`、`:keep-alive`、`:status`、`:headers`、`:on-close`） | | `[http/sse-send $stream $data ?opts]` | 发送事件（`:event`、`:id`）；true 或 `%Err` | | `[http/sse-comment $stream $text]` | 注释行 / 手动保活 | | `[http/sse-close $stream]` | 终止块 + 关闭 | 另见 HTTP 服务器和应用服务器。

文档