v1.4.0: generic Stream[T] — typed stream messages, drop (T, *Stream, error)

- Stream[T]: Send(v T) type-safe; message type recovered from the instantiated Send signature (reflect has no type-args API) - method signatures reduced to (error), (T, error), (*Stream[T], error); the (T, *Stream, error) first-message form is gone — send it via Send - return T and stream message T now run checkType at bind time (Stream[any] remains the explicit untyped escape hatch) - generated clients: TS onMessage: (data: T); Go <-chan T - fix vacuous panic assertions (fmt.Sprint(nil) -> "<nil>") in hardening/route test catch helpers Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
fix(stream): OnClose 可在返回前注册，消除 Inject 循环等待死锁
2026-09-19 15:39:41 +08:00 · 2026-09-16 13:54:00 +08:00 · 2026-09-11 18:56:48 +08:00 · 2026-09-09 11:31:22 +08:00
18 changed files with 570 additions and 130 deletions
@@ -6,12 +6,13 @@ NDJSON 流式响应、自定义路由与 TypeScript 客户端生成。

 ## 特性

- **单端点 RPC**：`POST /cell`，方法签名 `(error)`、`(T, error)`、`(stream, error)`、`(T, stream, error)`
+- **单端点 RPC**：`POST /cell`，方法签名 `(error)`、`(T, error)`、`(*Stream[T], error)`
 - **依赖注入**：`fun.Wired[T]()` 建单例，`auto` 标签字段递归装配，`New()` 初始化连接资源
 - **Guard 鉴权**：全局 + 服务级中间件，panic 兜底转统一错误响应
- **NDJSON 流式**：`*fun.Stream` 逐行推送，支持首条消息 + 后续流
+- **NDJSON 流式**：`*fun.Stream[T]` 逐行推送，消息类型由类型实参声明
 - **自定义路由**（v1.3.0+）：`BindRoute` 注册 GET/POST 回调、健康检查、通配符路径
 - **请求体上限控制**（v1.3.3+）：`SetBodyLimit` 支持大体积 multipart 上传
+- **CORS 跨域**（v1.3.5+）：`f.CORS(origins ...string)` 白名单按需放行，预检 204 直答，覆盖 `/cell` 与全部自定义路由
 - **TypeScript 客户端生成**：`BindServiceForGen` + `GenCode(fun.GenTs{})`，免基础设施即可生成，产物带 `result<T>` 归一化错误与拦截器

 ## 快速开始
@@ -27,13 +27,14 @@ func (s *BugSvc) Ping() error { return nil }

 func (s *BugSvc) Save(dto BugDto) (string, error) { return "ok", nil }

-func (s *BugSvc) Ticker() (string, *Stream, error) {
-	st := Stream{}
+func (s *BugSvc) Ticker() (*Stream[string], error) {
+	st := &Stream[string]{}
 	go func() {
+		st.Send("first")
 		st.Send("tick")
 		st.Close()
 	}()
-	return "first", &st, nil
+	return st, nil
 }

 func bugInvoke(t *testing.T, method string, data map[string]any) (*Result[any], error) {
@@ -99,7 +100,7 @@ func TestBugGenErrorOnly(t *testing.T) {
 	}
 }

-// bug4+5: 响应键应为小写；(T, stream, error) 的 T 应作为流的第一条消息下发
+// bug4+5: 响应键应为小写；流消息按 Send 顺序逐行下发
 func TestBugJsonKeysAndStreamFirst(t *testing.T) {
 	f := New()
 	if err := f.BindService(&BugSvc{}); err != nil {
@@ -173,8 +174,8 @@ var leakDone chan struct{}

 type LeakSvc struct{}

-func (s *LeakSvc) Fail() (*Stream, error) {
-	st := Stream{}
+func (s *LeakSvc) Fail() (*Stream[any], error) {
+	st := Stream[any]{}
 	leakDone = make(chan struct{})
 	go func() {
 		st.Send("never")
@@ -0,0 +1,71 @@
+package fun
+
+import (
+	"strings"
+
+	"github.com/valyala/fasthttp"
+)
+
+const (
+	corsMaxAge = "86400" // 预检结果缓存 24h，减少浏览器 OPTIONS 探测
+)
+
+// CORS 配置跨域来源白名单，须在 Start 前调用；作用于全部端点（/cell、自定义路由与通配路由）。
+// 按需放行：仅列出的来源会被放行——请求 Origin 命中白名单时回显该来源并允许
+// 携带凭据（Cookie），未命中则不附加 CORS 头（浏览器侧自然拦截）。
+//
+//	f.CORS("https://a.com", "https://b.com")
+//
+// 预检请求（OPTIONS 且带 Access-Control-Request-Method 头）直接以 204 应答，
+// 不进入路由与 /cell 处理。
+func (f *Fun) CORS(origins ...string) {
+	f.mustNotStarted("CORS")
+	whitelist := make(map[string]struct{}, len(origins))
+	for _, o := range origins {
+		if o != "" {
+			whitelist[strings.ToLower(o)] = struct{}{}
+		}
+	}
+	f.corsOrigins = whitelist
+}
+
+// corsEnabled CORS 是否已配置
+func (f *Fun) corsEnabled() bool {
+	return len(f.corsOrigins) > 0
+}
+
+// handleCors 处理 CORS：Origin 命中白名单的请求附加跨域响应头，
+// 预检请求（OPTIONS + Access-Control-Request-Method）在此直接应答并返回 true。
+func (f *Fun) handleCors(fastCtx *fasthttp.RequestCtx) bool {
+	if !f.corsEnabled() {
+		return false
+	}
+	origin := string(fastCtx.Request.Header.Peek("Origin"))
+	if origin == "" {
+		return false // 同源请求与服务器到服务器调用，无跨域语义
+	}
+	if _, ok := f.corsOrigins[strings.ToLower(origin)]; !ok {
+		return false // 未命中白名单：不加 CORS 头，交给浏览器拦截
+	}
+
+	h := &fastCtx.Response.Header
+	h.Set("Access-Control-Allow-Origin", origin)
+	h.Set("Access-Control-Allow-Credentials", "true")
+	// 放行结果随 Origin 变化，防中间层缓存错配
+	h.Add("Vary", "Origin")
+
+	if !fastCtx.IsOptions() || fastCtx.Request.Header.Peek("Access-Control-Request-Method") == nil {
+		return false // 实际请求：头已附加，继续正常路由
+	}
+
+	// 预检请求：回显浏览器声明的目标方法与请求头，24h 内同请求免预检
+	h.Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
+	if reqHeaders := fastCtx.Request.Header.Peek("Access-Control-Request-Headers"); len(reqHeaders) > 0 {
+		h.Set("Access-Control-Allow-Headers", string(reqHeaders))
+	} else {
+		h.Set("Access-Control-Allow-Headers", "Content-Type")
+	}
+	h.Set("Access-Control-Max-Age", corsMaxAge)
+	fastCtx.SetStatusCode(fasthttp.StatusNoContent)
+	return true
+}
@@ -0,0 +1,134 @@
+package fun
+
+import (
+	"bytes"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+)
+
+func startCorsServer(t *testing.T, port uint16, withCors bool) *Fun {
+	t.Helper()
+	f := New()
+	if err := f.BindService(&TestSvc{}); err != nil {
+		t.Fatal(err)
+	}
+	if withCors {
+		f.CORS("https://a.com", "HTTPS://B.Com") // 大小写归一化
+	}
+	f.BindRoute("GET", "/ping", func(ctx *RouteCtx) error {
+		ctx.RequestCtx.WriteString("pong")
+		return nil
+	})
+	go f.Start(port)
+	time.Sleep(300 * time.Millisecond)
+	return f
+}
+
+func corsGet(t *testing.T, url, origin string) *http.Response {
+	t.Helper()
+	req, _ := http.NewRequest("GET", url, nil)
+	if origin != "" {
+		req.Header.Set("Origin", origin)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return resp
+}
+
+func TestCorsWhitelistEcho(t *testing.T) {
+	startCorsServer(t, 39201, true)
+	resp := corsGet(t, "http://127.0.0.1:39201/ping", "https://a.com")
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d", resp.StatusCode)
+	}
+	if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "https://a.com" {
+		t.Fatalf("allow-origin = %q, want echo", got)
+	}
+	if got := resp.Header.Get("Access-Control-Allow-Credentials"); got != "true" {
+		t.Fatalf("allow-credentials = %q", got)
+	}
+	if got := resp.Header.Get("Vary"); !strings.Contains(got, "Origin") {
+		t.Fatalf("vary = %q, want Origin", got)
+	}
+
+	// 白名单大小写不敏感
+	resp2 := corsGet(t, "http://127.0.0.1:39201/ping", "https://B.COM")
+	defer resp2.Body.Close()
+	if got := resp2.Header.Get("Access-Control-Allow-Origin"); got != "https://B.COM" {
+		t.Fatalf("allow-origin = %q, want case-insensitive match", got)
+	}
+}
+
+func TestCorsOriginNotWhitelisted(t *testing.T) {
+	startCorsServer(t, 39202, true)
+	resp := corsGet(t, "http://127.0.0.1:39202/ping", "https://evil.com")
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d（未命中白名单请求本身仍应正常处理）", resp.StatusCode)
+	}
+	if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "" {
+		t.Fatalf("allow-origin = %q, want empty", got)
+	}
+}
+
+func TestCorsPreflight(t *testing.T) {
+	startCorsServer(t, 39203, true)
+	req, _ := http.NewRequest("OPTIONS", "http://127.0.0.1:39203/cell", nil)
+	req.Header.Set("Origin", "https://a.com")
+	req.Header.Set("Access-Control-Request-Method", "POST")
+	req.Header.Set("Access-Control-Request-Headers", "content-type, x-custom")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNoContent {
+		t.Fatalf("status = %d, want 204", resp.StatusCode)
+	}
+	if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "https://a.com" {
+		t.Fatalf("allow-origin = %q", got)
+	}
+	if got := resp.Header.Get("Access-Control-Allow-Methods"); !strings.Contains(got, "POST") {
+		t.Fatalf("allow-methods = %q, want POST", got)
+	}
+	// 请求头回显
+	if got := resp.Header.Get("Access-Control-Allow-Headers"); !strings.Contains(got, "x-custom") {
+		t.Fatalf("allow-headers = %q, want echo x-custom", got)
+	}
+	if got := resp.Header.Get("Access-Control-Max-Age"); got != "86400" {
+		t.Fatalf("max-age = %q", got)
+	}
+}
+
+func TestCorsHeadersOnCellRpc(t *testing.T) {
+	startCorsServer(t, 39204, true)
+	body := `{"ServiceName":"TestSvc","MethodName":"Hello","Data":{"Name":"tom","Age":1}}`
+	req, _ := http.NewRequest("POST", "http://127.0.0.1:39204/cell", bytes.NewBufferString(body))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Origin", "https://a.com")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d", resp.StatusCode)
+	}
+	if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "https://a.com" {
+		t.Fatalf("/cell 响应缺少 CORS 头: %q", got)
+	}
+}
+
+func TestCorsNotConfigured(t *testing.T) {
+	startCorsServer(t, 39205, false)
+	resp := corsGet(t, "http://127.0.0.1:39205/ping", "https://a.com")
+	defer resp.Body.Close()
+	if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "" {
+		t.Fatalf("未配置 CORS 却有 allow-origin = %q", got)
+	}
+}
@@ -1,6 +1,6 @@
 # fun 框架（github.com/cyi-cc/fun）使用文档

-> 适用版本：**v1.3.3**（当前最新发布）。基于 fasthttp 的单端点 RPC 框架，
+> 适用版本：**v1.3.6**（当前最新发布）。基于 fasthttp 的单端点 RPC 框架，
 > 自带依赖注入、Guard 鉴权、NDJSON 流式响应、自定义路由与 TypeScript 客户端生成。

 ## 版本沿革
@@ -12,6 +12,9 @@
 | v1.3.1 | TS 客户端可靠性：所有失败统一归一为 Result 并经过响应拦截器 |
 | v1.3.2 | 每请求上下文（request/stream options + `state`）与免基础设施的生成期注册 `BindServiceForGen` |
 | v1.3.3 | 新增 `SetBodyLimit`：自定义路由可放宽请求体上限，支持大体积 multipart 上传 |
+| v1.3.4 | 错误通道 API、正确性修复与服务端加固 |
+| v1.3.5 | 新增 `CORS`：来源白名单按需放行，预检 204 直答，覆盖 `/cell` 与全部自定义路由 |
+| v1.3.6 | 流式长连接可靠性：服务端 25s 空行心跳；TS 客户端原生重连（dto 工厂刷新游标 + 指数退避）与空闲看门狗（`retry` / `idleTimeoutMs`，默认关闭） |

 ## 1. 启动与服务注册

@@ -28,8 +31,8 @@ func main() {

 - 服务结构体嵌入 `fun.Ctx` + 依赖字段（指针结构体字段自动装配）。
 - **每请求新建服务实例**并注入依赖，服务内不放共享状态。
- 方法签名四种：`() error`、`(dto) (T, error)`、`(dto) (*fun.Stream, error)`、
-  `(dto) (T, *fun.Stream, error)`（首条消息 T + 后续流）。
+- 方法签名三种：`() error`、`(dto) (T, error)`、`(dto) (*fun.Stream[T], error)`。
+  流式消息类型由 `Stream[T]` 的类型实参声明（`Stream[any]` 为任意消息流）。
 - 只有导出方法成为端点，注册名 `服务名.方法名`。

 ## 2. DTO 规则（违反即注册期 panic）
@@ -92,10 +95,25 @@ f.BindRoute("POST", "/upload", func(c *fun.RouteCtx) error {
 })
 ```

-## 6. 流式响应（NDJSON）
+## 6. CORS 跨域

 ```go
-st := &fun.Stream{}
+f.CORS("https://a.com", "https://b.com") // Start 前调用，可变参数白名单
+```
+
+- **按需放行**：请求 Origin 命中白名单（大小写不敏感）才回显该来源并允许
+  携带凭据（`Access-Control-Allow-Credentials: true`）；未命中不附加任何
+  CORS 头，交给浏览器自然拦截。
+- **覆盖全部端点**：`/cell`、自定义路由与通配路由统一生效；预检请求
+  （OPTIONS + `Access-Control-Request-Method`）由框架直接以 204 应答，
+  不进入业务处理；请求头按 `Access-Control-Request-Headers` 回显，
+  预检结果缓存 24h。
+- 未调用 `CORS` 则不附加任何跨域头，行为与之前完全一致。
+
+## 7. 流式响应（NDJSON）
+
+```go
+st := &fun.Stream[string]{}
 go func() {
    for _, chunk := range chunks {
        if st.Send(chunk) != nil { return } // 连接断开
@@ -108,7 +126,7 @@ return st, nil
 `Content-Type: application/x-ndjson`，每行一个 JSON；合法零消息流正常结束；
 业务出错在建流前返回普通 Result；`OnClose` 注册清理回调。

-## 7. TS 客户端生成与请求上下文（v1.3.2 核心）
+## 8. TS 客户端生成与请求上下文（v1.3.2 核心）

 ```go
 f := fun.GetFun()
@@ -153,17 +171,17 @@ c.addResponseInterceptor((svc, m, result, context) => {
 旧三参签名仍兼容。所有失败（网络、HTTP、HTML、非法 JSON、取消、拦截器异常）
 统一归一为 Result 并**必经响应拦截器**，不存在绕过拦截器的错误路径。

-## 8. 依赖注入（box.go）
+## 9. 依赖注入（box.go）

 - `fun.Wired[T]()`：按 `*T` 建单例；先注入 `fun:"auto"` 字段（缺则递归创建），
  再调 `New()`（无参；连接类资源在此初始化，失败可 log.Fatalf）。
 - 启动顺序：先 `Wired` 基础配置/平台单例，再 `BindService`。

-## 9. 常见坑
+## 10. 常见坑

 - DTO 用普通 `int`/float/map → 注册期 panic；用 int64/字符串/指针。
 - 非指针字段漏传 → 运行期 "must be a pointer or have a corresponding field"。
 - 流式忘记 `Close()` → 客户端挂起；连接断开后 `Send` 返回 error，循环须检查。
 - 方法首字母小写 = 不注册；客户端报 method not found。
- vite 代理需重写前缀：`/api/cell → /cell`。
+- vite 代理需重写前缀：`/api/cell → /cell`；或服务端 `f.CORS` 加白名单放开浏览器直连。
 - 生成用 `BindService`（而非 `BindServiceForGen`）会把基础设施拉起来 —— 生成命令请用后者。
@@ -75,14 +75,14 @@ type AskDto struct {
 	Prompt string
 }

-// ChatSvc 流式服务：纯流 + 首条消息流两种签名
+// ChatSvc 流式服务：消息类型由 *Stream[T] 的类型实参声明
 type ChatSvc struct {
 	fun.Ctx
 }

 // Chat 纯流式：响应为 application/x-ndjson，逐行推送
-func (s *ChatSvc) Chat(dto ChatDto) (*fun.Stream, error) {
-	st := &fun.Stream{}
+func (s *ChatSvc) Chat(dto ChatDto) (*fun.Stream[string], error) {
+	st := &fun.Stream[string]{}
 	go func() {
 		for _, chunk := range []string{"你好", "，这是", "流式示例"} {
 			if err := st.Send(chunk); err != nil {
@@ -94,12 +94,15 @@ func (s *ChatSvc) Chat(dto ChatDto) (*fun.Stream, error) {
 	return st, nil
 }

-// Ask (T, stream, error)：T 作为流的第一条消息下发
-func (s *ChatSvc) Ask(dto AskDto) (string, *fun.Stream, error) {
-	st := &fun.Stream{}
+// Ask 回执也经流下发：先推一条确认，再推思考过程
+func (s *ChatSvc) Ask(dto AskDto) (*fun.Stream[string], error) {
+	st := &fun.Stream[string]{}
 	go func() {
+		if err := st.Send("收到：" + dto.Prompt); err != nil {
+			return
+		}
 		_ = st.Send("思考中...")
 		st.Close()
 	}()
-	return "收到：" + dto.Prompt, st, nil
+	return st, nil
 }
@@ -25,6 +25,8 @@ type Fun struct {
 	idleTimeout    time.Duration // keep-alive 空闲超时，默认 120s
 	maxConcurrency int           // 最大并发连接数；0 = 不限制

+	corsOrigins map[string]struct{} // CORS 来源白名单（小写比较）；nil/空表示未开启
+
 	server  atomic.Pointer[fasthttp.Server]
 	started atomic.Bool

@@ -68,7 +70,6 @@ type wildcardRoute struct {

 var (
 	errorType = reflect.TypeFor[error]()
-	streamType = reflect.TypeFor[*Stream]()
 )

 var (
@@ -81,7 +82,7 @@ type methodInfo struct {
 	serviceType reflect.Type // 服务值类型（非指针），每请求新建实例
 	methodIndex int          // 方法在实例上的反射索引
 	dtoType     reflect.Type // DTO 参数类型，无参数时为 nil
-	isStream    bool         // 返回签名带 *Stream，响应走 NDJSON 流式
+	isStream    bool         // 返回签名带 *Stream[T]，响应走 NDJSON 流式
 }

 func newFun() *Fun {
@@ -120,7 +121,7 @@ func GetFun() *Fun {
 // BindService 注册服务，要求传入指向结构体的指针
 // 方法签名约束：
 //   - 参数：最多一个，且必须是 struct（作为 DTO）
-//   - 返回值：只支持四种签名——(error)、(T, error)、(stream, error)、(T, stream, error)
+//   - 返回值：只支持三种签名——(error)、(T, error)、(*Stream[T], error)
 //
 // guardList 为该服务绑定的 Guard，方法调用前按注册顺序执行。
 // 依赖装配失败（New() 返回 error）以 error 返回，由调用方决定退出或降级；
@@ -186,31 +187,34 @@ func (f *Fun) bindServiceMethods(t reflect.Type, name string) {
 			checkType(dtoType)
 		}

-		// 返回值只支持四种签名：error / (T, error) / (stream, error) / (T, stream, error)
+		// 返回值只支持三种签名：error / (T, error) / (*Stream[T], error)
 		isStream := false
 		switch mt.NumOut() {
 		case 1:
 			// 情况 1：func(...) error
 			if mt.Out(0) != errorType {
-				panic(fmt.Sprintf("fun: method %s must return (error), (T, error), (stream, error) or (T, stream, error)", m.Name))
+				panic(fmt.Sprintf("fun: method %s must return (error), (T, error) or (*Stream[T], error)", m.Name))
 			}
 		case 2:
-			// 情况 2：func(...) (T, error) 或 func(...) (*Stream, error)
+			// 情况 2：func(...) (T, error) 或 func(...) (*Stream[T], error)
 			if mt.Out(1) != errorType {
 				panic(fmt.Sprintf("fun: method %s last return value must be error", m.Name))
 			}
-			isStream = mt.Out(0) == streamType
-		case 3:
-			// 情况 3：func(...) (T, *Stream, error)
-			if mt.Out(2) != errorType {
-				panic(fmt.Sprintf("fun: method %s last return value must be error", m.Name))
-			}
-			if mt.Out(1) != streamType {
-				panic(fmt.Sprintf("fun: method %s second return value must be *Stream", m.Name))
-			}
-			isStream = true
+			isStream = streamMsgType(mt.Out(0)) != nil
 		default:
-			panic(fmt.Sprintf("fun: method %s must return (error), (T, error), (stream, error) or (T, stream, error)", m.Name))
+			panic(fmt.Sprintf("fun: method %s must return (error), (T, error) or (*Stream[T], error)", m.Name))
+		}
+
+		// 数据返回 T 与 DTO 同受类型系统约束；流式校验消息类型 T
+		// （Stream[any] 是显式的任意消息流，客户端生成 data: any）
+		if mt.NumOut() == 2 {
+			if msgT := streamMsgType(mt.Out(0)); msgT != nil {
+				if msgT.Kind() != reflect.Interface {
+					checkType(msgT)
+				}
+			} else {
+				checkType(mt.Out(0))
+			}
 		}

 		// 注册到 "ServiceName.MethodName"
@@ -44,8 +44,8 @@ func (s *TestSvc) Hello(dto TestDto) (string, error) {
 	return "hi " + dto.Name, nil
 }

-func (s *TestSvc) Count(dto TestDto) (*Stream, error) {
-	st := Stream{}
+func (s *TestSvc) Count(dto TestDto) (*Stream[string], error) {
+	st := &Stream[string]{}
 	go func() {
 		for i := 0; i < 3; i++ {
 			st.Send(fmt.Sprintf("n%d", i))
@@ -53,7 +53,7 @@ func (s *TestSvc) Count(dto TestDto) (*Stream, error) {
 		}
 		st.Close()
 	}()
-	return &st, nil
+	return st, nil
 }

 func TestCtxBoxInject(t *testing.T) {
@@ -58,18 +58,19 @@ func (ctx GenGo) genService(svc *genSvc, serviceContext *genServiceType) {
 		returnType := gm.sig.Out(0)
 		if gm.isStream {
 			serviceContext.IsIncludeStream = true
-			if returnType == streamType {
-				genericTypeText = "any"
 			returnValueText = "Void"
-			} else {
-				t := ctx.typeToTemplateType(returnType)
+			if msgT := streamMsgType(returnType); msgT != nil && msgT.Kind() != reflect.Interface {
+				t := ctx.typeToTemplateType(msgT)
 				if !strings.Contains(t, "[]") && strings.Contains(t, "[") {
 					genericTypeText = getGenericTypeName(t) + parseGenericTypeParams(t)
 				} else {
 					genericTypeText = t
 				}
 				returnValueText = genericTypeText
-				ctx.genReturnTypes(returnType)
+				ctx.genReturnTypes(msgT)
+			} else {
+				// Stream[any]：消息类型由运行期决定，客户端通道给 any
+				genericTypeText = "any"
 			}
 		} else {
 			t := ctx.typeToTemplateType(returnType)
@@ -26,11 +26,11 @@ func (*AlphaGenSvc) Ping() error                                { return nil }
 type MixedGenSvc struct{}

 func (*MixedGenSvc) Request() (string, error)      { return "", nil }
-func (*MixedGenSvc) Stream() (*Stream, error) { return &Stream{}, nil }
+func (*MixedGenSvc) Stream() (*Stream[any], error) { return &Stream[any]{}, nil }

 type ZebraGenSvc struct{}

-func (*ZebraGenSvc) Watch() (*Stream, error) { return &Stream{}, nil }
+func (*ZebraGenSvc) Watch() (*Stream[any], error) { return &Stream[any]{}, nil }

 type GenOnlyDependency struct{}

@@ -70,18 +70,18 @@ func (ctx GenTs) genService(svc *genSvc, serviceContext *genServiceType) {
 		returnType := gm.sig.Out(0)
 		if gm.isStream {
 			serviceContext.IsIncludeStream = true
-			if returnType == streamType {
-				genericTypeText = "any"
 			returnValueText = "void"
-			} else {
-				t := firstLetterToLower(ctx.typeToTemplateType(returnType))
+			if msgT := streamMsgType(returnType); msgT != nil && msgT.Kind() != reflect.Interface {
+				t := firstLetterToLower(ctx.typeToTemplateType(msgT))
 				if !strings.Contains(t, "[]") && strings.Contains(t, "[") {
 					genericTypeText = getGenericTypeName(t) + parseGenericTypeParams(t)
 				} else {
 					genericTypeText = t
 				}
-				returnValueText = "void"
-				nestedImports = ctx.genReturnTypes(returnType, nestedImports)
+				nestedImports = ctx.genReturnTypes(msgT, nestedImports)
+			} else {
+				// Stream[any]：消息类型由运行期决定，客户端回调给 any
+				genericTypeText = "any"
 			}
 		} else {
 			serviceContext.IsIncludeRequest = true
@@ -9,6 +9,7 @@ import (
 	"runtime/debug"
 	"strings"
 	"sync"
+	"time"

 	"github.com/valyala/fasthttp"
 )
@@ -18,6 +19,11 @@ func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) {
 	ctx := &Ctx{RequestCtx: fastCtx}
 	defer f.handlePanic(ctx)

+	// CORS 置于路由之前：预检请求在此直接应答，实际请求附加跨域头后继续正常分发
+	if f.handleCors(fastCtx) {
+		return
+	}
+
 	method, path := string(fastCtx.Method()), string(fastCtx.Path())
 	if r, ok := f.routes[method+" "+path]; ok {
 		f.handleRoute(fastCtx, r, "")
@@ -71,7 +77,7 @@ func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) {
 	}

 	// 流式方法：响应保持打开，以 NDJSON 行推送（Streamable HTTP）
-	// streamCh != nil 表示流式方法；业务返回的 *Stream 在 invoke 内完成注入
+	// streamCh != nil 表示流式方法；业务返回的 *Stream[T] 在 invoke 内完成注入
 	var streamCh chan any
 	var streamDone chan struct{}

@@ -115,20 +121,29 @@ func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) {
 				}
 				return true
 			}
-			// (T, stream, error)：T 作为流的第一条消息下发
-			if result.Data != nil {
-				if !writeLine(*result.Data) {
+			// 空闲心跳：业务长时间不 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
 					}
-			}
-			for message := range streamCh {
 					if !writeLine(message) {
 						finish()
 						return
 					}
-			}
+				case <-heartbeat.C:
+					if _, err := w.WriteString("\n"); err != nil || w.Flush() != nil {
 						finish()
+						return
+					}
+				}
+			}
 		})
 		return
 	}
@@ -157,7 +172,7 @@ func (f *Fun) handlePanic(c *Ctx) {
 }

 // invoke 按 "Service.Method" 查找并调用，返回成功结果
-// 流式方法时，业务返回的 *Stream 完成通道注入（streamCh/streamDone 被创建并填充）
+// 流式方法时，业务返回的 *Stream[T] 完成通道注入（streamCh/streamDone 被创建并填充）
 // 预期错误（方法不存在、参数缺失、业务失败）以 error 返回，不 panic
 func (f *Fun) invoke(c *Ctx, streamCh *chan any, streamDone *chan struct{}) (*Result[any], error) {
 	key := c.ServiceName + "." + c.MethodName
@@ -201,10 +216,15 @@ func (f *Fun) invoke(c *Ctx, streamCh *chan any, streamDone *chan struct{}) (*Re
 	return callResult(c, values, method, streamCh, streamDone)
 }

+// streamInjector 流句柄的注入协议面：*Stream[T] 对任意 T 都满足
+type streamInjector interface {
+	Inject(chan any, chan struct{})
+}
+
 // callResult 将反射调用结果归一为 Result
-// 兼容四种签名：(error)、(T, error)、(stream, error)、(T, stream, error)
+// 兼容三种签名：(error)、(T, error)、(*Stream[T], error)
 //   - 末位返回值是 error 且非 nil → 业务失败，返回 error
-//   - 带 *Stream 的签名：注入推送通道后，仅 (T, stream, error) 返回 T 作为数据
+//   - *Stream[T] 签名：注入推送通道，数据全经流推送
 func callResult(c *Ctx, values []reflect.Value, method methodInfo, streamCh *chan any, streamDone *chan struct{}) (*Result[any], error) {
 	if last := values[len(values)-1]; last.Type().Implements(errorType) {
 		if !last.IsNil() {
@@ -224,23 +244,14 @@ func callResult(c *Ctx, values []reflect.Value, method methodInfo, streamCh *cha
 	}

 	if method.isStream {
-		// (stream, error)：流在第 0 位；(T, stream, error)：流在第 1 位
-		streamIdx := 0
-		if len(values) == 2 {
-			streamIdx = 1
-		}
-		s := values[streamIdx].Interface().(*Stream)
-		if s == nil {
+		// (*Stream[T], error)：流固定在第 0 位
+		s, ok := values[0].Interface().(streamInjector)
+		if !ok || values[0].IsNil() {
 			return nil, errors.New("fun: method returned nil stream")
 		}
 		*streamCh = make(chan any)
 		*streamDone = make(chan struct{})
 		s.Inject(*streamCh, *streamDone)
-		// (T, stream, error)：返回 T；纯流：不返回数据
-		if len(values) == 2 {
-			r := success(values[0].Interface())
-			return &r, nil
-		}
 		r := success(nil)
 		return &r, nil
 	}
@@ -257,12 +268,8 @@ func callResult(c *Ctx, values []reflect.Value, method methodInfo, streamCh *cha
 // injectCancelledStream 业务出错时注入已取消的流通道，
 // 使正在 Send/Close 上阻塞的业务 goroutine 立即解除并退出
 func injectCancelledStream(values []reflect.Value, method methodInfo, streamCh *chan any, streamDone *chan struct{}) {
-	streamIdx := 0
-	if len(values) == 3 { // (T, stream, error)
-		streamIdx = 1
-	}
-	s, ok := values[streamIdx].Interface().(*Stream)
-	if !ok || s == nil {
+	s, ok := values[0].Interface().(streamInjector)
+	if !ok || values[0].IsNil() {
 		return
 	}
 	*streamCh = make(chan any)
@@ -185,7 +185,11 @@ func TestStartedGuardAndShutdown(t *testing.T) {
 	}

 	catch := func(fn func()) (msg string) {
-		defer func() { msg = fmt.Sprint(recover()) }()
+		defer func() {
+			if r := recover(); r != nil {
+				msg = fmt.Sprint(r)
+			}
+		}()
 		fn()
 		return ""
 	}
@@ -248,8 +252,8 @@ func (boomMarshaler) MarshalJSON() ([]byte, error) { panic("boom-json") }

 type PanicStreamSvc struct{}

-func (s *PanicStreamSvc) Go() (*Stream, error) {
-	st := &Stream{}
+func (s *PanicStreamSvc) Go() (*Stream[any], error) {
+	st := &Stream[any]{}
 	go func() {
 		_ = st.Send(boomMarshaler{})
 		st.Close()
@@ -341,3 +345,52 @@ if (!requestBody.includes('"id":42')) throw new Error("small int broken: " + req
 		t.Fatalf("bigint round trip failed: %v\n%s", err, output)
 	}
 }
+
+// ---- 返回类型 T 与 DTO 同受类型系统约束（含 Stream[T] 的消息类型） ----
+
+type BadRetOnly struct{}
+
+func (*BadRetOnly) Ok() (string, error) { return "ok", nil }
+
+type mapRetSvc struct{}
+
+func (*mapRetSvc) M() (map[string]any, error) { return nil, nil }
+
+type anyRetSvc struct{}
+
+func (*anyRetSvc) M() (any, error) { return nil, nil }
+
+type badStreamSvc struct{}
+
+func (*badStreamSvc) M() (*Stream[map[string]any], error) { return nil, nil }
+
+type anyStreamSvc struct{}
+
+func (*anyStreamSvc) M() (*Stream[any], error) { return &Stream[any]{}, nil }
+
+func TestReturnTypeCheckedAtBind(t *testing.T) {
+	panics := func(fn func()) (p bool) {
+		defer func() { p = recover() != nil }()
+		fn()
+		return false
+	}
+	if panics(func() { _ = New().BindService(&BadRetOnly{}) }) {
+		t.Fatal("normal service must not panic")
+	}
+	for _, tc := range []struct {
+		name string
+		svc  any
+	}{
+		{"map return", &mapRetSvc{}},
+		{"any return", &anyRetSvc{}},
+		{"Stream[map]", &badStreamSvc{}},
+	} {
+		if !panics(func() { _ = New().BindService(tc.svc) }) {
+			t.Fatalf("%s must panic at bind", tc.name)
+		}
+	}
+	// Stream[any] 显式任意消息流：放行
+	if panics(func() { _ = New().BindService(&anyStreamSvc{}) }) {
+		t.Fatal("Stream[any] should pass")
+	}
+}
@@ -120,7 +120,11 @@ func TestBindRouteErrorAndCellUnaffected(t *testing.T) {
 func TestBindRoutePanics(t *testing.T) {
 	f := New()
 	catch := func(fn func()) (msg string) {
-		defer func() { msg = fmt.Sprint(recover()) }()
+		defer func() {
+			if r := recover(); r != nil {
+				msg = fmt.Sprint(r)
+			}
+		}()
 		fn()
 		return ""
 	}
@@ -2,13 +2,31 @@ package fun

 import (
 	"fmt"
+	"reflect"
+	"strings"
 	"sync"
 )

+// streamMsgType 提取 *Stream[T] 的消息类型 T；非 Stream 指针返回 nil。
+// reflect 没有类型实参 API，但实例化类型的 Send 方法签名已替换——In(1) 即 T
+func streamMsgType(t reflect.Type) reflect.Type {
+	if t.Kind() != reflect.Ptr || !strings.HasPrefix(t.Elem().Name(), "Stream[") {
+		return nil
+	}
+	sm, ok := t.MethodByName("Send")
+	if !ok || sm.Type.NumIn() != 2 {
+		return nil
+	}
+	if _, ok := t.MethodByName("Inject"); !ok {
+		return nil
+	}
+	return sm.Type.In(1)
+}
+
 // Stream 流式响应的业务句柄。
-// 业务方法返回 *Stream 后，框架调用 Inject 注入推送通道；
+// 业务方法返回 *Stream[T] 后，框架调用 Inject 注入推送通道；
 // 未注入时 Send/Close 自动阻塞等待，避免业务 goroutine 与注入之间的竞态。
-type Stream struct {
+type Stream[T any] struct {
 	mu      sync.Mutex
 	once    sync.Once
 	ready   chan struct{}
@@ -18,7 +36,7 @@ type Stream struct {
 	onClose func()
 }

-func (s *Stream) getReady() chan struct{} {
+func (s *Stream[T]) getReady() chan struct{} {
 	s.once.Do(func() {
 		if s.ready == nil {
 			s.ready = make(chan struct{})
@@ -28,14 +46,14 @@ func (s *Stream) getReady() chan struct{} {
 }

 // Inject 注入推送通道与结束信号，由框架在方法返回后调用
-func (s *Stream) Inject(ch chan any, done chan struct{}) {
+func (s *Stream[T]) Inject(ch chan any, done chan struct{}) {
 	s.ch = ch
 	s.done = done
 	close(s.getReady())
 }

 // Send 推送一条消息；连接断开或流已关闭时返回错误
-func (s *Stream) Send(message any) error {
+func (s *Stream[T]) Send(message T) error {
 	<-s.getReady()
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -51,7 +69,7 @@ func (s *Stream) Send(message any) error {
 }

 // Close 主动结束流，触发 OnClose 回调
-func (s *Stream) Close() {
+func (s *Stream[T]) Close() {
 	<-s.getReady()
 	s.mu.Lock()
 	if s.closed {
@@ -67,14 +85,19 @@ func (s *Stream) Close() {
 	close(s.ch)
 }

-// OnClose 注册关闭回调；流已关闭时立即执行
-func (s *Stream) OnClose(cb func()) {
-	<-s.getReady()
+// OnClose 注册关闭回调；可在业务方法 return *Stream[T] 之前同步调用。
+// 不得等待 Inject：Inject 只有业务方法返回后才发生，等待会形成循环依赖死锁。
+// 流已关闭时在锁外立即执行回调，避免回调重入 Stream 时自锁。
+func (s *Stream[T]) OnClose(cb func()) {
+	if cb == nil {
+		return
+	}
 	s.mu.Lock()
-	defer s.mu.Unlock()
 	if s.closed {
+		s.mu.Unlock()
 		cb()
 		return
 	}
 	s.onClose = cb
+	s.mu.Unlock()
 }
@@ -0,0 +1,46 @@
+package fun
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// OnClose 必须允许业务方法在 return *Stream 之前同步注册；旧实现等待 Inject，
+// 而 Inject 只有 invoke 在业务方法返回后才调用，形成精确死锁：
+//
+//	business OnClose -> wait ready -> invoke wait business return -> never Inject.
+func TestStreamOnCloseBeforeInjectDoesNotBlock(t *testing.T) {
+	st := &Stream[any]{}
+	registered := make(chan struct{})
+	var called atomic.Int32
+	go func() {
+		st.OnClose(func() { called.Add(1) })
+		close(registered)
+	}()
+	select {
+	case <-registered:
+		// expected: registration is independent from channel injection
+	case <-time.After(time.Second):
+		t.Fatal("OnClose blocked before Inject (business method would deadlock)")
+	}
+
+	ch := make(chan any)
+	done := make(chan struct{})
+	st.Inject(ch, done)
+	st.Close()
+	if got := called.Load(); got != 1 {
+		t.Fatalf("callback called %d times, want 1", got)
+	}
+}
+
+func TestStreamOnCloseAfterAlreadyClosedRunsImmediately(t *testing.T) {
+	st := &Stream[any]{}
+	st.Inject(make(chan any), make(chan struct{}))
+	st.Close()
+	var called atomic.Int32
+	st.OnClose(func() { called.Add(1) })
+	if got := called.Load(); got != 1 {
+		t.Fatalf("callback called %d times, want 1", got)
+	}
+}
@@ -14,17 +14,17 @@ import (

 type ProtocolStreamSvc struct{}

-func (*ProtocolStreamSvc) Empty() (*Stream, error) {
-	stream := &Stream{}
+func (*ProtocolStreamSvc) Empty() (*Stream[any], error) {
+	stream := &Stream[any]{}
 	go stream.Close()
 	return stream, nil
 }

-func (*ProtocolStreamSvc) Before() (*Stream, error) {
+func (*ProtocolStreamSvc) Before() (*Stream[any], error) {
 	return nil, errors.New("before stream")
 }

-func (*ProtocolStreamSvc) Business() (*Stream, error) {
+func (*ProtocolStreamSvc) Business() (*Stream[any], error) {
 	return nil, Error(4201, "business before stream")
 }

@@ -66,6 +66,23 @@ export type RequestOptions = {
 export type StreamOptions = {
  signal?: AbortSignal;
  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 = (
@@ -342,6 +359,46 @@ export class Client {
  }

  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,
    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<Uint8Array>;
        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();
+          }
        } 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<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)
  }{{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)
Author	SHA1	Message	Date
chiyiandDevin	84c11c6fde	v1.4.0: generic Stream[T] — typed stream messages, drop (T, Stream, error) - Stream[T]: Send(v T) type-safe; message type recovered from the instantiated Send signature (reflect has no type-args API) - method signatures reduced to (error), (T, error), (Stream[T], error); the (T, *Stream, error) first-message form is gone — send it via Send - return T and stream message T now run checkType at bind time (Stream[any] remains the explicit untyped escape hatch) - generated clients: TS onMessage: (data: T); Go <-chan T - fix vacuous panic assertions (fmt.Sprint(nil) -> "<nil>") in hardening/route test catch helpers Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>	2026-09-19 15:39:41 +08:00
chiyi	81d36f827e	fix(stream): OnClose 可在返回前注册，消除 Inject 循环等待死锁旧实现 OnClose 先等待 ready，但 Inject 只有业务方法返回后才发生：业务同步 OnClose → 等 Inject；框架等业务返回 → 永久死锁，流首帧永不输出。改为仅持锁注册回调，不等待 channel 注入；已关闭场景锁外立即回调，避免重入自锁。新增两条回归测试。	2026-09-16 13:54:00 +08:00
chiyi	94fded09d5	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	2026-09-11 18:56:48 +08:00
chiyi	a4993502bc	v1.3.5: add CORS whitelist support (f.CORS) covering /cell and all custom routes	2026-09-09 11:31:22 +08:00