diff --git a/docs/README.zh.md b/docs/README.zh.md index ec5529a..b4001a0 100644 --- a/docs/README.zh.md +++ b/docs/README.zh.md @@ -1,6 +1,6 @@ # fun 框架(github.com/cyi-cc/fun)使用文档 -> 适用版本:**v1.3.5**(当前最新发布)。基于 fasthttp 的单端点 RPC 框架, +> 适用版本:**v1.3.6**(当前最新发布)。基于 fasthttp 的单端点 RPC 框架, > 自带依赖注入、Guard 鉴权、NDJSON 流式响应、自定义路由与 TypeScript 客户端生成。 ## 版本沿革 @@ -14,6 +14,7 @@ | v1.3.3 | 新增 `SetBodyLimit`:自定义路由可放宽请求体上限,支持大体积 multipart 上传 | | v1.3.4 | 错误通道 API、正确性修复与服务端加固 | | v1.3.5 | 新增 `CORS`:来源白名单按需放行,预检 204 直答,覆盖 `/cell` 与全部自定义路由 | +| v1.3.6 | 流式长连接可靠性:服务端 25s 空行心跳;TS 客户端原生重连(dto 工厂刷新游标 + 指数退避)与空闲看门狗(`retry` / `idleTimeoutMs`,默认关闭) | ## 1. 启动与服务注册 diff --git a/handle.go b/handle.go index ac3c78e..dba535b 100644 --- a/handle.go +++ b/handle.go @@ -9,6 +9,7 @@ import ( "runtime/debug" "strings" "sync" + "time" "github.com/valyala/fasthttp" ) @@ -127,13 +128,29 @@ func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) { return } } - for message := range streamCh { - if !writeLine(message) { - finish() - return + // 空闲心跳:业务长时间不 Send 时周期写一个空行。NDJSON 解析器 + // 跳过空行(业务无感知),字节能让 NAT/CDN 的空闲计时器归零。 + // 25s 需小于常见中间盒阈值(移动 CGNAT 约 60s、CDN 约 120s)。 + heartbeat := time.NewTicker(25 * time.Second) + defer heartbeat.Stop() + for { + select { + case message, ok := <-streamCh: + if !ok { + finish() + return + } + if !writeLine(message) { + finish() + return + } + case <-heartbeat.C: + if _, err := w.WriteString("\n"); err != nil || w.Flush() != nil { + finish() + return + } } } - finish() }) return } diff --git a/template_ts.go b/template_ts.go index 7186938..8eb8920 100644 --- a/template_ts.go +++ b/template_ts.go @@ -66,6 +66,23 @@ export type RequestOptions = { export type StreamOptions = { signal?: AbortSignal; state?: Record; + /** + * 原生断线重连:流因传输失败(status 4/5)结束时按指数退避自动重连。 + * dto 传工厂函数时每次重连重新求值,用于刷新续传游标。 + * 未传时关闭(默认),与 request 语义一致。 + */ + retry?: StreamRetryOptions | false; + /** 空闲看门狗:超过该毫秒数未收到任何字节(含心跳空行)即判定流死亡并走重连。0 关闭。 */ + idleTimeoutMs?: number; +}; + +export type StreamRetryOptions = { + /** 最大重连次数,默认 Infinity(直到 signal 取消或成功)。 */ + maxAttempts?: number; + /** 首次重连延迟(毫秒),默认 500。 */ + baseDelayMs?: number; + /** 退避上限(毫秒),默认 10000。实际延迟为指数退避 × 0.75~1.25 抖动。 */ + maxDelayMs?: number; }; export type RequestInterceptor = ( @@ -342,6 +359,46 @@ export class Client { } async stream( + serviceName: string, + methodName: string, + dto: any | undefined | (() => any | undefined), + onMessage: (data: T) => unknown, + options?: StreamOptions + ): Promise> { + // dto 工厂:重连时重新求值,用于刷新续传游标(如 afterId) + const getDto = typeof dto === "function" ? (dto as () => any | undefined) : () => dto; + // 重连默认关闭(与 request 语义一致);传 retry: {...} 显式开启 + const retry = options?.retry + ? { + maxAttempts: Infinity, + baseDelayMs: 500, + maxDelayMs: 10000, + ...options.retry, + } + : null; + for (let attempt = 0; ; attempt++) { + const r = await this.streamOnce(serviceName, methodName, getDto(), onMessage, options); + // 0 成功、1 协议错误、2 业务错误:终结态,不重连 + if (r.status !== 4 && r.status !== 5) return r; + if (!retry) return r; + if (options?.signal?.aborted) return r; + if (attempt + 1 > retry.maxAttempts) return r; + const delay = + Math.min(retry.maxDelayMs, retry.baseDelayMs * 2 ** attempt) * (0.75 + Math.random() * 0.5); + // 退避等待:AbortSignal.timeout 免定时器 API,且后台标签页照常走时 + await new Promise((resolve) => { + if (options?.signal?.aborted) { + resolve(); + return; + } + AbortSignal.timeout(delay).addEventListener("abort", () => resolve(), { once: true }); + options?.signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + if (options?.signal?.aborted) return r; + } + } + + private async streamOnce( serviceName: string, methodName: string, dto: any | undefined, @@ -523,13 +580,30 @@ export class Client { }; try { + // 空闲看门狗:idleTimeoutMs 内未收到任何字节(含心跳空行)判定流死亡, + // 抛出哨兵错误 → status 5 → 交给上层重连。服务端 25s 心跳时建议 35000。 + const STREAM_IDLE = Symbol("stream-idle"); for (;;) { let part: ReadableStreamReadResult; try { - part = await reader.read(); + if (options?.idleTimeoutMs) { + const idlePromise = new Promise((_, reject) => { + AbortSignal.timeout(options.idleTimeoutMs!).addEventListener( + "abort", + () => reject(STREAM_IDLE), + { once: true } + ); + }); + idlePromise.catch(() => {}); // read 先返回时防 unhandled rejection + part = await Promise.race([reader.read(), idlePromise]); + } else { + part = await reader.read(); + } } catch (error) { cause = error; - if (isTimeout(error, options?.signal)) { + if (error === STREAM_IDLE) { + failed = failure(5, "Stream idle timeout"); + } else if (isTimeout(error, options?.signal)) { failed = failure(5, "Stream timed out"); } else if (options?.signal?.aborted === true) { failed = failure(4, "Stream aborted"); @@ -631,7 +705,7 @@ export default class {{.ServiceName}} { } {{- $serviceName := .ServiceName }} {{- range .GenMethodTypeList}} - {{if .IsStream }}async {{.MethodName}}({{if .DtoText}}{{.DtoText}}, {{end}}onMessage: (data: {{.GenericTypeText}}) => unknown, options?: StreamOptions): Promise> { + {{if .IsStream }}async {{.MethodName}}({{if .DtoText}}dto: {{.DtoText}} | (() => {{.DtoText}}), {{end}}onMessage: (data: {{.GenericTypeText}}) => unknown, options?: StreamOptions): Promise> { return await this.client.stream<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, onMessage, options) }{{else}}async {{.MethodName}}({{if .DtoText}}{{.DtoText}}, {{end}}options?: RequestOptions): Promise<{{.ReturnValueText}}> { return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, options)