diff --git a/bugfix_test.go b/bugfix_test.go index dcb55e7..e486f2f 100644 --- a/bugfix_test.go +++ b/bugfix_test.go @@ -75,6 +75,7 @@ func TestBugNullableEnum(t *testing.T) { // bug3: 含 () error 方法的代码生成不应 panic，且类型应生成为 Void/void func TestBugGenErrorOnly(t *testing.T) { + isolateGeneratorGlobals(t) GetFun().BindService(&BugSvc{}) SetOutput(t.TempDir()) GenCode(GenGo{}, GenTs{}) diff --git a/client_ts_test.go b/client_ts_test.go new file mode 100644 index 0000000..cc3b21d --- /dev/null +++ b/client_ts_test.go @@ -0,0 +1,340 @@ +package fun + +import ( + "os" + "os/exec" + "path/filepath" + "strings" + "testing" +) + +func TestTypeScriptClientBehavior(t *testing.T) { + node, err := exec.LookPath("node") + if err != nil { + t.Skip("node is not installed") + } + probe := exec.Command(node, "--experimental-strip-types", "--input-type=module", "-e", + `if (typeof fetch !== "function" || typeof ReadableStream !== "function") process.exit(1)`) + if err := probe.Run(); err != nil { + t.Skip("node does not support TypeScript stripping and web streams") + } + + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "client.ts"), []byte(templateTs{}.genClientTemplate()), 0o644); err != nil { + t.Fatal(err) + } + const script = ` +import assert from "node:assert/strict"; +import { Client } from "./client.ts"; + +const json = (value, status = 200) => new Response(JSON.stringify(value), { + status, + headers: { "Content-Type": "application/json" }, +}); +const ndjson = body => new Response(body, { + headers: { "Content-Type": "application/x-ndjson; charset=utf-8" }, +}); +const client = new Client("http://example.test///"); +const seen = []; +let cleanupCheck; +client.addResponseInterceptor((_service, _method, result) => { + if (cleanupCheck) { + assert.equal(cleanupCheck(), true); + cleanupCheck = undefined; + } + seen.push(result.status); +}); +const expectSeen = status => assert.deepEqual(seen.splice(0), [status]); + +let fetchInit; +globalThis.fetch = async (url, init) => { + fetchInit = { url, init }; + return json({ status: 0, data: "ok" }); +}; +let result = await client.request("Svc", "fetchShape", undefined, { signal: new AbortController().signal }); +assert.equal(result.status, 0); +assert.equal(fetchInit.url, "http://example.test/cell"); +assert.equal(fetchInit.init.method, "POST"); +assert.equal(fetchInit.init.headers["Content-Type"], "application/json"); +assert.ok(fetchInit.init.signal); +assert.deepEqual(JSON.parse(fetchInit.init.body), { serviceName: "Svc", methodName: "fetchShape" }); +expectSeen(0); + +globalThis.fetch = async () => json({ status: 2, code: 4001, msg: "business" }, 503); +result = await client.request("Svc", "business"); +assert.deepEqual(result, { status: 2, code: 4001, msg: "business" }); +expectSeen(2); + +globalThis.fetch = async () => json({ status: 1, msg: "framework" }, 500); +result = await client.request("Svc", "framework"); +assert.deepEqual(result, { status: 1, msg: "framework" }); +expectSeen(1); + +globalThis.fetch = async () => new Response("gateway timeout", { status: 504 }); +result = await client.request("Svc", "gatewayTimeout"); +assert.equal(result.status, 5); +assert.match(result.msg, /504/); +expectSeen(5); + +globalThis.fetch = async () => new Response("request timeout", { status: 408 }); +result = await client.request("Svc", "gateway408"); +assert.equal(result.status, 5); +expectSeen(5); + +globalThis.fetch = async () => json({ status: 5, msg: "server timeout" }); +result = await client.request("Svc", "serverTimeout"); +assert.equal(result.status, 5); +assert.equal(result.msg, "server timeout"); +expectSeen(5); + +globalThis.fetch = async () => new Response("bad gateway", { status: 502 }); +result = await client.request("Svc", "gateway"); +assert.equal(result.status, 4); +expectSeen(4); + +globalThis.fetch = async () => new Response("", { status: 200 }); +result = await client.request("Svc", "empty"); +assert.equal(result.status, 1); +assert.match(result.msg, /Empty/); +expectSeen(1); + +globalThis.fetch = async () => new Response("not-json", { status: 200 }); +result = await client.request("Svc", "invalidJson"); +assert.equal(result.status, 1); +assert.match(result.msg, /Invalid JSON/); +expectSeen(1); + +globalThis.fetch = async () => new Response("login", { + status: 200, + headers: { "Content-Type": "text/html" }, +}); +result = await client.request("Svc", "html"); +assert.equal(result.status, 1); +assert.match(result.msg, /HTML/); +expectSeen(1); + +globalThis.fetch = async () => json({ data: 1 }); +result = await client.request("Svc", "invalidResult"); +assert.equal(result.status, 1); +expectSeen(1); + +globalThis.fetch = async () => { throw new TypeError("DNS failed"); }; +result = await client.request("Svc", "network"); +assert.equal(result.status, 4); +assert.match(result.msg, /DNS failed/); +expectSeen(4); + +globalThis.fetch = async () => new Response(new ReadableStream({ + pull(controller) { controller.error(new Error("body read failed")); }, +})); +result = await client.request("Svc", "bodyRead"); +assert.equal(result.status, 4); +assert.match(result.msg, /body read failed/); +expectSeen(4); + +const manualAbort = new AbortController(); +manualAbort.abort(); +globalThis.fetch = async () => { throw new DOMException("aborted", "AbortError"); }; +result = await client.request("Svc", "abort", undefined, { signal: manualAbort.signal }); +assert.equal(result.status, 4); +assert.match(result.msg, /aborted/i); +expectSeen(4); + +const timeoutAbort = new AbortController(); +timeoutAbort.abort(new DOMException("timed out", "TimeoutError")); +result = await client.request("Svc", "timeoutSignal", undefined, { signal: timeoutAbort.signal }); +assert.equal(result.status, 5); +expectSeen(5); + +globalThis.fetch = async () => { throw new DOMException("timed out", "TimeoutError"); }; +result = await client.request("Svc", "timeoutError"); +assert.equal(result.status, 5); +expectSeen(5); + +const requestInterceptorClient = new Client("http://example.test"); +let requestInterceptorSeen; +requestInterceptorClient.addRequestInterceptor((_service, _method, state, dto) => { + state.token = "abc"; + requestInterceptorSeen = dto; +}); +requestInterceptorClient.addResponseInterceptor((_service, _method, value) => ({ ...value, msg: "intercepted" })); +globalThis.fetch = async (_url, init) => { + const payload = JSON.parse(init.body); + assert.deepEqual(payload.state, { token: "abc" }); + return json({ status: 0 }); +}; +result = await requestInterceptorClient.request("Svc", "interceptors", { value: 1 }); +assert.deepEqual(requestInterceptorSeen, { value: 1 }); +assert.equal(result.msg, "intercepted"); + +const failingRequestInterceptor = new Client("http://example.test"); +let normalized = []; +failingRequestInterceptor.addRequestInterceptor(() => { throw new Error("request hook"); }); +failingRequestInterceptor.addResponseInterceptor((_s, _m, value) => { normalized.push(value.status); }); +result = await failingRequestInterceptor.request("Svc", "hook"); +assert.equal(result.status, 1); +assert.match(result.msg, /request hook/); +assert.deepEqual(normalized, [1]); + +const failingResponseInterceptor = new Client("http://example.test"); +normalized = []; +failingResponseInterceptor.addResponseInterceptor(() => { throw new Error("response hook"); }); +failingResponseInterceptor.addResponseInterceptor((_s, _m, value) => { normalized.push(value.status); }); +globalThis.fetch = async () => json({ status: 2, code: 4003, msg: "original" }); +result = await failingResponseInterceptor.request("Svc", "responseHook"); +assert.equal(result.status, 1); +assert.match(result.msg, /response hook/); +assert.deepEqual(normalized, [1]); + +const cyclic = {}; +cyclic.self = cyclic; +globalThis.fetch = async () => assert.fail("fetch must not run for serialization failure"); +result = await client.request("Svc", "serialize", cyclic); +assert.equal(result.status, 1); +assert.match(result.msg, /serialize/i); +expectSeen(1); + +globalThis.fetch = async () => json({ status: 2, code: 4002, msg: "stream business" }, 422); +result = await client.stream("Svc", "streamBusiness", undefined, () => {}); +assert.deepEqual(result, { status: 2, code: 4002, msg: "stream business" }); +expectSeen(2); + +globalThis.fetch = async () => json({ status: 0 }); +result = await client.stream("Svc", "wrongMedia", undefined, () => {}); +assert.equal(result.status, 1); +assert.match(result.msg, /x-ndjson/); +expectSeen(1); + +const bytes = new TextEncoder().encode('{"text":"你好"}\r\n\n{"n":2}'); +const split = bytes.indexOf(0xe5) + 1; +globalThis.fetch = async () => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(bytes.slice(0, split)); + controller.enqueue(bytes.slice(split)); + controller.close(); + }, +}), { headers: { "Content-Type": "application/x-ndjson; charset=utf-8" } }); +const messages = []; +result = await client.stream("Svc", "valid", undefined, value => messages.push(value)); +assert.equal(result.status, 0); +assert.deepEqual(messages, [{ text: "你好" }, { n: 2 }]); +expectSeen(0); + +let malformedCancelled = false; +globalThis.fetch = async () => new Response(new ReadableStream({ + start(controller) { controller.enqueue(new TextEncoder().encode("bad\n")); }, + cancel() { malformedCancelled = true; }, +}), { headers: { "Content-Type": "application/x-ndjson" } }); +cleanupCheck = () => malformedCancelled; +result = await client.stream("Svc", "malformed", undefined, () => {}); +assert.equal(result.status, 1); +assert.match(result.msg, /line 1/); +assert.equal(malformedCancelled, true); +expectSeen(1); + +globalThis.fetch = async () => ndjson('{"n":1}\n'); +result = await client.stream("Svc", "callback", undefined, () => { throw new Error("callback boom"); }); +assert.equal(result.status, 1); +assert.match(result.msg, /callback boom/); +expectSeen(1); + +globalThis.fetch = async () => new Response(new ReadableStream({ + pull(controller) { controller.error(new Error("read boom")); }, +}), { headers: { "Content-Type": "application/x-ndjson" } }); +result = await client.stream("Svc", "read", undefined, () => {}); +assert.equal(result.status, 4); +assert.match(result.msg, /read boom/); +expectSeen(4); + +globalThis.fetch = async () => ndjson(new Uint8Array([0xff, 0x0a])); +result = await client.stream("Svc", "utf8", undefined, () => {}); +assert.equal(result.status, 1); +assert.match(result.msg, /UTF-8/); +expectSeen(1); + +globalThis.fetch = async () => ndjson(""); +result = await client.stream("Svc", "emptyStream", undefined, () => assert.fail("empty stream callback")); +assert.deepEqual(result, { status: 0 }); +expectSeen(0); + +globalThis.fetch = async () => new Response(null, { + headers: { "Content-Type": "application/x-ndjson" }, +}); +result = await client.stream("Svc", "nullBody", undefined, () => assert.fail("null stream callback")); +assert.deepEqual(result, { status: 0 }); +expectSeen(0); + +const streamAbort = new AbortController(); +streamAbort.abort(); +globalThis.fetch = async () => { throw new DOMException("aborted", "AbortError"); }; +result = await client.stream("Svc", "abortStream", undefined, () => {}, { signal: streamAbort.signal }); +assert.equal(result.status, 4); +expectSeen(4); + +const readAbort = new AbortController(); +globalThis.fetch = async () => new Response(new ReadableStream({ + start(controller) { + readAbort.signal.addEventListener("abort", () => controller.error(new DOMException("aborted", "AbortError")), { once: true }); + readAbort.abort(); + }, +}), { headers: { "Content-Type": "application/x-ndjson" } }); +result = await client.stream("Svc", "readAbort", undefined, () => {}, { signal: readAbort.signal }); +assert.equal(result.status, 4); +assert.match(result.msg, /aborted/i); +expectSeen(4); + +const readTimeout = new AbortController(); +readTimeout.abort(new DOMException("timed out", "TimeoutError")); +globalThis.fetch = async () => new Response(new ReadableStream({ + pull(controller) { controller.error(new DOMException("aborted", "AbortError")); }, +}), { headers: { "Content-Type": "application/x-ndjson" } }); +result = await client.stream("Svc", "timeoutStream", undefined, () => {}, { signal: readTimeout.signal }); +assert.equal(result.status, 5); +expectSeen(5); + +const streamHookClient = new Client("http://example.test"); +normalized = []; +streamHookClient.addRequestInterceptor(() => { throw new Error("stream hook"); }); +streamHookClient.addResponseInterceptor((_s, _m, value) => { normalized.push(value.status); }); +result = await streamHookClient.stream("Svc", "hook", undefined, () => {}); +assert.equal(result.status, 1); +assert.deepEqual(normalized, [1]); +` + scriptPath := filepath.Join(dir, "behavior.mjs") + if err := os.WriteFile(scriptPath, []byte(script), 0o644); err != nil { + t.Fatal(err) + } + command := exec.Command(node, "--experimental-strip-types", scriptPath) + if output, err := command.CombinedOutput(); err != nil { + t.Fatalf("TypeScript client behavior failed: %v\n%s", err, output) + } +} + +func TestTypeScriptClientStrictTypecheck(t *testing.T) { + tsc, err := exec.LookPath("tsc") + if err != nil { + t.Skip("tsc is not installed") + } + dir := t.TempDir() + path := filepath.Join(dir, "client.ts") + if err := os.WriteFile(path, []byte(templateTs{}.genClientTemplate()), 0o644); err != nil { + t.Fatal(err) + } + command := exec.Command(tsc, + "--strict", "--noEmit", "--target", "ES2022", "--module", "ESNext", "--lib", "ES2022,DOM", path) + if output, err := command.CombinedOutput(); err != nil { + t.Fatalf("strict TypeScript check failed: %v\n%s", err, output) + } +} + +func TestTypeScriptClientSimplicityGate(t *testing.T) { + source := templateTs{}.genClientTemplate() + templateTs{}.genServiceTemplate() + for _, forbidden := range []string{"RpcError", "httpStatus", "result.aborted", "ClientErrorCode", "setTimeout"} { + if strings.Contains(source, forbidden) { + t.Errorf("generated TypeScript templates contain forbidden %q", forbidden) + } + } + if !strings.Contains(source, "Promise>") { + t.Fatal("stream methods must resolve result") + } +} diff --git a/fun_test.go b/fun_test.go index 455d720..ab402c2 100644 --- a/fun_test.go +++ b/fun_test.go @@ -93,6 +93,7 @@ func TestCheckDtoRequired(t *testing.T) { } func TestGenCode(t *testing.T) { + isolateGeneratorGlobals(t) GetFun().BindService(&TestSvc{}) SetOutput(t.TempDir()) GenCode(GenGo{}, GenTs{}) diff --git a/gen.go b/gen.go index 4a43916..e520745 100644 --- a/gen.go +++ b/gen.go @@ -6,6 +6,7 @@ import ( "path/filepath" "reflect" "regexp" + "sort" "strings" "text/template" ) @@ -33,8 +34,8 @@ type genMethod struct { isStream bool } -// serviceGroups 按服务名分组已注册方法 -func (f *Fun) serviceGroups() map[string][]*genMethod { +// serviceGroups 按服务名和方法名稳定分组已注册方法 +func (f *Fun) serviceGroups() []*genSvc { groups := map[string][]*genMethod{} for key, m := range f.methods { parts := strings.SplitN(key, ".", 2) @@ -47,7 +48,14 @@ func (f *Fun) serviceGroups() map[string][]*genMethod { isStream: m.isStream, }) } - return groups + + services := make([]*genSvc, 0, len(groups)) + for name, methods := range groups { + sort.Slice(methods, func(i, j int) bool { return methods[i].name < methods[j].name }) + services = append(services, &genSvc{name: name, methods: methods}) + } + sort.Slice(services, func(i, j int) bool { return services[i].name < services[j].name }) + return services } type genType struct { @@ -79,6 +87,7 @@ type genServiceType struct { GenMethodTypeList []*genMethodType GenImport []*genImportType IsIncludeProxy bool + IsIncludeRequest bool IsIncludeStream bool } @@ -103,6 +112,7 @@ func deduplicateServiceImports(imports []*genImportType) []*genImportType { result = append(result, imp) } } + sort.Slice(result, func(i, j int) bool { return result[i].Name < result[j].Name }) return result } diff --git a/gen_go.go b/gen_go.go index 82bcd17..17048e1 100644 --- a/gen_go.go +++ b/gen_go.go @@ -132,13 +132,13 @@ func (ctx GenGo) genDefaultService() { f := GetFun() genContext := genType{GenServiceList: []*genServiceType{}} - for svcName, methods := range f.serviceGroups() { + for _, svc := range f.serviceGroups() { serviceContext := &genServiceType{ - ServiceName: svcName, + ServiceName: svc.name, GenMethodTypeList: []*genMethodType{}, } genContext.GenServiceList = append(genContext.GenServiceList, serviceContext) - ctx.genService(&genSvc{name: svcName, methods: methods}, serviceContext) + ctx.genService(svc, serviceContext) } genCode(ctx.template.genDefaultServiceTemplate(), "fun", genContext, ctx.getName()) } diff --git a/gen_reliability_test.go b/gen_reliability_test.go new file mode 100644 index 0000000..83ad770 --- /dev/null +++ b/gen_reliability_test.go @@ -0,0 +1,179 @@ +package fun + +import ( + "os" + "path/filepath" + "sort" + "strings" + "testing" +) + +type AlphaGenDto struct { + Value string +} + +type ZebraGenDto struct { + Value string +} + +type AlphaGenSvc struct{} + +func (*AlphaGenSvc) Zebra(dto ZebraGenDto) (AlphaGenDto, error) { return AlphaGenDto{}, nil } +func (*AlphaGenSvc) Alpha(dto AlphaGenDto) (ZebraGenDto, error) { return ZebraGenDto{}, nil } +func (*AlphaGenSvc) Ping() error { return nil } + +type MixedGenSvc struct{} + +func (*MixedGenSvc) Request() (string, error) { return "", nil } +func (*MixedGenSvc) Stream() (*Stream, error) { return &Stream{}, nil } + +type ZebraGenSvc struct{} + +func (*ZebraGenSvc) Watch() (*Stream, error) { return &Stream{}, nil } + +func isolateGeneratorGlobals(t *testing.T) { + t.Helper() + oldFun, oldDirectory := fun, directory + fun = nil + directory = "./gen" + t.Cleanup(func() { + fun = oldFun + directory = oldDirectory + }) +} + +func generatedFiles(t *testing.T, root string) map[string]string { + t.Helper() + files := map[string]string{} + err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + if info.IsDir() { + return nil + } + body, err := os.ReadFile(path) + if err != nil { + return err + } + rel, err := filepath.Rel(root, path) + if err != nil { + return err + } + files[rel] = string(body) + return nil + }) + if err != nil { + t.Fatal(err) + } + return files +} + +func TestGeneratedTypeScriptSignaturesAndImports(t *testing.T) { + isolateGeneratorGlobals(t) + f := GetFun() + f.BindService(&ZebraGenSvc{}) + f.BindService(&MixedGenSvc{}) + f.BindService(&AlphaGenSvc{}) + SetOutput(t.TempDir()) + GenCode(GenTs{}) + + read := func(name string) string { + t.Helper() + body, err := os.ReadFile(filepath.Join(getDirectory(), "ts", name)) + if err != nil { + t.Fatal(err) + } + return string(body) + } + + alpha := read("alphaGenSvc.ts") + if first := strings.SplitN(alpha, "\n", 2)[0]; first != `import { Client, type result, type RequestOptions } from "./client";` { + t.Fatalf("unexpected request-only imports: %s", first) + } + for _, want := range []string{ + `async ping(options?: RequestOptions): Promise>`, + `this.client.request("alphaGenSvc", "ping", undefined, options)`, + `async alpha(dto:alphaGenDto, options?: RequestOptions): Promise>`, + `this.client.request("alphaGenSvc", "alpha", dto, options)`, + } { + if !strings.Contains(alpha, want) { + t.Errorf("alphaGenSvc.ts missing %q:\n%s", want, alpha) + } + } + if strings.Index(alpha, `import type alphaGenDto`) > strings.Index(alpha, `import type zebraGenDto`) { + t.Fatalf("DTO imports are not sorted:\n%s", alpha) + } + if strings.Index(alpha, `async alpha`) > strings.Index(alpha, `async ping`) || + strings.Index(alpha, `async ping`) > strings.Index(alpha, `async zebra`) { + t.Fatalf("methods are not sorted:\n%s", alpha) + } + + stream := read("zebraGenSvc.ts") + if first := strings.SplitN(stream, "\n", 2)[0]; first != `import { Client, type result, type StreamOptions } from "./client";` { + t.Fatalf("unexpected stream-only imports: %s", first) + } + for _, want := range []string{ + `async watch(onMessage: (data: any) => unknown, options?: StreamOptions): Promise>`, + `this.client.stream("zebraGenSvc", "watch", undefined, onMessage, options)`, + } { + if !strings.Contains(stream, want) { + t.Errorf("zebraGenSvc.ts missing %q:\n%s", want, stream) + } + } + + mixed := read("mixedGenSvc.ts") + if first := strings.SplitN(mixed, "\n", 2)[0]; first != `import { Client, type result, type RequestOptions, type StreamOptions } from "./client";` { + t.Fatalf("unexpected mixed imports: %s", first) + } +} + +func TestGeneratedSourcesAreDeterministic(t *testing.T) { + isolateGeneratorGlobals(t) + f := GetFun() + f.BindService(&ZebraGenSvc{}) + f.BindService(&AlphaGenSvc{}) + f.BindService(&MixedGenSvc{}) + root := t.TempDir() + SetOutput(root) + GenCode(GenGo{}, GenTs{}) + first := generatedFiles(t, root) + GenCode(GenGo{}, GenTs{}) + second := generatedFiles(t, root) + if len(first) != len(second) { + t.Fatalf("generated file count changed: %d != %d", len(first), len(second)) + } + for name, body := range first { + if second[name] != body { + t.Errorf("generated file changed between runs: %s", name) + } + } + + tsFun := first[filepath.Join("ts", "fun.ts")] + positions := []int{ + strings.Index(tsFun, `import alphaGenSvc`), + strings.Index(tsFun, `import mixedGenSvc`), + strings.Index(tsFun, `import zebraGenSvc`), + } + if !sort.IntsAreSorted(positions) || positions[0] < 0 { + t.Fatalf("TypeScript services are not sorted:\n%s", tsFun) + } + goFun := first[filepath.Join("go", "fun.go")] + positions = []int{ + strings.Index(goFun, "AlphaGenSvc *AlphaGenSvc"), + strings.Index(goFun, "MixedGenSvc *MixedGenSvc"), + strings.Index(goFun, "ZebraGenSvc *ZebraGenSvc"), + } + if !sort.IntsAreSorted(positions) || positions[0] < 0 { + t.Fatalf("Go services are not sorted:\n%s", goFun) + } + goService := first[filepath.Join("go", "alpha_gen_svc.go")] + positions = []int{ + strings.Index(goService, "func (ctx *AlphaGenSvc) Alpha("), + strings.Index(goService, "func (ctx *AlphaGenSvc) Ping("), + strings.Index(goService, "func (ctx *AlphaGenSvc) Zebra("), + } + if !sort.IntsAreSorted(positions) || positions[0] < 0 { + t.Fatalf("Go methods are not sorted:\n%s", goService) + } +} diff --git a/gen_ts.go b/gen_ts.go index 4e66a7b..34cfca8 100644 --- a/gen_ts.go +++ b/gen_ts.go @@ -56,6 +56,7 @@ func (ctx GenTs) genService(svc *genSvc, serviceContext *genServiceType) { argsText += ",dto" nestedImports = append(nestedImports, ctx.genStruct(gm.dtoType)) } + serviceContext.IsIncludeRequest = true serviceContext.GenMethodTypeList = append(serviceContext.GenMethodTypeList, &genMethodType{ MethodName: firstLetterToLower(gm.name), ReturnValueText: returnValueText, @@ -83,6 +84,7 @@ func (ctx GenTs) genService(svc *genSvc, serviceContext *genServiceType) { nestedImports = ctx.genReturnTypes(returnType, nestedImports) } } else { + serviceContext.IsIncludeRequest = true t := firstLetterToLower(ctx.typeToTemplateType(returnType)) if !strings.Contains(t, "[]") && strings.Contains(t, "[") { returnValueText = getGenericTypeName(t) + parseGenericTypeParams(t) @@ -151,13 +153,13 @@ func (ctx GenTs) genDefaultService() { f := GetFun() genContext := genType{GenServiceList: []*genServiceType{}} - for svcName, methods := range f.serviceGroups() { + for _, svc := range f.serviceGroups() { serviceContext := &genServiceType{ - ServiceName: firstLetterToLower(svcName), + ServiceName: firstLetterToLower(svc.name), GenMethodTypeList: []*genMethodType{}, } genContext.GenServiceList = append(genContext.GenServiceList, serviceContext) - ctx.genService(&genSvc{name: svcName, methods: methods}, serviceContext) + ctx.genService(svc, serviceContext) } genCode(ctx.template.genClientTemplate(), "client", nil, ctx.getName()) genCode(ctx.template.genDefaultServiceTemplate(), "fun", genContext, ctx.getName()) diff --git a/stream_protocol_test.go b/stream_protocol_test.go new file mode 100644 index 0000000..d261c9b --- /dev/null +++ b/stream_protocol_test.go @@ -0,0 +1,109 @@ +package fun + +import ( + "encoding/json" + "errors" + "io" + "net" + "net/http" + "strings" + "testing" + + "github.com/valyala/fasthttp" +) + +type ProtocolStreamSvc struct{} + +func (*ProtocolStreamSvc) Empty() (*Stream, error) { + stream := &Stream{} + go stream.Close() + return stream, nil +} + +func (*ProtocolStreamSvc) Before() (*Stream, error) { + return nil, errors.New("before stream") +} + +func (*ProtocolStreamSvc) Business() (*Stream, error) { + return nil, Error(4201, "business before stream") +} + +func serveFun(t *testing.T, f *Fun) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- fasthttp.Serve(listener, f.handle) }() + t.Cleanup(func() { + _ = listener.Close() + <-done + }) + return "http://" + listener.Addr().String() +} + +func streamPost(t *testing.T, url, method string) (*http.Response, []byte) { + t.Helper() + body := strings.NewReader(`{"serviceName":"ProtocolStreamSvc","methodName":"` + method + `"}`) + request, err := http.NewRequest(http.MethodPost, url+"/cell", body) + if err != nil { + t.Fatal(err) + } + request.Header.Set("Content-Type", "application/json") + request.Close = true + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + data, err := io.ReadAll(response.Body) + if err != nil { + t.Fatal(err) + } + return response, data +} + +func TestServerStreamContentTypeAndEmptyStream(t *testing.T) { + f := New() + f.BindService(&ProtocolStreamSvc{}) + response, body := streamPost(t, serveFun(t, f), "Empty") + if got := response.Header.Get("Content-Type"); got != "application/x-ndjson" { + t.Fatalf("Content-Type = %q", got) + } + if len(body) != 0 { + t.Fatalf("empty stream returned %q", body) + } +} + +func TestServerStreamSetupErrorsUseResultProtocol(t *testing.T) { + f := New() + f.BindService(&ProtocolStreamSvc{}) + url := serveFun(t, f) + for _, test := range []struct { + method string + status uint8 + code uint16 + msg string + }{ + {method: "Before", status: 1, msg: "before stream"}, + {method: "Business", status: 2, code: 4201, msg: "business before stream"}, + } { + t.Run(test.method, func(t *testing.T) { + response, body := streamPost(t, url, test.method) + if strings.HasPrefix(response.Header.Get("Content-Type"), "application/x-ndjson") { + t.Fatalf("setup error used stream Content-Type: %q", response.Header.Get("Content-Type")) + } + var result Result[any] + if err := json.Unmarshal(body, &result); err != nil { + t.Fatalf("invalid Result body %q: %v", body, err) + } + if result.Status != test.status || result.Msg == nil || *result.Msg != test.msg { + t.Fatalf("unexpected Result: %+v", result) + } + if test.code != 0 && (result.Code == nil || *result.Code != test.code) { + t.Fatalf("code = %v, want %d", result.Code, test.code) + } + }) + } +} diff --git a/template_ts.go b/template_ts.go index c930757..270bd41 100644 --- a/template_ts.go +++ b/template_ts.go @@ -12,7 +12,15 @@ func (ctx templateTs) genClientTemplate() string { }; export type resultStatus = 0 | 1 | 2 | 4 | 5; -// 0 成功；1 框架错误；2 业务错误；4 网络错误；5 超时 +// 0 success; 1 framework/client protocol error; 2 business error; 4 external request error; 5 external timeout + +export type RequestOptions = { + signal?: AbortSignal; +}; + +export type StreamOptions = { + signal?: AbortSignal; +}; export type RequestInterceptor = ( serviceName: string, @@ -21,13 +29,110 @@ export type RequestInterceptor = ( dto?: any ) => Promise | void; -// 返回新 result 将替换原结果继续向下传递（可用于集中换 token / 错误处理） export type ResponseInterceptor = ( serviceName: string, methodName: string, result: result ) => Promise | void> | result | void; +function messageOf(error: unknown): string { + if (error instanceof Error && error.message) return error.message; + if (typeof error === "string" && error) return error; + return "unknown error"; +} + +function failure(status: resultStatus, msg: string): result { + return { status, msg }; +} + +function isTimeout(error: unknown, signal?: AbortSignal): boolean { + const errorName = error !== null && typeof error === "object" + ? (error as { name?: unknown }).name + : undefined; + const reason = signal?.reason; + const reasonName = reason !== null && typeof reason === "object" + ? (reason as { name?: unknown }).name + : undefined; + return errorName === "TimeoutError" || reasonName === "TimeoutError"; +} + +function requestFailure(error: unknown, signal: AbortSignal | undefined, stream: boolean): result { + if (isTimeout(error, signal)) { + return failure(5, stream ? "Stream timed out" : "Request timed out"); + } + if (signal?.aborted === true) { + return failure(4, stream ? "Stream aborted" : "Request aborted"); + } + const kind = stream ? "External stream request" : "External request"; + return failure(4, ` + "`${kind} failed: ${messageOf(error)}`" + `); +} + +function isResult(value: unknown): value is result { + return value !== null && typeof value === "object" && + typeof (value as { status?: unknown }).status === "number"; +} + +function mediaType(response: Response): string { + return (response.headers.get("content-type") || "").split(";", 1)[0].trim().toLowerCase(); +} + +function excerpt(text: string, limit = 180): string { + const value = text.replace(/\s+/g, " ").trim(); + return value.length <= limit ? value : ` + "`${value.slice(0, limit)}...`" + `; +} + +function externalFailure(response: Response, detail?: string): result { + const timeout = response.status === 408 || response.status === 504; + const statusText = response.statusText || (timeout ? "timeout" : "request failed"); + const suffix = detail ? ` + "`: ${detail}`" + ` : ""; + return failure(timeout ? 5 : 4, ` + "`HTTP ${response.status} ${statusText}${suffix}`" + `); +} + +function responseReadFailure( + error: unknown, + response: Response, + signal: AbortSignal | undefined, + stream: boolean +): result { + if (isTimeout(error, signal)) { + return failure(5, stream ? "Stream timed out" : "Request timed out"); + } + if (signal?.aborted === true) { + return failure(4, stream ? "Stream aborted" : "Request aborted"); + } + const kind = stream ? "Stream" : "Response body"; + return response.ok + ? failure(4, ` + "`${kind} failed: ${messageOf(error)}`" + `) + : externalFailure(response, ` + "`response body failed: ${messageOf(error)}`" + `); +} + +function parseResult(response: Response, text: string): result { + const body = text.trim(); + if (!body) { + return response.ok + ? failure(1, "Empty response body") + : externalFailure(response); + } + + let value: unknown; + try { + value = JSON.parse(body); + } catch { + if (!response.ok) return externalFailure(response, excerpt(body)); + const type = mediaType(response); + if (type === "text/html" || /^\s*(?: = {}; @@ -42,81 +147,293 @@ export class Client { this.state = state; } - addRequestInterceptor(i: RequestInterceptor) { - this.requestInterceptors.push(i); + addRequestInterceptor(interceptor: RequestInterceptor) { + this.requestInterceptors.push(interceptor); } - addResponseInterceptor(i: ResponseInterceptor) { - this.responseInterceptors.push(i); + addResponseInterceptor(interceptor: ResponseInterceptor) { + this.responseInterceptors.push(interceptor); } - async request(serviceName: string, methodName: string, dto?: any): Promise> { - const state: Record = { ...this.state }; - for (const i of this.requestInterceptors) { - await i(serviceName, methodName, state, dto); + private async interceptResponse( + serviceName: string, + methodName: string, + initial: result + ): Promise> { + let current = initial; + for (const interceptor of this.responseInterceptors) { + try { + const replaced = await interceptor(serviceName, methodName, current); + if (replaced) current = replaced; + } catch (error) { + current = failure(1, ` + "`Response interceptor failed: ${messageOf(error)}`" + `); + } } - let out: result; + return current; + } + + private async requestState(serviceName: string, methodName: string, dto: any): Promise> { + const state: Record = { ...this.state }; + for (const interceptor of this.requestInterceptors) { + await interceptor(serviceName, methodName, state, dto); + } + return state; + } + + async request( + serviceName: string, + methodName: string, + dto?: any, + options?: RequestOptions + ): Promise> { + let state: Record; try { - const res = await fetch(this.url + "/cell", { + state = await this.requestState(serviceName, methodName, dto); + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + failure(1, ` + "`Request interceptor failed: ${messageOf(error)}`" + `) + ) as result; + } + + let body: string; + try { + const serialized = JSON.stringify({ + serviceName, + methodName, + data: dto, + ...(Object.keys(state).length ? { state } : {}), + }); + if (serialized === undefined) throw new Error("serialization produced no output"); + body = serialized; + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + failure(1, ` + "`Could not serialize request: ${messageOf(error)}`" + `) + ) as result; + } + + let output: result; + try { + const response = await fetch(` + "`${this.url}/cell`" + `, { method: "POST", headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ serviceName, methodName, data: dto, ...(Object.keys(state).length ? { state } : {}) }), + body, + signal: options?.signal, }); - out = (await res.json()) as result; - } catch (e: any) { - out = { status: 4, msg: (e && e.message) || "网络错误" } as result; + try { + output = parseResult(response, await response.text()); + } catch (error) { + output = responseReadFailure(error, response, options?.signal, false); + } + } catch (error) { + output = requestFailure(error, options?.signal, false); } - let cur: result = out as result; - for (const i of this.responseInterceptors) { - const replaced = await i(serviceName, methodName, cur); - if (replaced) cur = replaced; - } - return cur as result; + return await this.interceptResponse(serviceName, methodName, output) as result; } async stream( serviceName: string, methodName: string, dto: any | undefined, - onMessage: (data: T) => void - ): Promise { - const state: Record = { ...this.state }; - for (const i of this.requestInterceptors) { - await i(serviceName, methodName, state, dto); + onMessage: (data: T) => unknown, + options?: StreamOptions + ): Promise> { + let state: Record; + try { + state = await this.requestState(serviceName, methodName, dto); + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + failure(1, ` + "`Request interceptor failed: ${messageOf(error)}`" + `) + ); } - const res = await fetch(this.url + "/cell", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ serviceName, methodName, data: dto, ...(Object.keys(state).length ? { state } : {}) }), - }); - if (!res.ok) return; - const anyResult: result = { status: 0 }; - for (const i of this.responseInterceptors) { - await i(serviceName, methodName, anyResult); + + let body: string; + try { + const serialized = JSON.stringify({ + serviceName, + methodName, + data: dto, + ...(Object.keys(state).length ? { state } : {}), + }); + if (serialized === undefined) throw new Error("serialization produced no output"); + body = serialized; + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + failure(1, ` + "`Could not serialize request: ${messageOf(error)}`" + `) + ); } - if (!res.body) return; - const reader = res.body.getReader(); - const decoder = new TextDecoder(); + + let response: Response; + try { + response = await fetch(` + "`${this.url}/cell`" + `, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body, + signal: options?.signal, + }); + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + requestFailure(error, options?.signal, true) + ); + } + + if (!response.ok) { + let text: string; + try { + text = await response.text(); + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + responseReadFailure(error, response, options?.signal, true) + ); + } + return await this.interceptResponse(serviceName, methodName, parseResult(response, text)); + } + + if (mediaType(response) !== "application/x-ndjson") { + let text: string; + try { + text = await response.text(); + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + responseReadFailure(error, response, options?.signal, true) + ); + } + const rpcResult = parseResult(response, text); + return await this.interceptResponse( + serviceName, + methodName, + rpcResult.status === 0 + ? failure(1, "Expected application/x-ndjson response") + : rpcResult + ); + } + + if (!response.body) { + return await this.interceptResponse(serviceName, methodName, { status: 0 }); + } + + let reader: ReadableStreamDefaultReader; + try { + reader = response.body.getReader(); + } catch (error) { + return await this.interceptResponse( + serviceName, + methodName, + responseReadFailure(error, response, options?.signal, true) + ); + } + const decoder = new TextDecoder("utf-8", { fatal: true }); let buffer = ""; - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - for (const line of lines) { - const payload = line.trim(); - if (!payload) continue; - const data = JSON.parse(payload) as T; - onMessage(data); + let lineNumber = 0; + let failed: result | undefined; + let cause: unknown; + + const emitLine = async (line: string) => { + const payload = line.replace(/\r$/, "").trim(); + if (!payload) return; + let data: T; + try { + data = JSON.parse(payload) as T; + } catch (error) { + failed = failure(1, ` + "`Invalid NDJSON at line ${lineNumber}: ${excerpt(payload)}`" + `); + cause = error; + return; + } + try { + await onMessage(data); + } catch (error) { + failed = failure(1, ` + "`Stream callback failed: ${messageOf(error)}`" + `); + cause = error; + } + }; + + try { + for (;;) { + let part: ReadableStreamReadResult; + try { + part = await reader.read(); + } catch (error) { + cause = error; + if (isTimeout(error, options?.signal)) { + failed = failure(5, "Stream timed out"); + } else if (options?.signal?.aborted === true) { + failed = failure(4, "Stream aborted"); + } else { + failed = failure(4, ` + "`Stream read failed: ${messageOf(error)}`" + `); + } + break; + } + if (part.done) break; + try { + buffer += decoder.decode(part.value, { stream: true }); + } catch (error) { + failed = failure(1, ` + "`Invalid UTF-8 stream data: ${messageOf(error)}`" + `); + cause = error; + break; + } + for (;;) { + const newline = buffer.indexOf("\n"); + if (newline < 0) break; + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + lineNumber++; + await emitLine(line); + if (failed) break; + } + if (failed) break; + } + + if (!failed) { + try { + buffer += decoder.decode(); + } catch (error) { + failed = failure(1, ` + "`Invalid UTF-8 stream data: ${messageOf(error)}`" + `); + cause = error; + } + } + if (!failed && buffer.length > 0) { + lineNumber++; + await emitLine(buffer); + } + + } finally { + if (failed) { + try { + await reader.cancel(cause); + } catch { + // The reader may already be closed by the runtime. + } + } + try { + reader.releaseLock(); + } catch { + // The reader may already be errored or released. } } + return await this.interceptResponse( + serviceName, + methodName, + failed || { status: 0 } + ); } }` } func (ctx templateTs) genDefaultServiceTemplate() string { - return `import { Client, type result } from "./client"; + return `import { Client } from "./client"; {{- range .GenServiceList}} import {{.ServiceName}} from "./{{.ServiceName}}"; {{- end}} @@ -138,7 +455,7 @@ export default class api { } func (ctx templateTs) genServiceTemplate() string { - return `import { Client, type result } from "./client" + return `import { Client{{if .IsIncludeRequest}}, type result, type RequestOptions{{end}}{{if .IsIncludeStream}}{{if not .IsIncludeRequest}}, type result{{end}}, type StreamOptions{{end}} } from "./client"; {{- range .GenImport}} import type {{.Name}} from "./{{.Name}}"; {{- end}} @@ -150,10 +467,10 @@ export default class {{.ServiceName}} { } {{- $serviceName := .ServiceName }} {{- range .GenMethodTypeList}} - {{if .IsStream }}async {{.MethodName}}({{.DtoText}}{{if .DtoText}},{{end}}onMessage: (data: {{.GenericTypeText}}) => void): Promise { - return await this.client.stream<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, onMessage) - }{{else}}async {{.MethodName}}({{.DtoText}}): Promise<{{.ReturnValueText}}> { - return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}"{{.ArgsText}}) + {{if .IsStream }}async {{.MethodName}}({{if .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) }{{end}} {{- end}} }`