jsonstream

README

JsonStream

基于自定义二进制帧协议承载 JSON 业务数据——一条 TCP 连接多路复用请求/响应、流式、双工、单向与发布/订阅，内建心跳、断线恢复、可选压缩/加密与 credit 背压。纯标准库 Go 参考实现，零第三方依赖；协议规范语言无关，其他语言实现规划中。

• 协议规范（语言无关，跨语言实现的单一事实源）：docs/protocol.md——只拿到这一份文档就应能写出可互通的其他语言实现；Go 实现已可用（go get github.com/cuihairu/jsonstream），Python/Rust 等其他语言实现规划中。
• 题目要求：docs/interview-requirements.md——题面原文分组与逐条实现现状（含分帧三条的如实标注）。
• 在线文档：https://cuihairu.github.io/jsonstream/（随 main 分支更新），全部页面的入口导航：

文档（仓库源文件）	在线版
上手指南	getting-started
API 参考	api
基准实测	benchmarks
故障排查 / FAQ	faq
协议规范	protocol
题目要求	interview-requirements
与 WebSocket 对照	websocket-comparison
TCP 流特性与协议横评	tcp-and-landscape
架构与取舍	DESIGN
设计笔记	design-notes
知识点梳理	NOTES

设计与知识点

本节讲协议设计的逐点依据：帧格式、交互模型、元数据分离、心跳、断线恢复、压缩加密、背压、pub/sub 混用与实测教训。与 WebSocket 的能力对照，摘成三行——

• WS 有、本协议没有或显式不做：浏览器原生可达、TLS 一等承载与 443 复用、子协议/扩展协商、文本/二进制 opcode 区分（客户端 Masking 则明确不需要——raw TCP 直连没有那个威胁模型）。分片传输已从「不做」名单移出：规范定稿为独立 FRAGMENT 帧类型（见下）。
• WS 标准没有、本协议内建：请求/响应、发布/订阅、流式、credit 背压、断线恢复、Stream ID 多路复用——这些在 WS 应用里都要自造。
• 逐条依据（协议章节与代码位置）见 docs/websocket-comparison.md；更宽的协议横评见 docs/tcp-and-landscape.md。

帧格式：定长头 + 长度前缀，对着 WebSocket 抄作业再还回去一点

 0               8               16              24              32
 +---------------+---------------+---------------+---------------+
 |     Magic "JS" (0x4A53)       |    Version=1  |     Flags     |
 +---------------+---------------+---------------+---------------+
 |     Type      |   Reserved    |          Stream ID            |
 +---------------+---------------+---------------+---------------+
 |                        Payload Length                          |
 +---------------+---------------+---------------+---------------+
 |   Meta Length (2B, Flags.HasMeta 时) | Metadata (JSON, 路由/主题) |
 +---------------------------------------------------------------+
 |                     Payload (JSON)                             |
 +---------------------------------------------------------------+

这里考察的是二进制协议设计的基本功：TCP 粘包怎么破、字节序、变长还是定长。TCP 是字节流没有消息边界，分帧手段无非三种：定长、分隔符、长度前缀。JSON 里有分隔符歧义，文本帧不可靠；我在 14B 定长头里放 32 位大端长度前缀，解码端「读完头 → 两次 ReadFull」成为无条件操作。

对 WebSocket（RFC 6455 §5.2）的帧格式，我做了两处减法、一处换形，每个都有明确理由：

• 分片换形，不做 FIN+continuation。16 MiB 单帧上限继续管住单帧内存（超限长度直接断开）；更大的逻辑消息由协议拆成连续 chunk 运行——开帧保留原类型并置 FlagFragmented，续段/收尾用独立 FRAGMENT 帧类型且不带元数据与变换标志，运行不可穿插故接收端免块序号，重组总量以 MaxMessageSize（默认 64 MiB）为硬帽（逐字段规范见 docs/protocol.md §3.4，Go 参考实现落地中）。不能包体过大就拒传；也不用 FIN 位形式——那套状态机是为浏览器流式解析设计的，独立续段类型的帧形更简单、解析分支同样克制。
• 不做客户端掩码。MASK 防的是代理缓存投毒这个历史包袱，专用客户端/服务端直连场景没有这个威胁模型。
• 不做 7/16/64 位变长长度。变长编码多数帧省 2~6 字节，换来解码端的分支状态机；固定 4B 的 16 MiB 上限同时是内存防线（对端报多大的长度，就预分配多大的缓冲，上限即闸门）。

净效果是帧头 14B，比 WebSocket 的 2~6B 厚。每帧多花的约 10B 对数百字节起的 JSON 消息是 <3% 的税；对心跳小帧是实打实的浪费，所以心跳帧不带 Metadata、payload 为空，14B 就是下限，我接受。

比 RSocket 的 6~10B 帧头多出的 8B 是 Magic 和 Version：RSocket 跑在 Reactor Netty 之上，传输层已替它做了连接识别；我直接裸奔 TCP，Magic 给误连/端口探测一个立即判废的机会，Version 给握手期快速失败的依据——这两个字节是我自付的保险费。Flags 的保留位必须为 0，解码端见到非零保留位直接按 Malformed 断开，给未来升级留语义清晰的门。

交互模型：RSocket 四原语，Stream ID 多路复用

考察点是从 HTTP/1.1 的「一请求一连接」到 HTTP/2 的「一连接多流」这个跃迁。一条 TCP 连接上并发跑请求/响应、无限流、双工通道、订阅，归属怎么定——答案是每帧必带 Stream ID，控制帧除外，一切皆流：请求/响应是生命周期极短的流，订阅是以 SUBACK 为起点长流。不存在「广播帧」这种无主帧。

四种交互原语照 RSocket 的划分：REQUEST→RESPONSE（一问一答）、REQUEST+Flags.Stream→N×RESPONSE+COMPLETE（流式）、ONEWAY（单向，连错误都不回）、REQUEST+Flags.Channel（双向多帧）。两个我认为值得写的细节：

• 响应帧自带终结语义：请求/响应不追加 COMPLETE 帧，一次交互少一个往返的帧和一次状态转换。这是 RSocket 比「gRPC 流必须显式 end」更适合小消息的地方。
• CANCEL 是流的一部分：背压解决「生产快于消费」，CANCEL 解决「消费者根本不要了」。HTTP/1.1 没有对应物，只能断连接。

我和 RSocket 的分歧在一个点：它把三种请求拆成三个帧类型，我用一个 REQUEST 类型加 Flags.Stream/Flags.Channel 位表达。理由是类型表还要装 SUBSCRIBE/PUBLISH、心跳、恢复这些控制帧，解析分支数希望克制；代价也要诚实——收到帧那一刻不能预判交互模式，要等 Metadata 里的路由查到 handler 注册类型才能校验，对「按模式做准入」的网关不够友好，中间件只看帧头无法区分一次性请求和无限流。v1 用「Flags 声明与 handler 不一致回 ERROR(PROTOCOL)」兜底，真出现该场景，升级路径是把这两个位提为独立帧类型，语义完全等价。

Stream ID 按发起方分奇偶（客户端奇数、服务端偶数），和 HTTP/2 同思路：不用等待协商就知道新到的流 ID 是谁发起的，两端各自单调递增互不冲突。

Metadata 与 Payload 分离：路由可读，业务不透明

route/topic 放在帧头的 Metadata 段而不是 JSON payload 里，模仿 RSocket 的 composite metadata。考察点是协议对中间件的友好性：鉴权、限流、metrics、网关路由不解码 payload 就能工作。

反面后果想清楚过：如果路由混进 payload，加密后它就是不透明字节，中间件想路由就必须持有密钥——等于强迫每一跳都拿到端到端秘钥，加密的意义就没了。代价是 Metadata 恒为明文，「哪条连接在订阅哪个主题」可以被链路旁路统计；需要隐藏时把路由名纳入加密范围就要放弃中间件可读性，二选一，没有免费午餐。

心跳：为什么不用 TCP keepalive

应用层 PING/PONG、读空闲超过 1.5× 间隔判死。考察点是连接健康和进程健康是两回事：TCP keepalive 探测的是对端内核协议栈还活着，探不出对端进程死锁、事件循环卡住、GC 停顿——应用层心跳测的才是「对端应用还能处理这条连接」。1.5× 而不是 1× 是容忍一次丢帧抖动，2× 则判死太慢。心跳是双向各自独立发的，两端配置通过 CONNACK 协商一致。

断线恢复：会话保留 + 重放，以及诚实的边界

考察点是可靠传输的状态机设计：断线重连恢复不是「自动重连」四个字，是「断开期间的状态谁替你记着、记多久、怎么续」。

我的设计：客户端自动重连（指数退避加抖动），CONNECT 携带 session_id；服务端在保留期内（默认 30s、每会话 4 MiB 上限）保留订阅关系和断开期间产生的下行帧，重连以 CONNACK{resumed=true} 确认并重放（at-least-once）；同会话新连接顶替旧连接（takeover），防止半死连接复活打架；保留超期则 resumed=false，挂起流以 SESSION_EXPIRED 失败、订阅自动重订、OnResumeFailed 回调交给应用重建状态。

这套设计的边界我在文档里写死了，因为它们不是实现瑕疵，是 v1 的主动取舍：

• 重放只覆盖进入保留队列的下行帧，断连瞬间 TCP 在途的帧不保证送达也不重放——恢复窗口内「可能丢最后一帧」，且无 per-frame 序号，客户端无法判重，非幂等操作（下单、扣款）要靠业务 ID 幂等兜底。要做精确一次就得给每帧配序号加 ACK，那是另一层复杂度，题面没要求。
• 上行不缓存：断连瞬间客户端发出的请求会丢，「重连成功但刚才那个请求发出去过没有」是未定义状态。把难题诚实下放，也比假装解决了强。

压缩与加密：帧内自描述，先压后加不可逆

考察点是参数化设计和对称加密的正确用法。三个决策：

• 每帧的 Flags 如实标注本帧是否压缩/加密，解码管线只看帧不看协商结果。这样心跳帧永远裸奔、大 payload 才压缩，单连接内混合存在；CONNACK 协商只决定默认值而不是枷锁。反过来「协商定了就只能怎样」会让小帧白付税、大帧错过收益。
• 顺序必须先压缩后加密。GCM 输出在统计上近似随机字节，对密文压缩得率为零；先加后压等于白付 28B 开销（12B nonce + 16B tag）还保留着明文可压缩的冗余。这个顺序错了性能直接不可用，而且错误顺序不报错——只能靠设计冻结。
• 加密用 AES-256-GCM + 32B PSK，握手帧恒为明文（握手帧加密就鸡生蛋了）。要诚实说：PSK 不解决密钥分发，题面要的是「可选加密」的参数化，不是「替代 TLS」；生产上外层套 TLS，Flags.Encrypted 留给「经过中间件仍要保密」的端到端场景。

压缩策略 v1 是全有或全无（Config 决定，载荷 ≥64B 才实际压），没做「压缩率不达标回退明文」的自适应——那需要每帧压两次或边压边判断，复杂度收益不成比例。

背压：credit 而不是 LEASE，以及它为什么必须是可选的

三种流控模型的对照，考察点是流控到底在约束什么：

模型	代表	语义	复杂度
滑动窗口	TCP / HTTP/2	按字节授权	高（字节级记账）
credit 按条数	本协议、Reactive Streams request(n)	授权 N 条，交付后归还	中
租约按时间	RSocket LEASE	约束速率不约束在途量	低

题面要的背压是「生产快于消费时把压力传回去」，credit 直接表达「我还没消费完，你别再发」；LEASE 防的是滥用不是背压（RSocket 自己也靠订阅方 request(n) 做真流控）；滑动窗口按字节记账对 JSON 消息过度工程，按条计数恰好是业务方心智单位。

我把 credit 做成默认关闭的可选项，因为它不是免费能力：额度归零时 Emit() 必须阻塞，是全协议唯一反向影响应用并发模型的机制；归还时机即语义（收到帧≠归还，交付应用才算，早了背压失效、晚了吞吐骤降）；双工模式下双方互等对方 credit 有死锁风险；连接级窗口一个慢订阅拖住同连接所有流（队头阻塞的流控版）。这些坑每个都要工程决策，不该默认强加给所有用户。请求/响应和 ONEWAY 我明确不纳入 credit——单帧即终结的交互没有「连续生产」可言，硬套只白加一个 RTT。

pub/sub 与 req/res 混用一条连接：边界划在流上

两种模式共享连接、共享帧头，边界不清就会出「这帧属于谁」的歧义。考察点是协议语义的正交性。我的划分：主题是双向命名空间，方向由生产方决定——C→S 的 PUBLISH 打服务端注册的 handler，S→C 的 PUBLISH 投递给订阅，同一帧类型两个方向语义对偶，避免「上行主题/下行主题」两套类型。错误边界不对称是刻意的：订阅建立失败回 ERROR，投递失败（订阅方回调出错）不回帧——回帧就成了「响应的响应」，方向倒置。这条不对称必须写进文档，否则实现者必然在这里犹豫。

混用的收益是一条 TCP、一次握手、一套心跳与恢复；代价是共享流控与队头阻塞。我在文档里把两边都挑明了，不粉饰。

测试里抓到的真 bug：不这么设计的后果实证

「为什么这么设计」最有说服力的部分，是我实测后自己踩到又修掉的 9 个真缺陷——每一个都是一条设计教训：

1. 解压炸弹（fuzz 抓到）：解压结果不设上限，单帧 16 MiB 的 flate 数据能膨胀三个数量级——不防等于把 OOM 开关交给对端。
2. 编码不自洽（fuzz 抓到）：Flags 声明带 Metadata 但内容为空时跳过 metaLen 段，解析-编码不再是互逆——序列化必须满足 round-trip 恒等。
3. credit 破坏 at-least-once：断连后取额度立即失败，handler 误判退出、保留队列变空，重放丢了——连接级流控泄漏进了会话级恢复语义，修复为失败改道保留队列。
4. 死连接上的 select 双就绪：发送通道有空位时 select 随机选中已死分支，帧静默丢失——Go 的 select 随机性在错误路径上是真陷阱。
5. Stream ID 撞号：重连后 ID 计数器归零，新流与迁移流同 ID，旧流迟到的 COMPLETE 误杀新流——多路复用加恢复，ID 必须跨重连单调。
6. 陈旧连接引用：订阅句柄持有创建时的 endpoint，重连后退订帧发给死连接被静默吞——出站一律取当前 endpoint，让编译器消灭陈旧引用。
7. responder 流不迁移：服务端被动流的 handler 在重连后悬空白产帧，能把保留队列撑爆——会话迁移必须连 handler 一起搬。
8. 断连后悬挂：服务端主动发起的交互在连接死亡后永久挂起等待一个必然不会来的响应——响应是上行、不缓存，等待无意义，断连即败。
9. 元数据上限 off-by-one（gosec 抓到）：上限写成 64 KiB 整，但长度字段是 uint16——恰 64 KiB 会静默截断成 0 编出错乱帧。上界必须等于字段可表达的值，不是顺手的整数。

实现与测试结果

关键实现要点

• 纯标准库、零第三方依赖，核心十来个文件：帧编解码（frame.go）、变换管线（压缩/加密，transform.go）、传输与流控（transport.go、credit.go）、流与会话（flow.go、endpoint.go、client.go、server.go）。
• 客户端与服务端共用同一套 endpoint 抽象：交互模式、流状态机、流控只写一遍，两侧只差 ID 奇偶与握手方向。出站帧统一经当前 endpoint，杜绝重连后的陈旧引用。
• 压缩路径按帧复用 flate 编解码器（sync.Pool + Reset）：按帧新建 writer 的实测代价 ~1.3ms / ~800KB 垃圾每帧，池化后压缩往返 4.1 倍提速、分配降 162 倍。
• 测试上唯一「为测试而设」的是五个包级接缝（marshal/aes/gcm/rand/超时），其余覆盖率全靠并发时序构造，不动生产逻辑。

测试与质量结果

• 库包语句覆盖率 100.0%（812/812，2026-09-28 go test -coverprofile 实测），CI 有门禁（跌破即失败）。两个示例包同为 100.0%：曾长期留作例外的 5 条"连接恰好在这一瞬间死掉"的窄竞态分支已全部补齐——日志泊点掐线、盲发连发硬币、信用抽干三种确定性构造，外加一个示例侧 Publish 检查点钩子（先经 90 轮扫描实证纯时序抓不住）；逐条构造与教训在 DESIGN.md §11.3。
• 212 个测试/基准函数（198 测试 + 9 基准 + 5 fuzz 靶），覆盖契约、集成、并发时序构造，并钉住文档记录的已知盲区契约（typed-nil *Error 降级：见 DESIGN.md §7 末条）；examples/ 从"零测试的展示代码"变成"被真实客户端端到端跑过"。
• -race -count=2 全绿；goroutine 泄漏守卫（20 并发客户端回归基线 ±2）。
• 五条 fuzz 靶（帧解析/元数据编解码/变换层/握手状态机/握手 JSON 协商）长跑累计千万级 execs 零 crash（后两靶 2026-09-27 加入，首日各 120s 深挖 174 万/141 万 execs），实锤修复 2 个真 bug（上文 1、2）。CI 里还有一层 fuzz 冒烟：靶名自动发现、每靶固定 20s 预算、崩溃即失败并把 crash 语料落成 testdata/fuzz/ 回归用例（DESIGN §11.4）。
• 六件静态检查零告警：staticcheck、vet、gofmt、revive、gosec、gocritic；govulncheck 零可触达漏洞；nilness 零告警——全部已纳入 CI 门禁（双工具链腿：1.24 证 go.mod 的最老支持版本可构建可测试，stable 腿承载工具，工具全部钉在实测兼容的版本）。
• 五平台交叉编译通过（windows/darwin × amd64/arm64、linux/arm64，CGO 关）——已纳入 CI 门禁，且放在 1.24 腿跑：最老支持工具链上五平台可构建，对更新工具链是更强的保证；go.mod 声明的 go 1.24 同样经真实工具链在 CI 实测可构建。

性能（go test -bench . -benchtime 2s，i9-10880H / Go 1.24，量级参考；更多基准——变换四组合、allocs 逐项解读与负载口径注记——见 docs/benchmarks.md，其数字取自高负载窗口，与本表空载值相差 2~4× 属预期）：

基准	结果
帧编解码往返 64B / 1KiB / 64KiB	~1.1µs / ~1.1µs / ~67µs
变换管线（1.4KiB JSON）	明文 ~15ns；AES-GCM ~5µs；flate ~60µs
请求/响应 RTT（本机回环）	~0.1ms

验证方式

以下命令可直接复制执行（Go ≥ 1.24，无第三方依赖）：

git clone https://github.com/cuihairu/jsonstream.git
cd jsonstream

# 构建 + 全量测试 + 静态检查
go build ./...
go test ./...
go vet ./...

# 覆盖率：库包 100.0%（CI 门禁），examples/server 100.0%，examples/client 100.0%
go test -cover ./...

# 性能基准
go test -bench . -benchtime 2s

# fuzz 冒烟（CI 同款：逐包自动发现全部靶，每靶 20s）
for pkg in $(go list ./...); do
  for t in $(go test -list 'Fuzz.*' "$pkg" | grep '^Fuzz'); do
    go test -run '^$' -fuzz "^${t}$" -fuzztime 20s "$pkg"
  done
done

# fuzz 深挖单靶（注意 -fuzz 只接受单个包，./... 会报错）
go test -run '^$' -fuzz FuzzReadFrame -fuzztime 300s .

# 端到端示例：终端 1 启动服务端
go run ./examples/server
# 终端 2 运行客户端（请求/响应、流式取消、双工、单向、发布/订阅、服务端主动发起）
go run ./examples/client

引入（Go ≥ 1.24，零第三方依赖）：

go get github.com/cuihairu/jsonstream

客户端最小用法（逐方法参考见 docs/api.md，完整可运行示例见 examples/）：

c, _ := jsonstream.Dial(ctx, addr, jsonstream.DefaultConfig())
m, _ := c.Request(ctx, "math.add", map[string]int{"a": 2, "b": 40}) // 请求/响应
s, _ := c.Stream(ctx, "range", map[string]int{"n": 100})           // 流式
defer s.Cancel()
sub, _ := c.Subscribe(ctx, "ticks", func(m *jsonstream.Message) error { return nil })
_ = c.Publish(ctx, "metrics", v) // client → server

发起类 API 一律以 ctx 打头：SendOneWay/Publish（Client 与 Server 两侧）与 Request/Stream/Channel 同规，ctx 约束「等可用连接」与「等发送入队」两段等待（DESIGN §8.4）。该签名在首发（v0.1.0）前定版，不为旧签名留别名——旧行为等价于传 context.Background()。

压缩/加密/背压都是参数化开关，握手协商后生效（双方都开才启用）：

cfg := jsonstream.DefaultConfig()
cfg.Compress = true // ≥64B 载荷走 flate
cfg.Encrypt = true  // AES-256-GCM，先压后加
cfg.Key = key32     // 32 字节预共享密钥
cfg.Credit = 64     // 连接级信用窗口，生效值取双方最小

仓库布局与多语言规划

本仓库规划未来实现其他语言版本（Python/Rust/Java 等），布局按此定位：

• Go 实现保持仓库根（go.mod 在根目录，Go 生态标准布局）。不会把 Go 代码挪进子目录——module path 变更会破坏所有现有 import。
• 未来其他语言实现以平级子目录进入（python/、rust/…），互不干扰。
• 协调中枢是语言无关的协议规范：docs/protocol.md 是所有实现的单一事实源，任何语言的实现互通性以它为准；实现细节文档（DESIGN/NOTES 等）描述的是 Go 参考实现，不构成跨语言契约。

Documentation

Overview

Package jsonstream 实现一套基于 TCP 的 JSON 消息帧协议：定长二进制 帧头承载路由与元信息，载荷为 JSON，单条连接多路复用五种交互模式 （请求/响应、流式、双工通道、单向、发布/订阅），并在传输层参数化 心跳保活、断线重连与会话恢复、flate 压缩、AES-256-GCM 加密与连接级 信用窗口背压。

客户端入口是 Dial，服务端入口是 NewServer；协议规范见 docs/protocol.md，设计权衡见 docs/design-notes.md。

并发模型

Client 与 Server 的公开方法（注册回调、发起交互、Close 等）均可 被多个 goroutine 并发调用：出站帧经带缓冲通道串行写出，会话与端点 引用有锁保护，Close 幂等且与在途请求竞争时以错误终结全部挂起流。

回调与消费端的边界：

• 流式与双工 handler（HandleStream/HandleChannel 及客户端对称面） 每条流一个 goroutine 顺序执行，handler 内的 Emitter.Emit、 Channel.Send/Receive 无需额外同步；
• ReadStream.Next 与 Channel.Receive 是单消费接口：多个 goroutine 并发调用会互相抢帧（不崩溃，但帧的归属不确定）；
• Subscription 的主题回调在内部独占 goroutine 中逐帧串行执行， 回调内可安全调用 Client 的其他方法；
• HandlePublish 与 HandleOneWay 注册的回调每帧一个 goroutine 并发 执行——同一主题高频发布时回调可能同时运行且不保证顺序，需要 顺序或互斥时由回调自行保证。

Config 是值类型：传入 Dial/NewServer 之后再修改原变量，不影响已 建立的端。

Index

• Constants
• Variables
• type Channel
  • func (c *Channel) Cancel() error
  • func (c *Channel) Close() error
  • func (c *Channel) Receive(ctx context.Context) (*Message, error)
  • func (c *Channel) Send(v any) error
• type Client
  • func Dial(ctx context.Context, addr string, cfg Config) (*Client, error)
  • func (c *Client) Channel(ctx context.Context, route string, payload any) (*Channel, error)
  • func (c *Client) Close() error
  • func (c *Client) Handle(route string, h func(*Request) (any, error))
  • func (c *Client) HandleChannel(route string, h func(*Channel) error)
  • func (c *Client) HandleOneWay(route string, h func(*Message) error)
  • func (c *Client) HandleStream(route string, h func(*Request, Emitter) error)
  • func (c *Client) OnReconnect(fn func())
  • func (c *Client) OnResumeFailed(fn func(error))
  • func (c *Client) Publish(ctx context.Context, topic string, payload any) error
  • func (c *Client) Request(ctx context.Context, route string, payload any) (*Message, error)
  • func (c *Client) SendOneWay(ctx context.Context, route string, payload any) error
  • func (c *Client) SessionID() string
  • func (c *Client) Stream(ctx context.Context, route string, payload any) (*ReadStream, error)
  • func (c *Client) Subscribe(ctx context.Context, topic string, h func(*Message) error) (*Subscription, error)
• type Config
  • func DefaultConfig() Config
• type ConnectJSON
• type Emitter
• type Error
  • func (e *Error) Error() string
  • func (e *Error) IsProtocol() bool
• type ErrorCode
  • func (c ErrorCode) String() string
• type Frame
  • func ReadFrame(r io.Reader) (*Frame, error)
  • func (f *Frame) String() string
• type FrameType
  • func (t FrameType) String() string
• type Header
• type Logger
• type Message
  • func (m *Message) Decode(v any) error
• type ReadStream
  • func (s *ReadStream) Cancel() error
  • func (s *ReadStream) Err() error
  • func (s *ReadStream) Next(ctx context.Context) (*Message, bool)
• type Request
  • func (r *Request) Context() context.Context
• type Server
  • func NewServer(ln net.Listener, cfg Config) (*Server, error)
  • func (s *Server) Channel(ctx context.Context, sessionID, route string, payload any) (*Channel, error)
  • func (s *Server) Close() error
  • func (s *Server) Handle(route string, h func(*Request) (any, error))
  • func (s *Server) HandleChannel(route string, h func(*Channel) error)
  • func (s *Server) HandleOneWay(route string, h func(*Message) error)
  • func (s *Server) HandlePublish(topic string, h func(*Message) error)
  • func (s *Server) HandleStream(route string, h func(*Request, Emitter) error)
  • func (s *Server) OnAuth(h func(*ConnectJSON) error)
  • func (s *Server) Publish(ctx context.Context, topic string, payload any) error
  • func (s *Server) Request(ctx context.Context, sessionID, route string, payload any) (*Message, error)
  • func (s *Server) SendOneWay(ctx context.Context, sessionID, route string, payload any) error
  • func (s *Server) Serve() error
  • func (s *Server) Sessions() []string
  • func (s *Server) Stream(ctx context.Context, sessionID, route string, payload any) (*ReadStream, error)
• type Subscription
  • func (s *Subscription) Close() error

Constants

const (
	DefaultHeartbeat      = 10 * time.Second
	DefaultRetention      = 30 * time.Second
	DefaultRetentionBytes = 4 << 20
	DefaultBackoffInitial = 100 * time.Millisecond
	DefaultBackoffMax     = 5 * time.Second
	DefaultDialTimeout    = 5 * time.Second
)

协议参数默认值。

const (

	// ProtocolVersion 是当前协议版本。
	ProtocolVersion uint8 = 1

	// MaxPayloadSize 是单帧载荷（变换后）的上限，超限直接断开以防 OOM。
	MaxPayloadSize = 16 << 20 // 16 MiB
	// MaxMetadataSize 是 Metadata 的上限。线上 metaLen 字段是 uint16，
	// 恰 64 KiB 的元数据无法表达（编码会截断成 0），故上界为 64 KiB-1。
	MaxMetadataSize = math.MaxUint16 // 64 KiB - 1

)

线上帧布局（大端序），见 docs/protocol.md §3：

 0  1   Magic "JS"
 2      Version
 3      Flags
 4      Type
 5      Reserved
 6..9   Stream ID
10..13  Payload Length（变换后字节数）
14..15  Meta Length（仅 Flags.HasMeta）+ Metadata（明文 JSON）
...     Payload

const (
	FlagCompressed uint8 = 1 << 0
	FlagEncrypted  uint8 = 1 << 1
	FlagHasMeta    uint8 = 1 << 2
	FlagStream     uint8 = 1 << 3
	FlagChannel    uint8 = 1 << 4
)

Flags 位定义。Stream/Channel 由请求发起方声明，服务端据此校验注册的 handler 类型；Compressed/Encrypted 描述本帧 payload 是否经过对应变换。

Variables

var ErrClosed = errors.New("jsonstream: connection closed")

ErrClosed 表示连接已关闭。

var ErrMalformed = errors.New("jsonstream: malformed frame")

ErrMalformed 表示帧结构损坏（坏魔数、长度越界、截断等）。

Types

type Channel

type Channel struct {
	// contains filtered or unexported fields
}

Channel 是双工（Flags.Channel）流的两端通用视图：双方都可 Send/Receive，Close 发送 COMPLETE（半关闭："我说完了"）， Cancel 立即终结整条流。Send 可并发调用；Receive 是单消费接口， 在单个 goroutine 中循环调用（见包文档「并发模型」）。

func (*Channel) Cancel

func (c *Channel) Cancel() error

Cancel 立即终结整条流。

func (*Channel) Close

func (c *Channel) Close() error

Close 半关闭：发送 COMPLETE 表示本端不再发送；仍可 Receive 直到对端关闭。

func (*Channel) Receive

func (c *Channel) Receive(ctx context.Context) (*Message, error)

Receive 接收对端一帧。对端 COMPLETE 后返回 io.EOF；流出错返回错误。

func (*Channel) Send

func (c *Channel) Send(v any) error

Send 发送一帧（受背压约束；额度耗尽时阻塞直到补充或 ctx/流终结）。 出站一律经当前 endpoint：会话恢复迁移后是重连的新连接，构造时的 endpoint 已死。

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client 是 JsonStream 客户端：自动重连（指数退避 + 抖动）、会话恢复 （resumed 时保留挂起流等待重放）、断连期间 API 调用挂起而非报错。 公开方法可被多个 goroutine 并发调用（见包文档「并发模型」）。

func Dial

func Dial(ctx context.Context, addr string, cfg Config) (*Client, error)

Dial 建立连接并完成握手（阻塞到 CONNACK）。 首连失败（网络不可达或握手被拒）立即返回错误，与常规客户端库直觉一致； 连接成功后的断连才由 connectLoop 自动重连（指数退避 + 抖动，受 Reconnect 控制）。

func (*Client) Channel

func (c *Client) Channel(ctx context.Context, route string, payload any) (*Channel, error)

Channel 发起双工通道。

func (*Client) Close

func (c *Client) Close() error

Close 关闭客户端并停止重连循环；所有挂起的流以连接关闭错误终结。

func (*Client) Handle

func (c *Client) Handle(route string, h func(*Request) (any, error))

Handle 注册请求/响应处理器，应答服务端发起的 Request。

func (*Client) HandleChannel

func (c *Client) HandleChannel(route string, h func(*Channel) error)

HandleChannel 注册双工通道处理器（双向多帧，Close 半关闭）。

func (*Client) HandleOneWay

func (c *Client) HandleOneWay(route string, h func(*Message) error)

HandleOneWay 注册单向消息处理器（无应答，错误仅记日志）。

func (*Client) HandleStream

func (c *Client) HandleStream(route string, h func(*Request, Emitter) error)

HandleStream 注册服务端发起的流式处理器（Emitter 逐帧下发）。

func (*Client) OnReconnect

func (c *Client) OnReconnect(fn func())

OnReconnect 在每次（非首次）重连成功后回调。

func (*Client) OnResumeFailed

func (c *Client) OnResumeFailed(fn func(error))

OnResumeFailed 在会话恢复失败（过期/未知）后回调，应用可据此重建状态。

func (*Client) Publish

func (c *Client) Publish(ctx context.Context, topic string, payload any) error

Publish 把消息发布到服务端主题（由 Server.HandlePublish 处理）。 ctx 语义与 SendOneWay 相同：约束等连接与等入队两段。

func (*Client) Request

func (c *Client) Request(ctx context.Context, route string, payload any) (*Message, error)

Request 请求/响应：等待单帧响应（响应帧自带终结语义）。

func (*Client) SendOneWay

func (c *Client) SendOneWay(ctx context.Context, route string, payload any) error

SendOneWay 单向发送，协议保证不产生任何响应帧。ctx 同时约束「等可用 连接」（断连期间等重连）与「等发送入队」（对端停滞排满发送队列）。

func (*Client) SessionID

func (c *Client) SessionID() string

SessionID 返回服务端分配的当前会话 ID。

func (*Client) Stream

func (c *Client) Stream(ctx context.Context, route string, payload any) (*ReadStream, error)

Stream 发起流式请求，逐帧读取直到 COMPLETE/ERROR。

func (*Client) Subscribe

func (c *Client) Subscribe(ctx context.Context, topic string, h func(*Message) error) (*Subscription, error)

Subscribe 订阅服务端广播的主题，投递经回调交付。

type Config

type Config struct {
	// Heartbeat 是 PING 间隔；读空闲超过 1.5 倍即判定断连。零值取默认。
	Heartbeat time.Duration
	// Compress 启用 flate 压缩（载荷 ≥64B 才实际压缩）。
	Compress bool
	// Encrypt 启用 AES-256-GCM；启用时 Key 必须是 32 字节。
	Encrypt bool
	// Key 是预共享密钥（AES-256），不在线上传输。
	Key []byte
	// Credit >0 时启用连接级信用背压窗口（条数）；生效值 = min(双方配置)。
	Credit int
	// Auth 透传到 CONNECT 的令牌，服务端在握手钩子中校验。
	Auth string
	// Retention 是服务端断连后的会话保留期；负值禁用会话恢复。
	Retention time.Duration
	// RetentionBytes 是每会话下行保留队列的字节上限，超限则失去恢复资格。
	RetentionBytes int
	// DialTimeout 是建立 TCP 连接的超时。
	DialTimeout time.Duration
	// Reconnect 为 false 时客户端不自动重连（默认 true）。
	Reconnect *bool
	// BackoffInitial / BackoffMax 是重连退避的起点与上限（指数 + 抖动）。
	BackoffInitial time.Duration
	BackoffMax     time.Duration
	// Logger 为 nil 时静默。
	Logger Logger
}

Config 是连接两侧共享的协议参数。生效值以 CONNACK 下发的为准 （服务端是权威），字段含义见 docs/protocol.md。

func DefaultConfig

func DefaultConfig() Config

DefaultConfig 返回默认配置。

type ConnectJSON

type ConnectJSON struct {
	Version     int    `json:"version"`
	Compress    bool   `json:"compress"`
	Encrypt     bool   `json:"encrypt"`
	HeartbeatMS int    `json:"heartbeat_ms"`
	Credit      int    `json:"credit,omitempty"`
	SessionID   string `json:"session_id,omitempty"` // 断线重连时携带
	Auth        string `json:"auth,omitempty"`
}

ConnectJSON 是 CONNECT 帧的 payload：客户端能力声明与恢复凭证。

type Emitter

type Emitter interface {
	Emit(v any) error
}

Emitter 由流式 handler 用于逐帧下发（受背压约束，额度耗尽时阻塞）。 Emitter 是流式 handler 的下发端：Emit 逐帧发送，handler 正常返回时 自动 COMPLETE。仅在 handler 的 goroutine 内使用（每条流一个 goroutine，见包文档「并发模型」）。

type Error

type Error struct {
	Code    ErrorCode `json:"code"`
	Message string    `json:"message,omitempty"`
}

Error 同时实现标准 error 与错误帧 payload 的编解码载体。

func (*Error) Error

func (e *Error) Error() string

func (*Error) IsProtocol

func (e *Error) IsProtocol() bool

IsProtocol 报告该错误是否为协议违规（CodeProtocol）。握手与帧层的 协议违规随后断开连接；分发层的 Flags 声明与 handler 模式不符只终结 所在流，连接继续可用（protocol.md §7.10）。

type ErrorCode

type ErrorCode uint8

ErrorCode 是错误帧 payload 中机器可读的错误分类，取值见 docs/protocol.md §7.10。

const (
	CodeInternal       ErrorCode = 1  // INTERNAL: 处理器内部错误（含 handler panic）
	CodeNotFound       ErrorCode = 2  // NOT_FOUND: 路由不存在
	CodeInvalid        ErrorCode = 3  // INVALID: 载荷或参数非法
	CodeProtocol       ErrorCode = 4  // PROTOCOL: 协议违规（必须断开连接）
	CodeCancelled      ErrorCode = 5  // CANCELLED: 对端已取消
	CodeTimeout        ErrorCode = 6  // TIMEOUT: 处理超时（v1 预留：实现未内置处理超时，无产生路径）
	CodeAuthDenied     ErrorCode = 7  // AUTH_DENIED: 鉴权失败 / 拒绝连接
	CodeSessionExpired ErrorCode = 8  // SESSION_EXPIRED: 会话过期，恢复失败
	CodeBusy           ErrorCode = 9  // BUSY: 过载拒绝（含保留队列溢出）
	CodeUnsupported    ErrorCode = 10 // UNSUPPORTED: 对端未启用的能力被错误使用
)

错误码（§7.10）：线上取值即码值，名字映射见 ErrorCode.String。

func (ErrorCode) String

func (c ErrorCode) String() string

type Frame

type Frame struct {
	Header
	Metadata []byte
	Payload  []byte
}

Frame 是内存中的帧。Metadata 恒为明文 JSON（可为 nil）； Payload 在协议栈边界内是明文 JSON，读写线上字节时才做压缩/加密变换。

func ReadFrame

func ReadFrame(r io.Reader) (*Frame, error)

ReadFrame 从 r 流式读取一帧。读入的 Payload 仍是变换后的原始字节， 由调用方（transport）负责解压解密。

func (*Frame) String

func (f *Frame) String() string

type FrameType

type FrameType uint8

FrameType 帧类型，见 docs/protocol.md §4。

const (
	TypeConnect     FrameType = 0x01
	TypeConnAck     FrameType = 0x02
	TypePing        FrameType = 0x03
	TypePong        FrameType = 0x04
	TypeRequest     FrameType = 0x05
	TypeResponse    FrameType = 0x06
	TypeComplete    FrameType = 0x07
	TypeCancel      FrameType = 0x08
	TypeOneWay      FrameType = 0x09
	TypeError       FrameType = 0x0A
	TypeSubscribe   FrameType = 0x0B
	TypeSubAck      FrameType = 0x0C
	TypeUnsubscribe FrameType = 0x0D
	TypePublish     FrameType = 0x0E
	TypeCredit      FrameType = 0x0F
)

帧类型（§4）：未知类型一律立即断开（不回帧——连帧类型都不认识， 说明对端是异版本实现），为新能力演进留位。

func (FrameType) String

func (t FrameType) String() string

type Header

type Header struct {
	Version  uint8
	Flags    uint8
	Type     FrameType
	StreamID uint32
}

Header 是帧的定长头部。

type Logger

type Logger interface {
	Printf(format string, v ...any)
}

Logger 是协议栈的日志接口，适配 *log.Logger 等常见实现。

type Message

type Message struct {
	StreamID uint32
	// Route 是 REQUEST/ONEWAY 的路由名（来自 Metadata）。
	Route string
	// Topic 是 PUBLISH/SUBSCRIBE 的主题名（来自 Metadata）。
	Topic   string
	Payload json.RawMessage
}

Message 是交付给应用的数据单元：一帧的业务语义视图。

func (*Message) Decode

func (m *Message) Decode(v any) error

Decode 把 payload 反序列化到 v；空 payload 什么都不做。

type ReadStream

type ReadStream struct {
	// contains filtered or unexported fields
}

ReadStream 是流式响应（Flags.Stream）的读取端。Next 是单消费接口： 在单个 goroutine 中循环调用（见包文档「并发模型」）。

func (*ReadStream) Cancel

func (s *ReadStream) Cancel() error

Cancel 取消流：发 CANCEL 并本地终结。帧经当前 endpoint 发送——会话恢复 迁移后是重连的新连接，构造时的 endpoint 已死（陈旧 ep 引用只会让 CANCEL 静默丢失，对端 handler 永远收不到取消）。

func (*ReadStream) Err

func (s *ReadStream) Err() error

Err 返回终结错误；正常终结返回 nil。

func (*ReadStream) Next

func (s *ReadStream) Next(ctx context.Context) (*Message, bool)

Next 返回下一帧；正常结束（COMPLETE/CANCEL）返回 false 且 Err() 为 nil， 出错返回 false 且 Err() 非 nil。

终结帧与数据帧的交付是异步的：doneCh 关闭时 frames 里可能还有余帧， 因此终结分支必须先排空 chan 再返回 false——select 双就绪是随机选择， 直接返回会随机丢帧。

type Request

type Request struct {
	*Message
	// contains filtered or unexported fields
}

Request 是路由 handler 收到的请求。Context 在所在流被取消/终结后 Done， 长处理可据此提前退出。

func (*Request) Context

func (r *Request) Context() context.Context

Context 返回与所在流绑定的 context。

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server 是 JsonStream 服务端。路由注册可并发调用，建议在 Serve 之前 完成；运行中注册对新请求立即生效（见包文档「并发模型」）。

func NewServer

func NewServer(ln net.Listener, cfg Config) (*Server, error)

NewServer 创建服务端（不开始接受连接，调用 Serve 启动）。

func (*Server) Channel

func (s *Server) Channel(ctx context.Context, sessionID, route string, payload any) (*Channel, error)

Channel 向指定会话发起双工通道。

func (*Server) Close

func (s *Server) Close() error

Close 停止监听并断开所有活跃连接。

func (*Server) Handle

func (s *Server) Handle(route string, h func(*Request) (any, error))

Handle 注册请求/响应路由。

func (*Server) HandleChannel

func (s *Server) HandleChannel(route string, h func(*Channel) error)

HandleChannel 注册双工通道路由。

func (*Server) HandleOneWay

func (s *Server) HandleOneWay(route string, h func(*Message) error)

HandleOneWay 注册单向路由（永不回帧，路由不存在时静默）。

func (*Server) HandlePublish

func (s *Server) HandlePublish(topic string, h func(*Message) error)

HandlePublish 注册客户端发布主题的处理。

func (*Server) HandleStream

func (s *Server) HandleStream(route string, h func(*Request, Emitter) error)

HandleStream 注册流式路由。

func (*Server) OnAuth

func (s *Server) OnAuth(h func(*ConnectJSON) error)

OnAuth 注册握手鉴权钩子；返回非 nil error 时以 AUTH_DENIED 拒绝连接。

func (*Server) Publish

func (s *Server) Publish(ctx context.Context, topic string, payload any) error

Publish 向所有已订阅该主题的连接广播一帧 PUBLISH。ctx 约束每目标 投递的入队等待；取消发生在扇出中途时已投递的目标不撤回，返回 ctx 错误 让调用方知道这次广播是不完整的。

func (*Server) Request

func (s *Server) Request(ctx context.Context, sessionID, route string, payload any) (*Message, error)

Request 向指定会话发起请求/响应。

func (*Server) SendOneWay

func (s *Server) SendOneWay(ctx context.Context, sessionID, route string, payload any) error

SendOneWay 向指定会话单向发送。ctx 约束发送入队等待（sessionEp 本身 非阻塞：只查锁与死连接标志）。

func (*Server) Serve

func (s *Server) Serve() error

Serve 开始接受连接，阻塞直到 Close。

func (*Server) Sessions

func (s *Server) Sessions() []string

Sessions 返回当前有活跃连接的会话 ID（升序），可作为下列方法的目标。

func (*Server) Stream

func (s *Server) Stream(ctx context.Context, sessionID, route string, payload any) (*ReadStream, error)

Stream 向指定会话发起流式响应。

type Subscription

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription 是一条主题订阅。回调错误只记日志不回帧（投递方向上 回应会形成"响应的响应"，见 docs/protocol.md §7.8）。回调在内部 goroutine 中逐帧串行执行，回调内可安全调用 Client 的其他方法。

func (*Subscription) Close

func (s *Subscription) Close() error

Close 退订：发 UNSUBSCRIBE 并终结本地流。帧必须经当前 endpoint 发送 ——会话恢复后订阅流已迁移到新连接，构造时的旧连接上发送只会得到 ErrClosed，退订帧静默丢失（服务端永远摘不掉该订阅）。