fix: close server stream on observer unsubscribe (#309)

* close server stream on unsubscribe * Fix lint error, use a lambda just in case. Co-authored-by: Stephen Haberman <stephen.haberman@gmail.com>
stephenh · Jun 13, 2021 · 4b72563 · 4b72563
1 parent 5cd3ef9
commit 4b72563
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/integration/grpc-web/example.ts b/integration/grpc-web/example.ts
@@ -992,7 +992,7 @@ export class GrpcWebImpl {
         : metadata || this.options.metadata;
     return new Observable((observer) => {
       const upStream = () => {
-        grpc.invoke(methodDesc, {
+        const client = grpc.invoke(methodDesc, {
           host: this.host,
           request,
           transport: this.options.streamingTransport || this.options.transport,
@@ -1009,6 +1009,7 @@ export class GrpcWebImpl {
             }
           },
         });
+        observer.add(() => client.close());
       };
       upStream();
     }).pipe(share());

diff --git a/src/generate-grpc-web.ts b/src/generate-grpc-web.ts
@@ -300,7 +300,7 @@ function createInvokeMethod() {
         : metadata || this.options.metadata;
       return new Observable(observer => {
         const upStream = (() => {
-          ${grpc}.invoke(methodDesc, {
+          const client = ${grpc}.invoke(methodDesc, {
             host: this.host,
             request,
             transport: this.options.streamingTransport || this.options.transport,
@@ -317,6 +317,7 @@ function createInvokeMethod() {
               }
             },
           });
+          observer.add(() => client.close());
         });
         upStream();
       }).pipe(${share}());