From 84c11c6fde5bce38783c5f06bf072c9c6ef178ff Mon Sep 17 00:00:00 2001 From: chiyi Date: Sat, 19 Sep 2026 15:39:41 +0800 Subject: [PATCH] =?UTF-8?q?v1.4.0:=20generic=20Stream[T]=20=E2=80=94=20typ?= =?UTF-8?q?ed=20stream=20messages,=20drop=20(T,=20*Stream,=20error)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) -> "") 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> --- README.md | 4 +-- bugfix_test.go | 13 ++++----- docs/README.zh.md | 6 ++--- example/demo/demo.go | 33 ++++++++++++----------- fun.go | 50 +++++++++++++++++----------------- fun_test.go | 6 ++--- gen_go.go | 13 ++++----- gen_reliability_test.go | 6 ++--- gen_ts.go | 14 +++++----- handle.go | 45 +++++++++++-------------------- hardening_test.go | 59 ++++++++++++++++++++++++++++++++++++++--- route_test.go | 6 ++++- stream.go | 34 ++++++++++++++++++------ stream_onclose_test.go | 4 +-- stream_protocol_test.go | 8 +++--- 15 files changed, 184 insertions(+), 117 deletions(-) diff --git a/README.md b/README.md index fd531f4..d78ccc7 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,10 @@ 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` 与全部自定义路由 diff --git a/bugfix_test.go b/bugfix_test.go index 10b76fa..7daafd3 100644 --- a/bugfix_test.go +++ b/bugfix_test.go @@ -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") diff --git a/docs/README.zh.md b/docs/README.zh.md index b4001a0..a912dcd 100644 --- a/docs/README.zh.md +++ b/docs/README.zh.md @@ -31,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） @@ -113,7 +113,7 @@ f.CORS("https://a.com", "https://b.com") // Start 前调用，可变参数白名 ## 7. 流式响应（NDJSON） ```go -st := &fun.Stream{} +st := &fun.Stream[string]{} go func() { for _, chunk := range chunks { if st.Send(chunk) != nil { return } // 连接断开 diff --git a/example/demo/demo.go b/example/demo/demo.go index 6b9d86f..2d386f1 100644 --- a/example/demo/demo.go +++ b/example/demo/demo.go @@ -16,19 +16,19 @@ func (OrderStatus) DisplayNames() []string { return []string{"待支付", "已 // CreateOrderDto 下单参数。规则：非指针字段必传；指针/slice 字段可省略； // 数值一律定宽整型（int64），小数用 string 传输 type CreateOrderDto struct { - Sku string // 必传 - Count int64 // 必传 - Note *string // 可空 + Sku string // 必传 + Count int64 // 必传 + Note *string // 可空 Status *OrderStatus // 可空，缺省 Pending - Tags []string // 可省略 + Tags []string // 可省略 } // OrderDto 订单视图。Id 为雪花 ID：全链路 int64 精度，TS 侧解析为 BigInt type OrderDto struct { - Id int64 - Status OrderStatus - Amount string // 小数金额用字符串传输 - Items []string + Id int64 + Status OrderStatus + Amount string // 小数金额用字符串传输 + Items []string } type GetOrderDto struct { @@ -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 } diff --git a/fun.go b/fun.go index 50f786e..0292187 100644 --- a/fun.go +++ b/fun.go @@ -13,19 +13,19 @@ import ( type Fun struct { methods map[string]methodInfo - routes map[string]boundRoute // 自定义路由："GET /path" → 绑定的处理器与 Guard（精确匹配） + routes map[string]boundRoute // 自定义路由："GET /path" → 绑定的处理器与 Guard（精确匹配） wildcardRoutes map[string][]wildcardRoute // 通配路由，按 HTTP 方法 - boxes *sync.Map // 依赖容器：reflect.Type → boxEntry（单例或粘性错误） - guards []*any // 全局 Guard - serviceGuards map[string][]*any // 服务级 Guard，按服务名 - bodyLimit int // 请求体上限（字节）；0 = fasthttp 默认 4MB + boxes *sync.Map // 依赖容器：reflect.Type → boxEntry（单例或粘性错误） + guards []*any // 全局 Guard + serviceGuards map[string][]*any // 服务级 Guard，按服务名 + bodyLimit int // 请求体上限（字节）；0 = fasthttp 默认 4MB readTimeout time.Duration // 读超时，默认 60s（slowloris 防线） writeTimeout time.Duration // 写超时，默认 0 不限制（避免掐断长流式响应） idleTimeout time.Duration // keep-alive 空闲超时，默认 120s maxConcurrency int // 最大并发连接数；0 = 不限制 - corsOrigins map[string]struct{} // CORS 来源白名单（小写比较）；nil/空表示未开启 + corsOrigins map[string]struct{} // CORS 来源白名单（小写比较）；nil/空表示未开启 server atomic.Pointer[fasthttp.Server] started atomic.Bool @@ -69,8 +69,7 @@ type wildcardRoute struct { } var ( - errorType = reflect.TypeFor[error]() - streamType = reflect.TypeFor[*Stream]() + errorType = reflect.TypeFor[error]() ) var ( @@ -83,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 { @@ -122,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 返回，由调用方决定退出或降级； @@ -188,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" diff --git a/fun_test.go b/fun_test.go index 13bc228..2573e0a 100644 --- a/fun_test.go +++ b/fun_test.go @@ -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) { diff --git a/gen_go.go b/gen_go.go index 239d825..2e970c9 100644 --- a/gen_go.go +++ b/gen_go.go @@ -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) + returnValueText = "Void" + 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) diff --git a/gen_reliability_test.go b/gen_reliability_test.go index 1b1853e..a5a5b37 100644 --- a/gen_reliability_test.go +++ b/gen_reliability_test.go @@ -25,12 +25,12 @@ 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) Request() (string, error) { return "", 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{} diff --git a/gen_ts.go b/gen_ts.go index 34cfca8..0e22c8a 100644 --- a/gen_ts.go +++ b/gen_ts.go @@ -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)) + returnValueText = "void" + 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 diff --git a/handle.go b/handle.go index dba535b..a9c3ecf 100644 --- a/handle.go +++ b/handle.go @@ -77,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{} @@ -121,13 +121,6 @@ func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) { } return true } - // (T, stream, error)：T 作为流的第一条消息下发 - if result.Data != nil { - if !writeLine(*result.Data) { - finish() - return - } - } // 空闲心跳：业务长时间不 Send 时周期写一个空行。NDJSON 解析器 // 跳过空行（业务无感知），字节能让 NAT/CDN 的空闲计时器归零。 // 25s 需小于常见中间盒阈值（移动 CGNAT 约 60s、CDN 约 120s）。 @@ -179,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 @@ -210,7 +203,7 @@ func (f *Fun) invoke(c *Ctx, streamCh *chan any, streamDone *chan struct{}) (*Re } if decodeErr != nil { // 客户端只收通用提示；详细原因记服务端日志，不外泄 Go 类型等内部信息 - ErrorLogger("fun: decode request data (" + c.ServiceName + "." + c.MethodName + "): ", decodeErr.Error()) + ErrorLogger("fun: decode request data ("+c.ServiceName+"."+c.MethodName+"): ", decodeErr.Error()) return nil, errInvalidData } args = append(args, dto) @@ -223,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() { @@ -246,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 } @@ -279,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) diff --git a/hardening_test.go b/hardening_test.go index 9a23615..aad762f 100644 --- a/hardening_test.go +++ b/hardening_test.go @@ -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") + } +} diff --git a/route_test.go b/route_test.go index 417d33f..7b61dc6 100644 --- a/route_test.go +++ b/route_test.go @@ -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 "" } diff --git a/stream.go b/stream.go index b65f28a..aead514 100644 --- a/stream.go +++ b/stream.go @@ -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,10 +85,10 @@ func (s *Stream) Close() { close(s.ch) } -// OnClose 注册关闭回调；可在业务方法 return *Stream 之前同步调用。 +// OnClose 注册关闭回调；可在业务方法 return *Stream[T] 之前同步调用。 // 不得等待 Inject：Inject 只有业务方法返回后才发生，等待会形成循环依赖死锁。 // 流已关闭时在锁外立即执行回调，避免回调重入 Stream 时自锁。 -func (s *Stream) OnClose(cb func()) { +func (s *Stream[T]) OnClose(cb func()) { if cb == nil { return } diff --git a/stream_onclose_test.go b/stream_onclose_test.go index b5ad11a..71682c4 100644 --- a/stream_onclose_test.go +++ b/stream_onclose_test.go @@ -11,7 +11,7 @@ import ( // // business OnClose -> wait ready -> invoke wait business return -> never Inject. func TestStreamOnCloseBeforeInjectDoesNotBlock(t *testing.T) { - st := &Stream{} + st := &Stream[any]{} registered := make(chan struct{}) var called atomic.Int32 go func() { @@ -35,7 +35,7 @@ func TestStreamOnCloseBeforeInjectDoesNotBlock(t *testing.T) { } func TestStreamOnCloseAfterAlreadyClosedRunsImmediately(t *testing.T) { - st := &Stream{} + st := &Stream[any]{} st.Inject(make(chan any), make(chan struct{})) st.Close() var called atomic.Int32 diff --git a/stream_protocol_test.go b/stream_protocol_test.go index 0b4d997..516b291 100644 --- a/stream_protocol_test.go +++ b/stream_protocol_test.go @@ -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") }