fix: Dispose the pull stream.

googleapis · May 24, 2023 · 3518402 · 3518402
1 parent b46fe31
commit 3518402
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Tests/SubscriberClientTest.cs b/apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Tests/SubscriberClientTest.cs
@@ -223,7 +223,8 @@ private class PullStream : StreamingPullStream
                     _scheduler = scheduler;
                     _writeAsyncPreDelay = writeAsyncPreDelay; // delay within the WriteAsync() method. Simulating network or server slowness.
                     var responseStream = new En(msgs, scheduler, taskHelper, clock, useMsgAsId, ct, isExactlyOnceDelivery);
-                    _call = new AsyncDuplexStreamingCall<StreamingPullRequest, StreamingPullResponse>(null, responseStream, Task.FromResult(new Metadata()), null, null, null);
+                    // Set disposeAction parameter of AsyncDuplexStreamingCall to No-op as it is called internally while disposing stream.
+                    _call = new AsyncDuplexStreamingCall<StreamingPullRequest, StreamingPullResponse>(null, responseStream, Task.FromResult(new Metadata()), null, null, () => { });
                     _clock = clock;
                     _writeCompletes = writeCompletes;
                     _streamPings = streamPings;

diff --git a/apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.SingleChannel.cs b/apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.SingleChannel.cs
@@ -276,10 +276,19 @@ private void StopStreamingPull()
                 // Ignore all errors; the stream may be in any state.
                 try
                 {
-                    _registerTaskFn(_pull.WriteCompleteAsync());
+                    Add(_pull.WriteCompleteAsync(), Next(false, () =>
+                        {
+                            try
+                            {
+                                _pull.Dispose();
+                            }
+                            finally
+                            {
+                                _pull = null;
+                            }
+                        }));
                 }
                 catch { }
-                _pull = null;
             }
         }