Documentation
¶
Overview ¶
Package streamx 提供跨 SSE / WebSocket / Chunked Transfer 协议的 流式响应抽象(Event/Source/Sink/Hub),跟 ekit 已有 hertz/sse(Stream 单连接 SSE)+ wsutil(Broadcaster 房间广播) 协同工作,业务可一次写 Event,多协议扇出。
设计灵感:hertz 官方教程 https://www.cloudwego.io/zh/docs/hertz/tutorials/basic-feature/sse/ 跟 hertz-contrib/websocket 各自独立,ekit streamx 在它们之上加 "跨协议统一事件"层,让 AI 助手/聊天/通知等场景能用同一份事件源 扇出到 HTTP SSE + WebSocket(浏览器+App 端)。
已知未桥接:hertz-contrib/streamx(不存在)、kitex-contrib/streamx (也不存在),本包是 ekit 自己的抽象层。
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrSinkClosed = errors.New("streamx: sink closed")
ErrSinkClosed 表示 Sink 已关闭,Emit 返回它。
var ErrSinkEmpty = errors.New("streamx: empty sink data")
ErrSinkEmpty 保留 sentinel(避免 unused 误删,实际用 ErrSinkClosed)。
var ErrStreamExhausted = errors.New("streamx: source exhausted")
ErrStreamExhausted 表示 SliceSource 已 emit 完所有事件。
Functions ¶
func MarshalEvent ¶
MarshalEvent 把 Event.Data 序列化成 []byte(默认 JSON)。Sink 实现可 用这个 helper 走统一 JSON 编码。
Types ¶
type Event ¶
Event 是流式响应的最小数据单元,跨 SSE/WS/Chunked Transfer 通用。
- ID: SSE Last-Event-ID / WS 业务自定义;空表示不写入 id 字段。
- Type: SSE event / WS message type;空表示"message"默认。
- Data: 业务数据(序列化策略由 Sink 决定:JSON for SSE/WS,raw for chunked)。
- Retry: SSE retry 字段(毫秒)。⚠️ 当前 SSEStreamOp 适配层未暴露 retry 位,置值无效(2026-09-06 文档修正)。
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub 协调 N 个 Source → N 个 Sink 的多对多转发。 Run 阻塞直到 ctx 取消或所有 Source 耗尽/出错。
设计要点:
- 单 goroutine 串行读 Source(避免 backpressure 复杂处理)
- 每 emit 到一个 Sink 是独立调用,任一失败仅记录不中断流
- Source/Sink 须在 Run 前 Attach:Run 启动时对两者做快照,Run 中途 Attach 的对本次 Run 不可见(2026-09-06 文档修正,原注释"可在 Run 前后动态 Attach"与实现不符)
func (*Hub) AttachSink ¶
AttachSink 挂一个 Sink 到 hub,Hub 关闭后调用返回 false。
func (*Hub) AttachSource ¶
AttachSource 挂一个 Source 到 hub,Hub 关闭后调用返回 false。
type SSESink ¶
type SSESink struct {
// contains filtered or unexported fields
}
SSESink 把 Event 推到一个已打开的 hertz SSE 连接(由 hertz/sse.Stream 创建)。典型用法:在 handler 末尾 defer sse.NewStream(c),然后把 stream 4 方法传给 NewSSESinkFromFuncs 推给 Hub。
官方依据:https://www.cloudwego.io/zh/docs/hertz/tutorials/basic-feature/sse/ + ekit hertz/sse 桥接的 Stream.Send/SendRaw。
⚠️ 背压限制：SSE 写底层无写超时（hertz app 层未暴露 write deadline）， 读取停滞的客户端（TCP 窗口耗尽）会阻塞 Emit 调用——Hub 串行扇出场景 会连带阻塞其他 sink。生产部署建议在网关/负载均衡层配置发送空闲超时， 或避免把多个 sink 挂在同一个串行 Hub 上。
func NewSSESinkFromFuncs ¶
func NewSSESinkFromFuncs( send func(eventType string, data any) error, sendRaw func(id, eventType string, data []byte) error, keepAlive func() error, close func() error, ) *SSESink
NewSSESinkFromFuncs 显式传入 4 个方法构造 SSESink(避免 ekit package 内部循环依赖)。
典型用法(把 hertz/sse.Stream 适配成 SSESink):
stream := hertzsse.NewStream(c)
sink := streamx.NewSSESinkFromFuncs(stream.Send, stream.SendRaw,
stream.KeepAlive, stream.Close)
defer sink.Close()
func (*SSESink) Emit ¶
Emit 实现 Sink。
ID 透传(2026-09-06 复审修复):Last-Event-ID 断线续传依赖 event id, 原实现对 JSON 载荷走 Send 把 ev.ID 整个丢掉 — 现在带 ID 的事件一律 编码后走 SendRaw。[]byte 载荷仍直走 SendRaw(避免 JSON 双编码)。
⚠️ Event.Retry 当前不透传:SSEStreamOp 的 Send/SendRaw 签名没有 retry 位,上游 hertz sse.Writer.WriteEvent 虽支持,ekit 适配层未暴露 — 置该 字段无效(字段文档已同步修正,不再宣称"仅 SSE 生效")。
type SSEStreamOp ¶
type SSEStreamOp interface {
Send(eventType string, data any) error
SendRaw(id, eventType string, data []byte) error
KeepAlive() error
Close() error
}
SSEStreamOp 是 hertz/sse.Stream 暴露给 streamx 的最小子集(4 方法)。 用 type alias 而非直接 import hertz/sse 是为了避免 streamx ↔ hertz/sse 循环依赖;业务调用 NewSSESinkFromFuncs 时直接传 hertz/sse.Stream 的 方法引用。
type Sequence ¶
type Sequence uint64
Sequence 是单调递增的事件序号,业务可借此判断丢包/重排。 0 表示未分配(由 Sink/Hub 在 Emit 时填充)。
type Sink ¶
type Sink interface {
// Emit 推一个事件;返回 error 表示推送失败(io.ErrClosedPipe 表示
// 客户端断开)。
Emit(ctx context.Context, ev Event) error
// Close 关闭 sink,后续 Emit 返回 ErrSinkClosed。
Close() error
}
Sink 是流式输出目标抽象(推模式)。Emit 把事件写到具体协议端 (SSE / WS / chunked HTTP / 自定义)。
type SliceSource ¶
type SliceSource struct {
// contains filtered or unexported fields
}
SliceSource 把内存切片包装成 Source,适合测试和简单 case。
func NewSliceSource ¶
func NewSliceSource(evs []Event) *SliceSource
NewSliceSource 从事件列表构造 Source,事件会被 emit 0..N-1 次。
Source Files