fix connection leakage

sodahealth · Jan 27, 2024 · 14939d0 · 14939d0
1 parent bde063a
commit 14939d0
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 0 deletions.
diff --git a/internal/impl/nats/input.go b/internal/impl/nats/input.go
@@ -251,6 +251,9 @@ func (n *natsReader) Read(ctx context.Context) (*service.Message, service.AckFun
 }
 
 func (n *natsReader) Close(ctx context.Context) (err error) {
+	go func() {
+		n.disconnect()
+	}()
 	n.interruptOnce.Do(func() {
 		close(n.interruptChan)
 	})

diff --git a/internal/impl/nats/input_stream.go b/internal/impl/nats/input_stream.go
@@ -361,6 +361,9 @@ func (n *natsStreamReader) Read(ctx context.Context) (*service.Message, service.
 }
 
 func (n *natsStreamReader) Close(ctx context.Context) (err error) {
+	go func() {
+		n.disconnect()
+	}()
 	n.interruptOnce.Do(func() {
 		close(n.interruptChan)
 	})