From 3739a928840fad8afb8714fc21967a20f235396a Mon Sep 17 00:00:00 2001 From: Alexander Kolberg Date: Fri, 26 Jun 2026 11:04:58 +0300 Subject: [PATCH] Add async iterable stream client option --- README.md | 37 ++ client.go.tmpl | 11 +- clientHelpers.go.tmpl | 9 + clientInterface.go.tmpl | 2 +- clientSSE.go.tmpl | 168 +++++++++ main.go.tmpl | 9 + methodInputs.go.tmpl | 12 + tests-unit/package.json | 3 +- tests-unit/stream-async.gen.ts | 585 ++++++++++++++++++++++++++++++++ tests-unit/stream-async.test.ts | 155 +++++++++ tests-unit/stream.ridl | 15 + 11 files changed, 1002 insertions(+), 4 deletions(-) create mode 100644 tests-unit/stream-async.gen.ts create mode 100644 tests-unit/stream-async.test.ts create mode 100644 tests-unit/stream.ridl diff --git a/README.md b/README.md index 28192f6..0e52cc5 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,42 @@ function UserProfile({ userId }) { The query keys follow the pattern `[ServiceName, methodName, request?]` and are fully type-safe with `as const` assertions. +### Streaming client style + +For RIDL methods declared with `=> stream (...)`, the default generated TypeScript client uses the existing callback API: + +```typescript +client.subscribeMessages(req, { + onMessage(message) {}, + onError(error, reconnect) {}, +}) +``` + +Pass `-streamClient=asyncIterable` to generate stream methods that return `AsyncIterable` instead: + +```typescript +const stream = client.subscribeMessages(req, { signal }) + +for await (const message of stream) { + console.log(message) +} +``` + +This keeps the generated WebRPC client independent of any UI/cache library while making it easy to use with TanStack Query's `streamedQuery` helper: + +```typescript +import { experimental_streamedQuery as streamedQuery } from '@tanstack/react-query' + +useQuery({ + queryKey: client.queryKey.subscribeMessages(req), + queryFn: streamedQuery({ + streamFn: ({ signal }) => client.subscribeMessages(req, { signal }), + initialValue: [], + reducer: (messages, chunk) => [...messages, chunk.message], + }), +}) +``` + ### Enum style: `enum` vs `union` TypeScript best practices are moving away from `enum` declarations — they emit runtime @@ -142,6 +178,7 @@ Change any of the following values by passing `-option="Value"` CLI flag to `web | `-webrpcHeader` | send Webrpc header in all HTTP requests | `true` | v0.15.0 | | `-schemaHash=false` | don't emit schema hash + version consts | `true` | v0.28.0 | | `-enumStyle` | enum codegen style: `enum` or `union` | `enum` | v0.29.0 | +| `-streamClient` | streaming client style: `callback` or `asyncIterable` | `callback` | next | **Note:** Generated code requires ES2022+ runtime environment. diff --git a/client.go.tmpl b/client.go.tmpl index 9f47834..44ca61e 100644 --- a/client.go.tmpl +++ b/client.go.tmpl @@ -56,8 +56,14 @@ export class {{$service.Name}} implements {{$service.Name}}Client { {{- $methodRespName = printf "%sResponse" $method.Name -}} {{- end -}} {{- end}} - {{firstLetterToLower .Name}} = ({{template "methodInputs" dict "Method" . "Opts" $opts "TypeMap" $typeMap}}): {{if $method.StreamOutput}}WebrpcStreamController{{else}}{{if $method.Succinct}}Promise<{{(index $method.Outputs 0).Type}}>{{else}}Promise<{{$method.Name}}{{if $opts.compat}}Return{{else}}Response{{end}}>{{end}}{{end}} => { + {{firstLetterToLower .Name}} = ({{template "methodInputs" dict "Method" . "Opts" $opts "TypeMap" $typeMap}}): {{if $method.StreamOutput}}{{if $opts.streamClientAsyncIterable}}WebrpcStream<{{$methodRespName}}>{{else}}WebrpcStreamController{{end}}{{else}}{{if $method.Succinct}}Promise<{{(index $method.Outputs 0).Type}}>{{else}}Promise<{{$method.Name}}{{if $opts.compat}}Return{{else}}Response{{end}}>{{end}}{{end}} => { {{- if $method.StreamOutput }} + {{- if $opts.streamClientAsyncIterable }} + return webrpcAsyncIterable<{{$methodRespName}}>(() => this.fetch(this.url('{{.Name}}'), + {{if .Inputs | len }}createHttpRequest(JsonEncode(req), options?.headers, options?.signal){{- else}}createHttpRequest('{}', options?.headers, options?.signal){{end }} + ), '{{$methodRespName}}', options?.signal) + } + {{- else }} const abortController = new AbortController() const abortSignal = abortController.signal @@ -81,6 +87,7 @@ export class {{$service.Name}} implements {{$service.Name}}Client { closed: resp } } + {{- end }} {{- else }} return this.fetch( this.url('{{.Name}}'), @@ -102,6 +109,6 @@ export class {{$service.Name}} implements {{$service.Name}}Client { {{- end -}} {{end -}} {{if $opts.streaming}} - {{template "sse"}} + {{template "sse" dict "Opts" $opts}} {{end}} {{- end -}} diff --git a/clientHelpers.go.tmpl b/clientHelpers.go.tmpl index 754896c..88dc535 100644 --- a/clientHelpers.go.tmpl +++ b/clientHelpers.go.tmpl @@ -31,6 +31,14 @@ const buildResponse = (res: Response): Promise => { export type Fetch = (input: RequestInfo, init?: RequestInit) => Promise {{if $opts.streaming}} +{{- if $opts.streamClientAsyncIterable }} +export interface WebrpcOptions { + headers?: HeadersInit; + signal?: AbortSignal; +} + +export type WebrpcStream = AsyncIterable +{{- else }} export interface WebrpcStreamOptions extends WebrpcOptions { onMessage: (message: T) => void; onError: (error: WebrpcError, reconnect: () => void) => void; @@ -47,5 +55,6 @@ export interface WebrpcStreamController { abort: (reason?: any) => void; closed: Promise; } +{{- end }} {{end}} {{end}} diff --git a/clientInterface.go.tmpl b/clientInterface.go.tmpl index a7d3499..01011af 100644 --- a/clientInterface.go.tmpl +++ b/clientInterface.go.tmpl @@ -28,7 +28,7 @@ export interface {{$service.Name}}Client { * @deprecated {{ $deprecated.Value }} */ {{- end }} - {{firstLetterToLower $method.Name}}({{template "methodInputs" dict "Method" $method "TypeMap" $typeMap "Opts" $opts}}): {{if $method.StreamOutput}}WebrpcStreamController{{else}}{{if $method.Succinct}}Promise<{{(index $method.Outputs 0).Type}}>{{else}}Promise<{{$method.Name}}{{if $opts.compat}}Return{{else}}Response{{end}}>{{end}}{{end}} + {{firstLetterToLower $method.Name}}({{template "methodInputs" dict "Method" $method "TypeMap" $typeMap "Opts" $opts}}): {{if $method.StreamOutput}}{{if $opts.streamClientAsyncIterable}}WebrpcStream<{{if $method.Succinct}}{{(index $method.Outputs 0).Type}}{{else}}{{$method.Name}}{{if $opts.compat}}Return{{else}}Response{{end}}{{end}}>{{else}}WebrpcStreamController{{end}}{{else}}{{if $method.Succinct}}Promise<{{(index $method.Outputs 0).Type}}>{{else}}Promise<{{$method.Name}}{{if $opts.compat}}Return{{else}}Response{{end}}>{{end}}{{end}} {{- if lt (add $i 1) (len $service.Methods)}}{{"\n"}}{{end}} {{- end}} } diff --git a/clientSSE.go.tmpl b/clientSSE.go.tmpl index 8875431..68ffe89 100644 --- a/clientSSE.go.tmpl +++ b/clientSSE.go.tmpl @@ -1,4 +1,171 @@ {{ define "sse" }} +{{- $opts := .Opts -}} +{{- if $opts.streamClientAsyncIterable }} +const WebrpcStreamReadTimeoutMs = (10 + 1) * 1000 + +const webrpcAsyncIterable = ( + fetchResponse: () => Promise, + responseType: string, + signal?: AbortSignal +): WebrpcStream => ({ + async *[Symbol.asyncIterator]() { + let res: Response + try { + res = await fetchResponse() + } catch (error) { + if (signal?.aborted) { + return + } + throw WebrpcRequestFailedError.new({ cause: `fetch(): ${error instanceof Error ? error.message : String(error)}` }) + } + + yield* readWebrpcStream(res, responseType, signal) + } +}) + +const readWebrpcStream = async function* ( + res: Response, + responseType: string, + signal?: AbortSignal +): AsyncGenerator { + if (!res.ok) { + await buildResponse(res) + return + } + + if (!res.body) { + throw WebrpcBadResponseError.new({ + status: res.status, + cause: "Invalid response, missing body", + }) + } + + const reader = res.body.getReader() + const decoder = new TextDecoder() + let buffer = "" + + try { + while (true) { + const { value, done } = await readWebrpcStreamChunk(reader, signal) + + if (done) { + buffer += decoder.decode() + if (buffer.trim().length > 0) { + yield decodeWebrpcStreamLine(buffer, responseType, res.status) + } + return + } + + buffer += decoder.decode(value, { stream: true }) + + const lines = buffer.split("\n") + buffer = lines.pop() || "" + + for (const line of lines) { + if (line.trim().length === 0) { + continue + } + yield decodeWebrpcStreamLine(line, responseType, res.status) + } + } + } catch (error) { + if (error instanceof WebrpcError) { + try { + await reader.cancel(error) + } catch (_) { + // Ignore cleanup errors and report the protocol error below. + } + throw error + } + + if ( + signal?.aborted || + (error instanceof DOMException && error.name === "AbortError") + ) { + return + } + + try { + await reader.cancel(error) + } catch (_) { + // Ignore cleanup errors and report the original stream error below. + } + + throw WebrpcStreamLostError.new({ + cause: `reader.read(): ${error instanceof Error ? error.message : String(error)}`, + }) + } finally { + try { + reader.releaseLock() + } catch (_) { + // Ignore releaseLock errors; the stream is already finishing. + } + } +} + +const readWebrpcStreamChunk = ( + reader: ReadableStreamDefaultReader, + signal?: AbortSignal +): Promise> => { + return new Promise((resolve, reject) => { + const timeoutId = setTimeout(() => { + cleanup() + reject(new Error("Timeout, no data or heartbeat received")) + }, WebrpcStreamReadTimeoutMs) + + const abort = () => { + cleanup() + reject(new DOMException("AbortError", "AbortError")) + } + + const cleanup = () => { + clearTimeout(timeoutId) + signal?.removeEventListener("abort", abort) + } + + if (signal?.aborted) { + abort() + return + } + + signal?.addEventListener("abort", abort, { once: true }) + + reader.read().then( + (result) => { + cleanup() + resolve(result) + }, + (error) => { + cleanup() + reject(error) + } + ) + }) +} + +const decodeWebrpcStreamLine = (line: string, responseType: string, status: number): T => { + let data: any + try { + data = JsonDecode(line, responseType) + } catch (error) { + if (error instanceof WebrpcError) { + throw error + } + throw WebrpcBadResponseError.new({ + status, + cause: `JsonDecode(): ${error instanceof Error ? error.message : String(error)}`, + }) + } + + if (data && Object.prototype.hasOwnProperty.call(data, "webrpcError")) { + const error = data.webrpcError + const code: number = typeof error.code === "number" ? error.code : 0 + throw (webrpcErrorByCode[code] || WebrpcError).new(error) + } + + return data as T +} +{{- else }} const sseResponse = async ( res: Response, options: WebrpcStreamOptions, @@ -120,4 +287,5 @@ const sseResponse = async ( return; } }; +{{- end }} {{ end }} diff --git a/main.go.tmpl b/main.go.tmpl index 6d91006..d317924 100644 --- a/main.go.tmpl +++ b/main.go.tmpl @@ -8,6 +8,7 @@ {{- set $opts "webrpcHeader" (ternary (eq (default .Opts.webrpcHeader "true") "false") false true) -}} {{- set $opts "schemaHash" (ternary (eq (default .Opts.schemaHash "true") "false") false true) -}} {{- set $opts "enumStyle" (default .Opts.enumStyle "enum") -}} +{{- set $opts "streamClient" (default .Opts.streamClient "callback") -}} {{- /* Print help on -help. */ -}} {{- if exists .Opts "help" -}} @@ -29,6 +30,14 @@ {{- exit 1 -}} {{- end -}} +{{- if not (in $opts.streamClient "callback" "asyncIterable") -}} + {{- stderrPrintf "-streamClient=%q is not supported, must be \"callback\" or \"asyncIterable\"\n" $opts.streamClient -}} + {{- exit 1 -}} +{{- end -}} + +{{- set $opts "streamClientCallback" (eq $opts.streamClient "callback") -}} +{{- set $opts "streamClientAsyncIterable" (eq $opts.streamClient "asyncIterable") -}} + {{- if ne .WebrpcVersion "v1" -}} {{- stderrPrintf "%s generator error: unsupported Webrpc version %s\n" .WebrpcTarget .WebrpcVersion -}} {{- exit 1 -}} diff --git a/methodInputs.go.tmpl b/methodInputs.go.tmpl index fb783fa..e21c21c 100644 --- a/methodInputs.go.tmpl +++ b/methodInputs.go.tmpl @@ -8,7 +8,11 @@ {{- if gt (len $method.Inputs) 0}}req: {{(index $method.Inputs 0).Type}}, {{end}} {{- if $method.StreamOutput -}} + {{- if $opts.streamClientAsyncIterable -}} + options?: WebrpcOptions + {{- else -}} options: WebrpcStreamOptions<{{(index $method.Outputs 0).Type}}> + {{- end -}} {{- else -}} headers?: object, signal?: AbortSignal {{- end -}} @@ -17,7 +21,11 @@ {{- if gt (len $method.Inputs) 0}}req: {{$method.Name}}Args, {{end}} {{- if $method.StreamOutput -}} + {{- if $opts.streamClientAsyncIterable -}} + options?: WebrpcOptions + {{- else -}} options: WebrpcStreamOptions<{{$method.Name}}Return> + {{- end -}} {{- else -}} headers?: object, signal?: AbortSignal {{- end -}} @@ -26,7 +34,11 @@ {{- if gt (len $method.Inputs) 0}}req: {{$method.Name}}Request, {{end}} {{- if $method.StreamOutput -}} + {{- if $opts.streamClientAsyncIterable -}} + options?: WebrpcOptions + {{- else -}} options: WebrpcStreamOptions<{{$method.Name}}Response> + {{- end -}} {{- else -}} headers?: object, signal?: AbortSignal {{- end -}} diff --git a/tests-unit/package.json b/tests-unit/package.json index a4e417c..8aae7f5 100644 --- a/tests-unit/package.json +++ b/tests-unit/package.json @@ -9,7 +9,8 @@ "test:watch": "vitest", "test:update": "vitest run -u", "generate": "webrpc-gen -schema=service.ridl -target=../ -out=client.gen.ts", - "generate:enum": "webrpc-gen -schema=enum.ridl -target=../ -client -out=enum-default.gen.ts && webrpc-gen -schema=enum.ridl -target=../ -client -enumStyle=union -out=enum-union.gen.ts" + "generate:enum": "webrpc-gen -schema=enum.ridl -target=../ -client -out=enum-default.gen.ts && webrpc-gen -schema=enum.ridl -target=../ -client -enumStyle=union -out=enum-union.gen.ts", + "generate:stream": "webrpc-gen -schema=stream.ridl -target=../ -client -streamClient=asyncIterable -out=stream-async.gen.ts" }, "devDependencies": { "@tsconfig/strictest": "^2.0.7", diff --git a/tests-unit/stream-async.gen.ts b/tests-unit/stream-async.gen.ts new file mode 100644 index 0000000..cd3a4b7 --- /dev/null +++ b/tests-unit/stream-async.gen.ts @@ -0,0 +1,585 @@ +/* eslint-disable */ +// stream-async-unit-tests v1.0.0 fcb26070730539e1416474fd474fac962ab1eca6 +// -- +// Code generated by Webrpc-gen@v0.38.0 with ../ generator. DO NOT EDIT. +// +// webrpc-gen -schema=stream.ridl -target=../ -client -streamClient=asyncIterable -out=stream-async.gen.ts +// Webrpc description and code-gen version +export const WebrpcVersion = "v1" + +// Schema version of your RIDL schema +export const WebrpcSchemaVersion = "v1.0.0" + +// Schema hash generated from your RIDL schema +export const WebrpcSchemaHash = "fcb26070730539e1416474fd474fac962ab1eca6" + +// +// Client interface +// + +export interface StreamTestClient { + streamMessages(req: StreamMessagesRequest, options?: WebrpcOptions): WebrpcStream + + streamTicks(options?: WebrpcOptions): WebrpcStream +} + + +// +// Schema types +// + +export interface Message { + id: number + text: string +} + +export interface StreamMessagesRequest { + roomId: string +} + +export interface StreamMessagesResponse { + message: Message +} + +export interface StreamTicksRequest { +} + +export interface StreamTicksResponse { + tick: number +} + + + +// +// Client +// + +export class StreamTest implements StreamTestClient { + protected hostname: string + protected fetch: Fetch + protected path = '/rpc/StreamTest/' + + constructor(hostname: string, fetch: Fetch) { + this.hostname = hostname.replace(/\/*$/, '') + this.fetch = (input: RequestInfo, init?: RequestInit) => fetch(input, init) + } + + private url(name: string): string { + return this.hostname + this.path + name + } + + queryKey = { + streamMessages: (req: StreamMessagesRequest) => ['StreamTest', 'streamMessages', req] as const, + streamTicks: () => ['StreamTest', 'streamTicks'] as const, + } + + streamMessages = (req: StreamMessagesRequest, options?: WebrpcOptions): WebrpcStream => { + return webrpcAsyncIterable(() => this.fetch(this.url('StreamMessages'), + createHttpRequest(JsonEncode(req), options?.headers, options?.signal) + ), 'StreamMessagesResponse', options?.signal) + } + streamTicks = (options?: WebrpcOptions): WebrpcStream => { + return webrpcAsyncIterable(() => this.fetch(this.url('StreamTicks'), + createHttpRequest('{}', options?.headers, options?.signal) + ), 'StreamTicksResponse', options?.signal) + } +} + +const WebrpcStreamReadTimeoutMs = (10 + 1) * 1000 + +const webrpcAsyncIterable = ( + fetchResponse: () => Promise, + responseType: string, + signal?: AbortSignal +): WebrpcStream => ({ + async *[Symbol.asyncIterator]() { + let res: Response + try { + res = await fetchResponse() + } catch (error) { + if (signal?.aborted) { + return + } + throw WebrpcRequestFailedError.new({ cause: `fetch(): ${error instanceof Error ? error.message : String(error)}` }) + } + + yield* readWebrpcStream(res, responseType, signal) + } +}) + +const readWebrpcStream = async function* ( + res: Response, + responseType: string, + signal?: AbortSignal +): AsyncGenerator { + if (!res.ok) { + await buildResponse(res) + return + } + + if (!res.body) { + throw WebrpcBadResponseError.new({ + status: res.status, + cause: "Invalid response, missing body", + }) + } + + const reader = res.body.getReader() + const decoder = new TextDecoder() + let buffer = "" + + try { + while (true) { + const { value, done } = await readWebrpcStreamChunk(reader, signal) + + if (done) { + buffer += decoder.decode() + if (buffer.trim().length > 0) { + yield decodeWebrpcStreamLine(buffer, responseType, res.status) + } + return + } + + buffer += decoder.decode(value, { stream: true }) + + const lines = buffer.split("\n") + buffer = lines.pop() || "" + + for (const line of lines) { + if (line.trim().length === 0) { + continue + } + yield decodeWebrpcStreamLine(line, responseType, res.status) + } + } + } catch (error) { + if (error instanceof WebrpcError) { + try { + await reader.cancel(error) + } catch (_) { + // Ignore cleanup errors and report the protocol error below. + } + throw error + } + + if ( + signal?.aborted || + (error instanceof DOMException && error.name === "AbortError") + ) { + return + } + + try { + await reader.cancel(error) + } catch (_) { + // Ignore cleanup errors and report the original stream error below. + } + + throw WebrpcStreamLostError.new({ + cause: `reader.read(): ${error instanceof Error ? error.message : String(error)}`, + }) + } finally { + try { + reader.releaseLock() + } catch (_) { + // Ignore releaseLock errors; the stream is already finishing. + } + } +} + +const readWebrpcStreamChunk = ( + reader: ReadableStreamDefaultReader, + signal?: AbortSignal +): Promise> => { + return new Promise((resolve, reject) => { + const timeoutId = setTimeout(() => { + cleanup() + reject(new Error("Timeout, no data or heartbeat received")) + }, WebrpcStreamReadTimeoutMs) + + const abort = () => { + cleanup() + reject(new DOMException("AbortError", "AbortError")) + } + + const cleanup = () => { + clearTimeout(timeoutId) + signal?.removeEventListener("abort", abort) + } + + if (signal?.aborted) { + abort() + return + } + + signal?.addEventListener("abort", abort, { once: true }) + + reader.read().then( + (result) => { + cleanup() + resolve(result) + }, + (error) => { + cleanup() + reject(error) + } + ) + }) +} + +const decodeWebrpcStreamLine = (line: string, responseType: string, status: number): T => { + let data: any + try { + data = JsonDecode(line, responseType) + } catch (error) { + if (error instanceof WebrpcError) { + throw error + } + throw WebrpcBadResponseError.new({ + status, + cause: `JsonDecode(): ${error instanceof Error ? error.message : String(error)}`, + }) + } + + if (data && Object.prototype.hasOwnProperty.call(data, "webrpcError")) { + const error = data.webrpcError + const code: number = typeof error.code === "number" ? error.code : 0 + throw (webrpcErrorByCode[code] || WebrpcError).new(error) + } + + return data as T +} + + + +const createHttpRequest = (body: string = '{}', headers: object = {}, signal: AbortSignal | null = null): object => { + const reqHeaders: { [key: string]: string } = { ...headers, 'Content-Type': 'application/json', [WebrpcHeader]: WebrpcHeaderValue } + return { method: 'POST', headers: reqHeaders, body, signal } +} + +const buildResponse = (res: Response): Promise => { + return res.text().then(text => { + let data + try { + data = JSON.parse(text) + } catch(error) { + throw WebrpcBadResponseError.new({ + status: res.status, + cause: `JSON.parse(): ${error instanceof Error ? error.message : String(error)}: response text: ${text}`}, + ) + } + if (!res.ok) { + const code: number = (typeof data.code === 'number') ? data.code : 0 + throw (webrpcErrorByCode[code] || WebrpcError).new(data) + } + return data + }) +} + +export type Fetch = (input: RequestInfo, init?: RequestInit) => Promise + + +export interface WebrpcOptions { + headers?: HeadersInit; + signal?: AbortSignal; +} + +export type WebrpcStream = AsyncIterable + + + + + +export const JsonEncode = (obj: T): string => { + return JSON.stringify(obj) +} + +export const JsonDecode = (data: string | any, _typ: string = ''): T => { + let parsed: any = data + if (typeof data === 'string') { + try { parsed = JSON.parse(data) } catch (err) { + throw WebrpcBadResponseError.new({ cause: `JsonDecode: JSON.parse failed: ${(err as Error).message}` }) + } + } + return parsed as T +} + + +// +// Errors +// + +type WebrpcErrorParams = { name?: string, code?: number, message?: string, status?: number, cause?: string } + +export class WebrpcError extends Error { + code: number + status: number + + constructor(error: WebrpcErrorParams = {}) { + super(error.message) + this.name = error.name || 'WebrpcEndpointError' + this.code = typeof error.code === 'number' ? error.code : 0 + this.message = error.message || `endpoint error` + this.status = typeof error.status === 'number' ? error.status : 400 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcError.prototype) + } + + static new(payload: any): WebrpcError { + return new this({ message: payload.message, code: payload.code, status: payload.status, cause: payload.cause }) + } +} + + +export class WebrpcEndpointError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcEndpoint' + this.code = typeof error.code === 'number' ? error.code : 0 + this.message = error.message || `endpoint error` + this.status = typeof error.status === 'number' ? error.status : 400 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcEndpointError.prototype) + } +} + +export class WebrpcRequestFailedError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcRequestFailed' + this.code = typeof error.code === 'number' ? error.code : -1 + this.message = error.message || `request failed` + this.status = typeof error.status === 'number' ? error.status : 400 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcRequestFailedError.prototype) + } +} + +export class WebrpcBadRouteError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcBadRoute' + this.code = typeof error.code === 'number' ? error.code : -2 + this.message = error.message || `bad route` + this.status = typeof error.status === 'number' ? error.status : 404 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcBadRouteError.prototype) + } +} + +export class WebrpcBadMethodError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcBadMethod' + this.code = typeof error.code === 'number' ? error.code : -3 + this.message = error.message || `bad method` + this.status = typeof error.status === 'number' ? error.status : 405 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcBadMethodError.prototype) + } +} + +export class WebrpcBadRequestError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcBadRequest' + this.code = typeof error.code === 'number' ? error.code : -4 + this.message = error.message || `bad request` + this.status = typeof error.status === 'number' ? error.status : 400 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcBadRequestError.prototype) + } +} + +export class WebrpcBadResponseError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcBadResponse' + this.code = typeof error.code === 'number' ? error.code : -5 + this.message = error.message || `bad response` + this.status = typeof error.status === 'number' ? error.status : 500 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcBadResponseError.prototype) + } +} + +export class WebrpcServerPanicError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcServerPanic' + this.code = typeof error.code === 'number' ? error.code : -6 + this.message = error.message || `server panic` + this.status = typeof error.status === 'number' ? error.status : 500 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcServerPanicError.prototype) + } +} + +export class WebrpcInternalErrorError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcInternalError' + this.code = typeof error.code === 'number' ? error.code : -7 + this.message = error.message || `internal error` + this.status = typeof error.status === 'number' ? error.status : 500 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcInternalErrorError.prototype) + } +} + +export class WebrpcClientAbortedError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcClientAborted' + this.code = typeof error.code === 'number' ? error.code : -8 + this.message = error.message || `request aborted by client` + this.status = typeof error.status === 'number' ? error.status : 400 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcClientAbortedError.prototype) + } +} + +export class WebrpcStreamLostError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcStreamLost' + this.code = typeof error.code === 'number' ? error.code : -9 + this.message = error.message || `stream lost` + this.status = typeof error.status === 'number' ? error.status : 400 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcStreamLostError.prototype) + } +} + +export class WebrpcStreamFinishedError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'WebrpcStreamFinished' + this.code = typeof error.code === 'number' ? error.code : -10 + this.message = error.message || `stream finished` + this.status = typeof error.status === 'number' ? error.status : 200 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, WebrpcStreamFinishedError.prototype) + } +} + + +// +// Schema errors +// + +export class StreamBoomError extends WebrpcError { + constructor(error: WebrpcErrorParams = {}) { + super(error) + this.name = error.name || 'StreamBoom' + this.code = typeof error.code === 'number' ? error.code : 100 + this.message = error.message || `stream boom` + this.status = typeof error.status === 'number' ? error.status : 500 + if (error.cause !== undefined) this.cause = error.cause + Object.setPrototypeOf(this, StreamBoomError.prototype) + } +} + +export enum errors { + WebrpcEndpoint = 'WebrpcEndpoint', + WebrpcRequestFailed = 'WebrpcRequestFailed', + WebrpcBadRoute = 'WebrpcBadRoute', + WebrpcBadMethod = 'WebrpcBadMethod', + WebrpcBadRequest = 'WebrpcBadRequest', + WebrpcBadResponse = 'WebrpcBadResponse', + WebrpcServerPanic = 'WebrpcServerPanic', + WebrpcInternalError = 'WebrpcInternalError', + WebrpcClientAborted = 'WebrpcClientAborted', + WebrpcStreamLost = 'WebrpcStreamLost', + WebrpcStreamFinished = 'WebrpcStreamFinished', + StreamBoom = 'StreamBoom', +} + +export enum WebrpcErrorCodes { + WebrpcEndpoint = 0, + WebrpcRequestFailed = -1, + WebrpcBadRoute = -2, + WebrpcBadMethod = -3, + WebrpcBadRequest = -4, + WebrpcBadResponse = -5, + WebrpcServerPanic = -6, + WebrpcInternalError = -7, + WebrpcClientAborted = -8, + WebrpcStreamLost = -9, + WebrpcStreamFinished = -10, + StreamBoom = 100, +} + +export const webrpcErrorByCode: { [code: number]: any } = { + [0]: WebrpcEndpointError, + [-1]: WebrpcRequestFailedError, + [-2]: WebrpcBadRouteError, + [-3]: WebrpcBadMethodError, + [-4]: WebrpcBadRequestError, + [-5]: WebrpcBadResponseError, + [-6]: WebrpcServerPanicError, + [-7]: WebrpcInternalErrorError, + [-8]: WebrpcClientAbortedError, + [-9]: WebrpcStreamLostError, + [-10]: WebrpcStreamFinishedError, + [100]: StreamBoomError, +} + + + +// +// Webrpc +// + +export const WebrpcHeader = "Webrpc" + +export const WebrpcHeaderValue = "webrpc@v0.38.0;@unknown;stream-async-unit-tests@v1.0.0" + +type WebrpcGenVersions = { + WebrpcGenVersion: string; + codeGenName: string; + codeGenVersion: string; + schemaName: string; + schemaVersion: string; +}; + +export function VersionFromHeader(headers: Headers): WebrpcGenVersions { + const headerValue = headers.get(WebrpcHeader) + if (!headerValue) { + return { + WebrpcGenVersion: "", + codeGenName: "", + codeGenVersion: "", + schemaName: "", + schemaVersion: "", + }; + } + + return parseWebrpcGenVersions(headerValue) +} + +function parseWebrpcGenVersions(header: string): WebrpcGenVersions { + const versions = header.split(";") + if (versions.length < 3) { + return { + WebrpcGenVersion: "", + codeGenName: "", + codeGenVersion: "", + schemaName: "", + schemaVersion: "", + }; + } + + const [_, WebrpcGenVersion] = versions[0]!.split("@") + const [codeGenName, codeGenVersion] = versions[1]!.split("@") + const [schemaName, schemaVersion] = versions[2]!.split("@") + + return { + WebrpcGenVersion: WebrpcGenVersion ?? "", + codeGenName: codeGenName ?? "", + codeGenVersion: codeGenVersion ?? "", + schemaName: schemaName ?? "", + schemaVersion: schemaVersion ?? "", + }; +} + diff --git a/tests-unit/stream-async.test.ts b/tests-unit/stream-async.test.ts new file mode 100644 index 0000000..937501f --- /dev/null +++ b/tests-unit/stream-async.test.ts @@ -0,0 +1,155 @@ +import { describe, expect, it, vi } from 'vitest' +import { + StreamBoomError, + StreamTest, + WebrpcStreamLostError, +} from './stream-async.gen.js' +import type { Fetch } from './stream-async.gen.js' + +const encoder = new TextEncoder() + +const streamBody = ( + chunks: string[], + options: { close?: boolean; error?: unknown } = {}, +): ReadableStream => { + const { close = true, error } = options + + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)) + } + + if (error) { + controller.error(error) + } else if (close) { + controller.close() + } + }, + }) +} + +const streamResponse = ( + chunks: string[], + options: { close?: boolean; error?: unknown; status?: number } = {}, +): Response => { + return new Response(streamBody(chunks, options), { + status: options.status ?? 200, + headers: { 'Content-Type': 'application/x-ndjson' }, + }) +} + +const collect = async (iterable: AsyncIterable): Promise => { + const values: T[] = [] + for await (const value of iterable) { + values.push(value) + } + return values +} + +describe('async iterable stream client', () => { + it('streams every decoded line from a generated stream method', async () => { + const first = JSON.stringify({ message: { id: 1, text: 'hello' } }) + '\n' + const second = JSON.stringify({ message: { id: 2, text: 'world' } }) + '\n' + const mockFetch = vi.fn((_input: RequestInfo, _init?: RequestInit) => + Promise.resolve(streamResponse(['\n', first.slice(0, 12), first.slice(12), second])), + ) + const api = new StreamTest('https://api.test/', mockFetch as Fetch) + + const messages = await collect( + api.streamMessages( + { roomId: 'room-1' }, + { headers: { 'X-Test': 'yes' } }, + ), + ) + + expect(messages).toEqual([ + { message: { id: 1, text: 'hello' } }, + { message: { id: 2, text: 'world' } }, + ]) + expect(mockFetch).toHaveBeenCalledTimes(1) + + const [url, init] = mockFetch.mock.calls[0]! + expect(url).toBe('https://api.test/rpc/StreamTest/StreamMessages') + expect(JSON.parse(String(init?.body))).toEqual({ roomId: 'room-1' }) + expect(init?.headers).toMatchObject({ + 'Content-Type': 'application/json', + 'X-Test': 'yes', + }) + }) + + it('supports stream methods without request inputs', async () => { + const mockFetch = vi.fn((_input: RequestInfo, _init?: RequestInit) => + Promise.resolve( + streamResponse([ + JSON.stringify({ tick: 1 }) + '\n', + JSON.stringify({ tick: 2 }) + '\n', + ]), + ), + ) + const api = new StreamTest('https://api.test', mockFetch as Fetch) + + await expect(collect(api.streamTicks())).resolves.toEqual([ + { tick: 1 }, + { tick: 2 }, + ]) + + const [url, init] = mockFetch.mock.calls[0]! + expect(url).toBe('https://api.test/rpc/StreamTest/StreamTicks') + expect(init?.body).toBe('{}') + }) + + it('throws typed webrpc errors emitted inside the stream', async () => { + const mockFetch = vi.fn(() => + Promise.resolve( + streamResponse([ + JSON.stringify({ + webrpcError: { + code: 100, + message: 'stream boom', + status: 500, + }, + }) + '\n', + ]), + ), + ) + const api = new StreamTest('https://api.test', mockFetch as Fetch) + + await expect(collect(api.streamMessages({ roomId: 'room-1' }))).rejects.toBeInstanceOf( + StreamBoomError, + ) + }) + + it('throws WebrpcStreamLostError when the response stream errors', async () => { + const mockFetch = vi.fn(() => + Promise.resolve( + streamResponse([JSON.stringify({ tick: 1 }) + '\n'], { + error: new Error('socket closed'), + }), + ), + ) + const api = new StreamTest('https://api.test', mockFetch as Fetch) + + await expect(collect(api.streamTicks())).rejects.toBeInstanceOf( + WebrpcStreamLostError, + ) + }) + + it('finishes cleanly when the provided AbortSignal is aborted', async () => { + const abortController = new AbortController() + const mockFetch = vi.fn(() => + Promise.resolve( + streamResponse([JSON.stringify({ tick: 1 }) + '\n'], { close: false }), + ), + ) + const api = new StreamTest('https://api.test', mockFetch as Fetch) + const ticks = [] + + for await (const tick of api.streamTicks({ signal: abortController.signal })) { + ticks.push(tick) + abortController.abort() + } + + expect(ticks).toEqual([{ tick: 1 }]) + }) +}) diff --git a/tests-unit/stream.ridl b/tests-unit/stream.ridl new file mode 100644 index 0000000..dd6909b --- /dev/null +++ b/tests-unit/stream.ridl @@ -0,0 +1,15 @@ +webrpc = v1 + +name = stream-async-unit-tests +version = v1.0.0 +basepath = /rpc + +struct Message + - id: uint64 + - text: string + +service StreamTest + - StreamMessages(roomId: string) => stream (message: Message) + - StreamTicks() => stream (tick: uint64) + +error 100 StreamBoom "stream boom" HTTP 500