From 4b7256381c3b2ff2d4d5190e878de6903cd0d53a Mon Sep 17 00:00:00 2001
From: Justin Schuldt <justinschuldt@gmail.com>
Date: Sun, 13 Jun 2021 10:38:49 -0500
Subject: [PATCH] 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>
---
 integration/grpc-web/example.ts | 3 ++-
 src/generate-grpc-web.ts        | 3 ++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/integration/grpc-web/example.ts b/integration/grpc-web/example.ts
index c08474516..2c33dc068 100644
--- 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
index a9ae0ba8c..4909f2fac 100644
--- 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}());