v1.3.6: stream reliability — server idle heartbeat + TS native reconnect
- handle.go: stream writer emits a blank-line heartbeat every 25s while
the business sends nothing; invisible to NDJSON parsers but resets
NAT/CDN idle timers (mobile CGNAT ~60s, Bunny ~120s)
- template_ts.go: split stream() into retry loop + streamOnce()
- dto accepts a factory re-evaluated per attempt (cursor resume)
- retry only on transport failures (status 4/5); opt-in via
StreamOptions.retry, exponential backoff with jitter, abortable
- idle watchdog (idleTimeoutMs) built on AbortSignal.timeout; no
setTimeout in generated code (simplicity gate)
- docs: README.zh.md version table for v1.3.6
This commit is contained in:
+2
-1
@@ -1,6 +1,6 @@
|
|||||||
# fun 框架(github.com/cyi-cc/fun)使用文档
|
# fun 框架(github.com/cyi-cc/fun)使用文档
|
||||||
|
|
||||||
> 适用版本:**v1.3.5**(当前最新发布)。基于 fasthttp 的单端点 RPC 框架,
|
> 适用版本:**v1.3.6**(当前最新发布)。基于 fasthttp 的单端点 RPC 框架,
|
||||||
> 自带依赖注入、Guard 鉴权、NDJSON 流式响应、自定义路由与 TypeScript 客户端生成。
|
> 自带依赖注入、Guard 鉴权、NDJSON 流式响应、自定义路由与 TypeScript 客户端生成。
|
||||||
|
|
||||||
## 版本沿革
|
## 版本沿革
|
||||||
@@ -14,6 +14,7 @@
|
|||||||
| v1.3.3 | 新增 `SetBodyLimit`:自定义路由可放宽请求体上限,支持大体积 multipart 上传 |
|
| v1.3.3 | 新增 `SetBodyLimit`:自定义路由可放宽请求体上限,支持大体积 multipart 上传 |
|
||||||
| v1.3.4 | 错误通道 API、正确性修复与服务端加固 |
|
| v1.3.4 | 错误通道 API、正确性修复与服务端加固 |
|
||||||
| v1.3.5 | 新增 `CORS`:来源白名单按需放行,预检 204 直答,覆盖 `/cell` 与全部自定义路由 |
|
| v1.3.5 | 新增 `CORS`:来源白名单按需放行,预检 204 直答,覆盖 `/cell` 与全部自定义路由 |
|
||||||
|
| v1.3.6 | 流式长连接可靠性:服务端 25s 空行心跳;TS 客户端原生重连(dto 工厂刷新游标 + 指数退避)与空闲看门狗(`retry` / `idleTimeoutMs`,默认关闭) |
|
||||||
|
|
||||||
## 1. 启动与服务注册
|
## 1. 启动与服务注册
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"runtime/debug"
|
"runtime/debug"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/valyala/fasthttp"
|
"github.com/valyala/fasthttp"
|
||||||
)
|
)
|
||||||
@@ -127,13 +128,29 @@ func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for message := range streamCh {
|
// 空闲心跳:业务长时间不 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) {
|
if !writeLine(message) {
|
||||||
finish()
|
finish()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
case <-heartbeat.C:
|
||||||
|
if _, err := w.WriteString("\n"); err != nil || w.Flush() != nil {
|
||||||
finish()
|
finish()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
})
|
})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
+76
-2
@@ -66,6 +66,23 @@ export type RequestOptions = {
|
|||||||
export type StreamOptions = {
|
export type StreamOptions = {
|
||||||
signal?: AbortSignal;
|
signal?: AbortSignal;
|
||||||
state?: Record<string, string>;
|
state?: Record<string, string>;
|
||||||
|
/**
|
||||||
|
* 原生断线重连:流因传输失败(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 = (
|
export type RequestInterceptor = (
|
||||||
@@ -342,6 +359,46 @@ export class Client {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async stream<T>(
|
async stream<T>(
|
||||||
|
serviceName: string,
|
||||||
|
methodName: string,
|
||||||
|
dto: any | undefined | (() => any | undefined),
|
||||||
|
onMessage: (data: T) => unknown,
|
||||||
|
options?: StreamOptions
|
||||||
|
): Promise<result<void>> {
|
||||||
|
// 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<T>(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<void>((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<T>(
|
||||||
serviceName: string,
|
serviceName: string,
|
||||||
methodName: string,
|
methodName: string,
|
||||||
dto: any | undefined,
|
dto: any | undefined,
|
||||||
@@ -523,13 +580,30 @@ export class Client {
|
|||||||
};
|
};
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
// 空闲看门狗:idleTimeoutMs 内未收到任何字节(含心跳空行)判定流死亡,
|
||||||
|
// 抛出哨兵错误 → status 5 → 交给上层重连。服务端 25s 心跳时建议 35000。
|
||||||
|
const STREAM_IDLE = Symbol("stream-idle");
|
||||||
for (;;) {
|
for (;;) {
|
||||||
let part: ReadableStreamReadResult<Uint8Array>;
|
let part: ReadableStreamReadResult<Uint8Array>;
|
||||||
try {
|
try {
|
||||||
|
if (options?.idleTimeoutMs) {
|
||||||
|
const idlePromise = new Promise<never>((_, 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();
|
part = await reader.read();
|
||||||
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
cause = 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");
|
failed = failure(5, "Stream timed out");
|
||||||
} else if (options?.signal?.aborted === true) {
|
} else if (options?.signal?.aborted === true) {
|
||||||
failed = failure(4, "Stream aborted");
|
failed = failure(4, "Stream aborted");
|
||||||
@@ -631,7 +705,7 @@ export default class {{.ServiceName}} {
|
|||||||
}
|
}
|
||||||
{{- $serviceName := .ServiceName }}
|
{{- $serviceName := .ServiceName }}
|
||||||
{{- range .GenMethodTypeList}}
|
{{- range .GenMethodTypeList}}
|
||||||
{{if .IsStream }}async {{.MethodName}}({{if .DtoText}}{{.DtoText}}, {{end}}onMessage: (data: {{.GenericTypeText}}) => unknown, options?: StreamOptions): Promise<result<void>> {
|
{{if .IsStream }}async {{.MethodName}}({{if .DtoText}}dto: {{.DtoText}} | (() => {{.DtoText}}), {{end}}onMessage: (data: {{.GenericTypeText}}) => unknown, options?: StreamOptions): Promise<result<void>> {
|
||||||
return await this.client.stream<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, onMessage, options)
|
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}}> {
|
}{{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)
|
return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, options)
|
||||||
|
|||||||
Reference in New Issue
Block a user