Author SHA1 Message Date
chiyi eb290cc88c v1.3.1: improve TypeScript client reliability 2026-08-21 15:06:11 +08:00
chiyi 503d0904d5 v1.3.0: BindRoute wildcard routes (/prefix/*) with RouteCtx.Wildcard 2026-08-20 21:10:24 +08:00
chiyi e6c6f68a3a fun: upgrade TS client interceptors to fun-client parity (v1.2.0)
- request interceptor now receives the mutable per-request state map (token injection)
- response interceptor may return a replacement result (central token swap / error handling)
- fetch failures normalized to result status=4 (network error) instead of throwing
- stream requests also pass through request interceptors with state copy
2026-08-20 12:09:43 +08:00
12 changed files with 1084 additions and 85 deletions
+1
View File
@@ -75,6 +75,7 @@ func TestBugNullableEnum(t *testing.T) {
// bug3: 含 () error 方法的代码生成不应 panic,且类型应生成为 Void/void // bug3: 含 () error 方法的代码生成不应 panic,且类型应生成为 Void/void
func TestBugGenErrorOnly(t *testing.T) { func TestBugGenErrorOnly(t *testing.T) {
isolateGeneratorGlobals(t)
GetFun().BindService(&BugSvc{}) GetFun().BindService(&BugSvc{})
SetOutput(t.TempDir()) SetOutput(t.TempDir())
GenCode(GenGo{}, GenTs{}) GenCode(GenGo{}, GenTs{})
+340
View File
@@ -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("<html>gateway timeout</html>", { 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("<html>login</html>", {
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<result<void>>") {
t.Fatal("stream methods must resolve result<void>")
}
}
+10 -1
View File
@@ -10,12 +10,20 @@ import (
type Fun struct { type Fun struct {
methods map[string]methodInfo methods map[string]methodInfo
routes map[string]RouteHandler // 自定义路由:"GET /path" → 处理器 routes map[string]RouteHandler // 自定义路由:"GET /path" → 处理器(精确匹配)
wildcardRoutes map[string][]wildcardRoute
boxes *sync.Map // 依赖容器:reflect.Type → reflect.Value boxes *sync.Map // 依赖容器:reflect.Type → reflect.Value
guards []*any // 全局 Guard guards []*any // 全局 Guard
serviceGuards map[string][]*any // 服务级 Guard,按服务名 serviceGuards map[string][]*any // 服务级 Guard,按服务名
} }
// wildcardRoute 通配符路由(BindRoute path 以 "/*" 结尾注册):
// prefix 如 "/image",匹配 prefix 与 prefix 下任意子路径
type wildcardRoute struct {
prefix string
handler RouteHandler
}
var ( var (
errorType = reflect.TypeFor[error]() errorType = reflect.TypeFor[error]()
streamType = reflect.TypeFor[*Stream]() streamType = reflect.TypeFor[*Stream]()
@@ -35,6 +43,7 @@ func New() *Fun {
f := &Fun{ f := &Fun{
methods: map[string]methodInfo{}, methods: map[string]methodInfo{},
routes: map[string]RouteHandler{}, routes: map[string]RouteHandler{},
wildcardRoutes: map[string][]wildcardRoute{},
boxes: &sync.Map{}, boxes: &sync.Map{},
serviceGuards: map[string][]*any{}, serviceGuards: map[string][]*any{},
} }
+1
View File
@@ -93,6 +93,7 @@ func TestCheckDtoRequired(t *testing.T) {
} }
func TestGenCode(t *testing.T) { func TestGenCode(t *testing.T) {
isolateGeneratorGlobals(t)
GetFun().BindService(&TestSvc{}) GetFun().BindService(&TestSvc{})
SetOutput(t.TempDir()) SetOutput(t.TempDir())
GenCode(GenGo{}, GenTs{}) GenCode(GenGo{}, GenTs{})
+13 -3
View File
@@ -6,6 +6,7 @@ import (
"path/filepath" "path/filepath"
"reflect" "reflect"
"regexp" "regexp"
"sort"
"strings" "strings"
"text/template" "text/template"
) )
@@ -33,8 +34,8 @@ type genMethod struct {
isStream bool isStream bool
} }
// serviceGroups 按服务名分组已注册方法 // serviceGroups 按服务名和方法名稳定分组已注册方法
func (f *Fun) serviceGroups() map[string][]*genMethod { func (f *Fun) serviceGroups() []*genSvc {
groups := map[string][]*genMethod{} groups := map[string][]*genMethod{}
for key, m := range f.methods { for key, m := range f.methods {
parts := strings.SplitN(key, ".", 2) parts := strings.SplitN(key, ".", 2)
@@ -47,7 +48,14 @@ func (f *Fun) serviceGroups() map[string][]*genMethod {
isStream: m.isStream, 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 { type genType struct {
@@ -79,6 +87,7 @@ type genServiceType struct {
GenMethodTypeList []*genMethodType GenMethodTypeList []*genMethodType
GenImport []*genImportType GenImport []*genImportType
IsIncludeProxy bool IsIncludeProxy bool
IsIncludeRequest bool
IsIncludeStream bool IsIncludeStream bool
} }
@@ -103,6 +112,7 @@ func deduplicateServiceImports(imports []*genImportType) []*genImportType {
result = append(result, imp) result = append(result, imp)
} }
} }
sort.Slice(result, func(i, j int) bool { return result[i].Name < result[j].Name })
return result return result
} }
+3 -3
View File
@@ -132,13 +132,13 @@ func (ctx GenGo) genDefaultService() {
f := GetFun() f := GetFun()
genContext := genType{GenServiceList: []*genServiceType{}} genContext := genType{GenServiceList: []*genServiceType{}}
for svcName, methods := range f.serviceGroups() { for _, svc := range f.serviceGroups() {
serviceContext := &genServiceType{ serviceContext := &genServiceType{
ServiceName: svcName, ServiceName: svc.name,
GenMethodTypeList: []*genMethodType{}, GenMethodTypeList: []*genMethodType{},
} }
genContext.GenServiceList = append(genContext.GenServiceList, serviceContext) 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()) genCode(ctx.template.genDefaultServiceTemplate(), "fun", genContext, ctx.getName())
} }
+179
View File
@@ -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<result<void>>`,
`this.client.request<void>("alphaGenSvc", "ping", undefined, options)`,
`async alpha(dto:alphaGenDto, options?: RequestOptions): Promise<result<zebraGenDto>>`,
`this.client.request<zebraGenDto>("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<result<void>>`,
`this.client.stream<any>("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)
}
}
+5 -3
View File
@@ -56,6 +56,7 @@ func (ctx GenTs) genService(svc *genSvc, serviceContext *genServiceType) {
argsText += ",dto" argsText += ",dto"
nestedImports = append(nestedImports, ctx.genStruct(gm.dtoType)) nestedImports = append(nestedImports, ctx.genStruct(gm.dtoType))
} }
serviceContext.IsIncludeRequest = true
serviceContext.GenMethodTypeList = append(serviceContext.GenMethodTypeList, &genMethodType{ serviceContext.GenMethodTypeList = append(serviceContext.GenMethodTypeList, &genMethodType{
MethodName: firstLetterToLower(gm.name), MethodName: firstLetterToLower(gm.name),
ReturnValueText: returnValueText, ReturnValueText: returnValueText,
@@ -83,6 +84,7 @@ func (ctx GenTs) genService(svc *genSvc, serviceContext *genServiceType) {
nestedImports = ctx.genReturnTypes(returnType, nestedImports) nestedImports = ctx.genReturnTypes(returnType, nestedImports)
} }
} else { } else {
serviceContext.IsIncludeRequest = true
t := firstLetterToLower(ctx.typeToTemplateType(returnType)) t := firstLetterToLower(ctx.typeToTemplateType(returnType))
if !strings.Contains(t, "[]") && strings.Contains(t, "[") { if !strings.Contains(t, "[]") && strings.Contains(t, "[") {
returnValueText = getGenericTypeName(t) + parseGenericTypeParams(t) returnValueText = getGenericTypeName(t) + parseGenericTypeParams(t)
@@ -151,13 +153,13 @@ func (ctx GenTs) genDefaultService() {
f := GetFun() f := GetFun()
genContext := genType{GenServiceList: []*genServiceType{}} genContext := genType{GenServiceList: []*genServiceType{}}
for svcName, methods := range f.serviceGroups() { for _, svc := range f.serviceGroups() {
serviceContext := &genServiceType{ serviceContext := &genServiceType{
ServiceName: firstLetterToLower(svcName), ServiceName: firstLetterToLower(svc.name),
GenMethodTypeList: []*genMethodType{}, GenMethodTypeList: []*genMethodType{},
} }
genContext.GenServiceList = append(genContext.GenServiceList, serviceContext) 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.genClientTemplate(), "client", nil, ctx.getName())
genCode(ctx.template.genDefaultServiceTemplate(), "fun", genContext, ctx.getName()) genCode(ctx.template.genDefaultServiceTemplate(), "fun", genContext, ctx.getName())
+11 -3
View File
@@ -7,19 +7,27 @@ import (
"fmt" "fmt"
"reflect" "reflect"
"runtime/debug" "runtime/debug"
"strings"
"github.com/valyala/fasthttp" "github.com/valyala/fasthttp"
) )
// handle 处理 HTTP 请求:先匹配自定义路由(BindRoute),未命中走 /cell RPC // handle 处理 HTTP 请求:先匹配自定义路由(BindRoute 精确/通配),未命中走 /cell RPC
func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) { func (f *Fun) handle(fastCtx *fasthttp.RequestCtx) {
ctx := &Ctx{RequestCtx: fastCtx} ctx := &Ctx{RequestCtx: fastCtx}
defer f.handlePanic(ctx) defer f.handlePanic(ctx)
if handler, ok := f.routes[string(fastCtx.Method())+" "+string(fastCtx.Path())]; ok { method, path := string(fastCtx.Method()), string(fastCtx.Path())
f.handleRoute(fastCtx, handler) if handler, ok := f.routes[method+" "+path]; ok {
f.handleRoute(fastCtx, handler, "")
return return
} }
for _, r := range f.wildcardRoutes[method] {
if path == r.prefix || strings.HasPrefix(path, r.prefix+"/") {
f.handleRoute(fastCtx, r.handler, strings.TrimPrefix(path, r.prefix+"/"))
return
}
}
if ctx.path() != "/cell" { if ctx.path() != "/cell" {
ctx.setStatusCode(fasthttp.StatusNotFound) ctx.setStatusCode(fasthttp.StatusNotFound)
+22 -6
View File
@@ -16,10 +16,12 @@ type RouteHandler func(ctx *RouteCtx) error
// RouteCtx 自定义路由上下文:Data 合并了 URL 查询参数与 POST 表单参数(表单优先), // RouteCtx 自定义路由上下文:Data 合并了 URL 查询参数与 POST 表单参数(表单优先),
// 支付回调等第三方以 form-urlencoded 回调的场景可直接 Param 取值。 // 支付回调等第三方以 form-urlencoded 回调的场景可直接 Param 取值。
// Wildcard 为通配符路由(/prefix/*)匹配到的剩余路径(不含前导 "/")。
// 独立于服务内嵌的 Ctx:后者辅助方法刻意全小写以防混入 RPC 方法集,路由不复用该类型。 // 独立于服务内嵌的 Ctx:后者辅助方法刻意全小写以防混入 RPC 方法集,路由不复用该类型。
type RouteCtx struct { type RouteCtx struct {
RequestCtx *fasthttp.RequestCtx RequestCtx *fasthttp.RequestCtx
Data map[string]string Data map[string]string
Wildcard string
} }
// Param 取查询/表单参数,不存在返回空串 // Param 取查询/表单参数,不存在返回空串
@@ -27,10 +29,12 @@ func (c *RouteCtx) Param(name string) string {
return c.Data[name] return c.Data[name]
} }
// BindRoute 注册自定义路由(方法大小写不敏感 + 精确路径匹配),用于 GET 直链、 // BindRoute 注册自定义路由(方法大小写不敏感path 精确匹配,或以 "/*" 结尾做前缀通配),
// 健康检查、支付回调等无法走 POST /cell RPC 的场景。 // 用于 GET 直链、健康检查、支付回调等无法走 POST /cell RPC 的场景。
// //
// - path 必须以 "/" 开头;/cell 为 RPC 保留路径,不可注册 // - path 必须以 "/" 开头;/cell 为 RPC 保留路径,不可注册
// - 通配符形式如 "/image/*":匹配 "/image/a/b.png" 等任意子路径,
// 匹配到的剩余路径(去掉前导 "/",如 "a/b.png")经 RouteCtx.Wildcard 取出
// - 同一 方法+路径 重复注册直接 panic // - 同一 方法+路径 重复注册直接 panic
// - 与 BindService 一致,需在 Start 前完成注册(启动阶段单线程) // - 与 BindService 一致,需在 Start 前完成注册(启动阶段单线程)
func (f *Fun) BindRoute(method, path string, handler RouteHandler) { func (f *Fun) BindRoute(method, path string, handler RouteHandler) {
@@ -44,9 +48,21 @@ func (f *Fun) BindRoute(method, path string, handler RouteHandler) {
if !strings.HasPrefix(path, "/") { if !strings.HasPrefix(path, "/") {
panic(fmt.Sprintf("fun: BindRoute path %q must start with '/'", path)) panic(fmt.Sprintf("fun: BindRoute path %q must start with '/'", path))
} }
if path == "/cell" { if path == "/cell" || path == "/cell/*" {
panic("fun: /cell is reserved for RPC") panic("fun: /cell is reserved for RPC")
} }
if prefix, ok := strings.CutSuffix(path, "/*"); ok {
if prefix == "" || strings.HasSuffix(prefix, "/") {
panic(fmt.Sprintf("fun: BindRoute wildcard path %q invalid (no trailing '/' allowed before /*)", path))
}
for _, r := range f.wildcardRoutes[method] {
if r.prefix == prefix {
panic(fmt.Sprintf("fun: route %s %s/* already bound", method, prefix))
}
}
f.wildcardRoutes[method] = append(f.wildcardRoutes[method], wildcardRoute{prefix: prefix, handler: handler})
return
}
key := method + " " + path key := method + " " + path
if _, exists := f.routes[key]; exists { if _, exists := f.routes[key]; exists {
panic(fmt.Sprintf("fun: route %s already bound", key)) panic(fmt.Sprintf("fun: route %s already bound", key))
@@ -55,8 +71,8 @@ func (f *Fun) BindRoute(method, path string, handler RouteHandler) {
} }
// handleRoute 执行自定义路由:合并查询与表单参数(application/x-www-form-urlencoded), // handleRoute 执行自定义路由:合并查询与表单参数(application/x-www-form-urlencoded),
// 处理器返回 error 时按统一 Result 格式输出错误响应 // 处理器返回 error 时按统一 Result 格式输出错误响应;wildcard 为通配路由匹配的剩余路径
func (f *Fun) handleRoute(fastCtx *fasthttp.RequestCtx, handler RouteHandler) { func (f *Fun) handleRoute(fastCtx *fasthttp.RequestCtx, handler RouteHandler, wildcard string) {
data := map[string]string{} data := map[string]string{}
fastCtx.QueryArgs().VisitAll(func(k, v []byte) { fastCtx.QueryArgs().VisitAll(func(k, v []byte) {
data[string(k)] = string(v) data[string(k)] = string(v)
@@ -64,7 +80,7 @@ func (f *Fun) handleRoute(fastCtx *fasthttp.RequestCtx, handler RouteHandler) {
fastCtx.PostArgs().VisitAll(func(k, v []byte) { fastCtx.PostArgs().VisitAll(func(k, v []byte) {
data[string(k)] = string(v) data[string(k)] = string(v)
}) })
if err := handler(&RouteCtx{RequestCtx: fastCtx, Data: data}); err != nil { if err := handler(&RouteCtx{RequestCtx: fastCtx, Data: data, Wildcard: wildcard}); err != nil {
(&Ctx{RequestCtx: fastCtx}).sendError(err) (&Ctx{RequestCtx: fastCtx}).sendError(err)
} }
} }
+109
View File
@@ -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)
}
})
}
}
+375 -51
View File
@@ -11,17 +11,127 @@ func (ctx templateTs) genClientTemplate() string {
status: number; status: number;
}; };
export type resultStatus = 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 = ( export type RequestInterceptor = (
serviceName: string, serviceName: string,
methodName: string, methodName: string,
dto: any state: Record<string, string>,
dto?: any
) => Promise<void> | void; ) => Promise<void> | void;
export type ResponseInterceptor = ( export type ResponseInterceptor = (
serviceName: string, serviceName: string,
methodName: string, methodName: string,
result: result<any> result: result<any>
) => Promise<void> | void; ) => Promise<result<any> | void> | result<any> | 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<any> {
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<any> {
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<any> {
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<any> {
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<any> {
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<any> {
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*(?:<!doctype\s+html|<html\b)/i.test(body)) {
return failure(1, ` + "`Unexpected HTML response: ${excerpt(body)}`" + `);
}
return failure(1, ` + "`Invalid JSON response: ${excerpt(body)}`" + `);
}
if (!isResult(value)) {
return response.ok
? failure(1, "Invalid fun response")
: externalFailure(response, "invalid fun response");
}
return value;
}
export class Client { export class Client {
private url: string; private url: string;
@@ -37,79 +147,293 @@ export class Client {
this.state = state; this.state = state;
} }
addRequestInterceptor(i: RequestInterceptor) { addRequestInterceptor(interceptor: RequestInterceptor) {
this.requestInterceptors.push(i); this.requestInterceptors.push(interceptor);
} }
addResponseInterceptor(i: ResponseInterceptor) { addResponseInterceptor(interceptor: ResponseInterceptor) {
this.responseInterceptors.push(i); this.responseInterceptors.push(interceptor);
} }
async request<T>(serviceName: string, methodName: string, dto?: any): Promise<result<T>> { private async interceptResponse(
for (const i of this.requestInterceptors) { serviceName: string,
await i(serviceName, methodName, dto); methodName: string,
initial: result<any>
): Promise<result<any>> {
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)}`" + `);
} }
const res = await fetch(this.url + "/cell", { }
return current;
}
private async requestState(serviceName: string, methodName: string, dto: any): Promise<Record<string, string>> {
const state: Record<string, string> = { ...this.state };
for (const interceptor of this.requestInterceptors) {
await interceptor(serviceName, methodName, state, dto);
}
return state;
}
async request<T>(
serviceName: string,
methodName: string,
dto?: any,
options?: RequestOptions
): Promise<result<T>> {
let state: Record<string, string>;
try {
state = await this.requestState(serviceName, methodName, dto);
} catch (error) {
return await this.interceptResponse(
serviceName,
methodName,
failure(1, ` + "`Request interceptor failed: ${messageOf(error)}`" + `)
) as result<T>;
}
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<T>;
}
let output: result<any>;
try {
const response = await fetch(` + "`${this.url}/cell`" + `, {
method: "POST", method: "POST",
headers: { "Content-Type": "application/json" }, headers: { "Content-Type": "application/json" },
body: JSON.stringify({ serviceName, methodName, data: dto, ...(Object.keys(this.state).length ? { state: this.state } : {}) }), body,
signal: options?.signal,
}); });
const out = (await res.json()) as result<T>; try {
const anyResult: result<any> = { output = parseResult(response, await response.text());
id: out.id, } catch (error) {
code: out.code, output = responseReadFailure(error, response, options?.signal, false);
data: out.data,
msg: out.msg,
status: out.status,
};
for (const i of this.responseInterceptors) {
await i(serviceName, methodName, anyResult);
} }
return out; } catch (error) {
output = requestFailure(error, options?.signal, false);
}
return await this.interceptResponse(serviceName, methodName, output) as result<T>;
} }
async stream<T>( async stream<T>(
serviceName: string, serviceName: string,
methodName: string, methodName: string,
dto: any | undefined, dto: any | undefined,
onMessage: (data: T) => void onMessage: (data: T) => unknown,
): Promise<void> { options?: StreamOptions
for (const i of this.requestInterceptors) { ): Promise<result<void>> {
await i(serviceName, methodName, dto); let state: Record<string, string>;
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", {
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)}`" + `)
);
}
let response: Response;
try {
response = await fetch(` + "`${this.url}/cell`" + `, {
method: "POST", method: "POST",
headers: { "Content-Type": "application/json" }, headers: { "Content-Type": "application/json" },
body: JSON.stringify({ serviceName, methodName, data: dto, ...(Object.keys(this.state).length ? { state: this.state } : {}) }), body,
signal: options?.signal,
}); });
if (!res.ok) return; } catch (error) {
const anyResult: result<any> = { status: 0 }; return await this.interceptResponse(
for (const i of this.responseInterceptors) { serviceName,
await i(serviceName, methodName, anyResult); methodName,
requestFailure(error, options?.signal, true)
);
} }
if (!res.body) return;
const reader = res.body.getReader(); if (!response.ok) {
const decoder = new TextDecoder(); 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<Uint8Array>;
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 = ""; let buffer = "";
let lineNumber = 0;
let failed: result<any> | 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 (;;) { for (;;) {
const { done, value } = await reader.read(); let part: ReadableStreamReadResult<Uint8Array>;
if (done) break; try {
buffer += decoder.decode(value, { stream: true }); part = await reader.read();
const lines = buffer.split("\n"); } catch (error) {
buffer = lines.pop() ?? ""; cause = error;
for (const line of lines) { if (isTimeout(error, options?.signal)) {
const payload = line.trim(); failed = failure(5, "Stream timed out");
if (!payload) continue; } else if (options?.signal?.aborted === true) {
const data = JSON.parse(payload) as T; failed = failure(4, "Stream aborted");
onMessage(data); } 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 { func (ctx templateTs) genDefaultServiceTemplate() string {
return `import { Client, type result } from "./client"; return `import { Client } from "./client";
{{- range .GenServiceList}} {{- range .GenServiceList}}
import {{.ServiceName}} from "./{{.ServiceName}}"; import {{.ServiceName}} from "./{{.ServiceName}}";
{{- end}} {{- end}}
@@ -131,7 +455,7 @@ export default class api {
} }
func (ctx templateTs) genServiceTemplate() string { 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}} {{- range .GenImport}}
import type {{.Name}} from "./{{.Name}}"; import type {{.Name}} from "./{{.Name}}";
{{- end}} {{- end}}
@@ -143,10 +467,10 @@ export default class {{.ServiceName}} {
} }
{{- $serviceName := .ServiceName }} {{- $serviceName := .ServiceName }}
{{- range .GenMethodTypeList}} {{- range .GenMethodTypeList}}
{{if .IsStream }}async {{.MethodName}}({{.DtoText}}{{if .DtoText}},{{end}}onMessage: (data: {{.GenericTypeText}}) => void): Promise<void> { {{if .IsStream }}async {{.MethodName}}({{if .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) return await this.client.stream<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, onMessage, options)
}{{else}}async {{.MethodName}}({{.DtoText}}): Promise<{{.ReturnValueText}}> { }{{else}}async {{.MethodName}}({{if .DtoText}}{{.DtoText}}, {{end}}options?: RequestOptions): Promise<{{.ReturnValueText}}> {
return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}"{{.ArgsText}}) return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, options)
}{{end}} }{{end}}
{{- end}} {{- end}}
}` }`