SSE: Add abort and closed methods

_examples/node-ts/server/server.gen.ts

@@ -1,13 +1,13 @@
 /* eslint-disable */
 // node-ts v1.0.0 6713366104e62b8479d628a193e2a7ca03f37edc
 // --
-// Code generated by webrpc-gen@v0.21.0-1-g96d144e with ../../../gen-typescript generator. DO NOT EDIT.
+// Code generated by webrpc-gen@v0.21.1 with ../../../gen-typescript generator. DO NOT EDIT.
 //
 // webrpc-gen -schema=service.ridl -target=../../../gen-typescript -server -out=./server/server.gen.ts
 
 export const WebrpcHeader = "Webrpc"
 
-export const WebrpcHeaderValue = "webrpc@v0.21.0-1-g96d144e;gen-typescript@unknown;node-ts@v1.0.0"
+export const WebrpcHeaderValue = "webrpc@v0.21.1;gen-typescript@unknown;node-ts@v1.0.0"
 
 // WebRPC description and code-gen version
 export const WebRPCVersion = "v1"

_examples/node-ts/webapp/client.gen.ts

@@ -1,13 +1,13 @@
 /* eslint-disable */
 // node-ts v1.0.0 6713366104e62b8479d628a193e2a7ca03f37edc
 // --
-// Code generated by webrpc-gen@v0.21.0-1-g96d144e with ../../../gen-typescript generator. DO NOT EDIT.
+// Code generated by webrpc-gen@v0.21.1 with ../../../gen-typescript generator. DO NOT EDIT.
 //
 // webrpc-gen -schema=service.ridl -target=../../../gen-typescript -client -out=./webapp/client.gen.ts
 
 export const WebrpcHeader = "Webrpc"
 
-export const WebrpcHeaderValue = "webrpc@v0.21.0-1-g96d144e;gen-typescript@unknown;node-ts@v1.0.0"
+export const WebrpcHeaderValue = "webrpc@v0.21.1;gen-typescript@unknown;node-ts@v1.0.0"
 
 // WebRPC description and code-gen version
 export const WebRPCVersion = "v1"

_examples/sse/webapp/client.gen.ts

@@ -1,13 +1,13 @@
 /* eslint-disable */
 // webrpc-sse-chat v1.0.0 a799dc63b082644f5d003c8881424546aee23a2c
 // --
-// Code generated by webrpc-gen@v0.21.0-1-g96d144e with ../../ generator. DO NOT EDIT.
+// Code generated by webrpc-gen@v0.21.1 with ../../ generator. DO NOT EDIT.
 //
 // webrpc-gen -schema=service.ridl -target=../../ -client -out=./webapp/client.gen.ts
 
 export const WebrpcHeader = "Webrpc"
 
-export const WebrpcHeaderValue = "webrpc@v0.21.0-1-g96d144e;@unknown;webrpc-sse-chat@v1.0.0"
+export const WebrpcHeaderValue = "webrpc@v0.21.1;@unknown;webrpc-sse-chat@v1.0.0"
 
 // WebRPC description and code-gen version
 export const WebRPCVersion = "v1"
@@ -80,8 +80,8 @@ export interface Message {
 
 export interface Chat {
   sendMessage(args: SendMessageArgs, headers?: object, signal?: AbortSignal): Promise<SendMessageReturn>
-  subscribeMessages(args: SubscribeMessagesArgs, options: WebrpcStreamOptions<SubscribeMessagesReturn>): Promise<void>
-  subscribeUsers(options: WebrpcStreamOptions<SubscribeUsersReturn>): Promise<void>
+  subscribeMessages(args: SubscribeMessagesArgs, options: WebrpcStreamOptions<SubscribeMessagesReturn>): WebrpcStreamController
+  subscribeUsers(options: WebrpcStreamOptions<SubscribeUsersReturn>): WebrpcStreamController
 }
 
 export interface SendMessageArgs {
@@ -136,23 +136,51 @@ export class Chat implements Chat {
     })
   }
 
-  subscribeMessages = (args: SubscribeMessagesArgs, options: WebrpcStreamOptions<SubscribeMessagesReturn>): Promise<void> => {
-    const _fetch = () => this.fetch(this.url('SubscribeMessages'),createHTTPRequest(args, options.headers, options.signal)
+  subscribeMessages = (args: SubscribeMessagesArgs, options: WebrpcStreamOptions<SubscribeMessagesReturn>): WebrpcStreamController => {
+    const abortController = new AbortController();
+    const abortSignal = abortController.signal
+
+    if (options.signal) {
+      abortSignal.addEventListener("abort", () => abortController.abort(options.signal?.reason), {
+        signal: options.signal,
+      });
+    }
+
+    const _fetch = () => this.fetch(this.url('SubscribeMessages'),createHTTPRequest(args, options.headers, abortSignal)
       ).then(async (res) => {
         await sseResponse(res, options, _fetch);
     }, (error) => {
       options.onError(error, _fetch);
     });
-    return _fetch();
+
+    const resp = _fetch();
+    return {
+      abort: abortController.abort.bind(abortController),
+      closed: resp
+    };
   }
-  subscribeUsers = (options: WebrpcStreamOptions<SubscribeUsersReturn>): Promise<void> => {
+  subscribeUsers = (options: WebrpcStreamOptions<SubscribeUsersReturn>): WebrpcStreamController => {
+    const abortController = new AbortController();
+    const abortSignal = abortController.signal
+
+    if (options.signal) {
+      abortSignal.addEventListener("abort", () => abortController.abort(options.signal?.reason), {
+        signal: options.signal,
+      });
+    }
+
     const _fetch = () => this.fetch(this.url('SubscribeUsers'),createHTTPRequest({}, options.headers, options.signal)
       ).then(async (res) => {
         await sseResponse(res, options, _fetch);
     }, (error) => {
       options.onError(error, _fetch);
     });
-    return _fetch();
+
+    const resp = _fetch();
+    return {
+      abort: abortController.abort.bind(abortController),
+      closed: resp
+    };
   }
 }
 
@@ -549,8 +577,14 @@ export interface WebrpcStreamOptions<T> extends WebrpcOptions {
   onOpen?: () => void;
   onClose?: () => void;
 }
+
 export interface WebrpcOptions {
   headers?: HeadersInit;
   signal?: AbortSignal;
 }
 
+export interface WebrpcStreamController {
+  abort: (reason?: any) => void;
+  closed: Promise<void>;
+}
+

client.go.tmpl

@@ -23,16 +23,30 @@ export class {{.Name}} implements {{.Name}} {
     return this.hostname + this.path + name
   }
   {{range $_, $method := .Methods}}
-  {{firstLetterToLower .Name}} = ({{template "methodInputs" dict "Method" . "TypeMap" $typeMap}}): Promise<{{if $method.StreamOutput}}void{{else}}{{$method.Name}}Return{{end}}> => {
+  {{firstLetterToLower .Name}} = ({{template "methodInputs" dict "Method" . "TypeMap" $typeMap}}): {{if $method.StreamOutput}}WebrpcStreamController{{else}}Promise<{{$method.Name}}Return>{{end}} => {
   {{- if $method.StreamOutput }}
+    const abortController = new AbortController();
+    const abortSignal = abortController.signal
+
+    if (options.signal) {
+      abortSignal.addEventListener("abort", () => abortController.abort(options.signal?.reason), {
+        signal: options.signal,
+      });
+    }
+
     const _fetch = () => this.fetch(this.url('{{.Name}}'),
-      {{- if .Inputs | len }}createHTTPRequest(args, options.headers, options.signal){{- else}}createHTTPRequest({}, options.headers, options.signal){{end }}
+      {{- if .Inputs | len }}createHTTPRequest(args, options.headers, abortSignal){{- else}}createHTTPRequest({}, options.headers, options.signal){{end }}
       ).then(async (res) => {
         await sseResponse(res, options, _fetch);
     }, (error) => {
       options.onError(error, _fetch);
     });
-    return _fetch();
+
+    const resp = _fetch();
+    return {
+      abort: abortController.abort.bind(abortController),
+      closed: resp
+    };
   }
   {{- else }}
     return this.fetch(

clientHelpers.go.tmpl

@@ -127,9 +127,15 @@ export interface WebrpcStreamOptions<T> extends WebrpcOptions {
   onOpen?: () => void;
   onClose?: () => void;
 }
+
 export interface WebrpcOptions {
   headers?: HeadersInit;
   signal?: AbortSignal;
 }
+
+export interface WebrpcStreamController {
+  abort: (reason?: any) => void;
+  closed: Promise<void>;
+}
 {{end}}
 {{end}}

types.go.tmpl

@@ -44,7 +44,7 @@ export interface {{$type.Name}} {
 {{- range $_, $service := $services}}
 export interface {{$service.Name}} {
 {{- range $_, $method := $service.Methods}}
-  {{firstLetterToLower $method.Name}}({{template "methodInputs" dict "Method" $method "TypeMap" $typeMap}}): Promise<{{if $method.StreamOutput}}void{{else}}{{$method.Name}}Return{{end}}>
+  {{firstLetterToLower $method.Name}}({{template "methodInputs" dict "Method" $method "TypeMap" $typeMap}}): {{if $method.StreamOutput}}WebrpcStreamController{{else}}Promise<{{$method.Name}}Return>{{end}}
 {{- end}}
 }