chore(PubSub): Deprecate legacy StopAsync and update docs

Author: robertvoinescu-work

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.IntegrationTests/PubSubClientTest.cs

@@ -159,7 +159,7 @@ private async Task RunBulkMessagingImpl(
                     {
                         // Test finished, so stop subscriber
                         Console.WriteLine("All msgs received, stopping subscriber.");
-                        Task unused = subscriber.StopAsync(TimeSpan.FromSeconds(15));
+                        Task unused = subscriber.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, TimeSpan.FromSeconds(15));
                     }
                 }
                 else
@@ -194,11 +194,11 @@ private async Task RunBulkMessagingImpl(
                     {
                         if (noProgressCount > 60)
                         {
-                            // Deadlock, shutdown subscriber, and cancel
-                            Console.WriteLine("Deadlock detected. Cancelling test");
-                            subscriber.StopAsync(new CancellationToken(true));
-                            watchdogCts.Cancel();
-                            break;
+                             // Deadlock, shutdown subscriber, and cancel
+                             Console.WriteLine("Deadlock detected. Cancelling test");
+                             subscriber.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, cancellationToken: new CancellationToken(true));
+                             watchdogCts.Cancel();
+                             break;
                         }
                         noProgressCount += 1;
                     }
@@ -431,7 +431,7 @@ public async Task StopStartSubscriber(int totalMessageCount, double publisherFre
                     });
                     await Task.Delay(subscriberLifetime);
                     Console.WriteLine("Stopping subscriber");
-                    Task stopTask = subscriber.StopAsync(TimeSpan.FromSeconds(15));
+                    Task stopTask = subscriber.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, TimeSpan.FromSeconds(15));
                     // If shutdown times-out then stopTask, and also Task.WhenAll will cancel, causing the test to fail.
                     await Task.WhenAll(subscribeTask, stopTask);
                     int recvCount = recvedMsgs.Locked(() => recvedMsgs.Count);
@@ -538,8 +538,8 @@ await subscriberApi.IAMPolicyClient.SetIamPolicyAsync(new SetIamPolicyRequest
             {
                 result.Add((msg.GetDeliveryAttempt(), true));
                 // Received DLQ message, so stop test.
-                sub.StopAsync(TimeSpan.FromSeconds(10));
-                dlqSub.StopAsync(TimeSpan.FromSeconds(10));
+                sub.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, TimeSpan.FromSeconds(10));
+                dlqSub.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, TimeSpan.FromSeconds(10));
                 return Task.FromResult(SubscriberClient.Reply.Ack);
             });
 

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.OrderingKeyTester/Program.cs

@@ -133,7 +133,7 @@ Task Subscribe()
                         if (recvCount == inputLines.Count)
                         {
                             Console.WriteLine("Received all messages, shutting down");
-                            var dummyTask = sub.StopAsync(CancellationToken.None);
+                            var dummyTask = sub.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately);
                         }
                     }
                     if (rnd.Next(3) == 0)

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Snippets/SubscriberClientSnippets.cs

@@ -143,7 +143,7 @@ await _subscriberClient.StartAsync((msg, token) =>
                 });
 
         public override async Task StopAsync(CancellationToken stoppingToken) =>
-            await _subscriberClient.StopAsync(stoppingToken);
+            await _subscriberClient.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, cancellationToken: stoppingToken);
     }
     // End sample
 }

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Snippets/SubscriberServiceApiClientSnippets.cs

@@ -129,7 +129,7 @@ await subscriber.StartAsync((msg, cancellationToken) =>
                 Console.WriteLine($"Text: '{msg.Data.ToStringUtf8()}'");
                 // Stop this subscriber after one message is received.
                 // This is non-blocking, and the returned Task may be awaited.
-                subscriber.StopAsync(TimeSpan.FromSeconds(15));
+                subscriber.StopAsync(SubscriberClient.SubscriberShutdownSetting.NackImmediately, TimeSpan.FromSeconds(15));
                 // Return Reply.Ack to indicate this message has been handled.
                 return Task.FromResult(SubscriberClient.Reply.Ack);
             });

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1.Tests/SubscriberClientTest.cs

@@ -596,7 +596,9 @@ public void ImmediateStop_Obsolete(
                     });
                     await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromSeconds(1), CancellationToken.None));
                     var isCancelled = await fake.TaskHelper.ConfigureAwaitHideCancellation(
+#pragma warning disable CS0618 // allow use of obsolete method
                         () => fake.Subscriber.StopAsync(new CancellationToken(hardStop)));
+#pragma warning restore CS0618
                     Assert.Equal(hardStop, isCancelled);
                     Assert.Equal(1, fake.Subscribers.Count);
                     Assert.Empty(fake.Subscribers[0].Acks);
@@ -645,7 +647,9 @@ public void StopBeforeStart_Obsolete()
                     await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromSeconds(1), CancellationToken.None));
                     var exception = await Assert.ThrowsAsync<InvalidOperationException>(
                         async () => await fake.TaskHelper.ConfigureAwaitHideCancellation(
+#pragma warning disable CS0618 // allow use of obsolete method
                             () => fake.Subscriber.StopAsync(TimeSpan.FromHours(1))));
+#pragma warning restore CS0618
                     Assert.Equal("Can only stop a started instance.", exception.Message);
                     Assert.Equal(1, fake.Subscribers.Count);
                     Assert.Empty(fake.Subscribers[0].Acks);
@@ -705,7 +709,9 @@ await fake.TaskHelper.ConfigureAwaitHideCancellation(
                        () => fake.Subscriber.DisposeAsync().AsTask());
                     // Call StopAsync. It shouldn't throw an exception.
                     await fake.TaskHelper.ConfigureAwaitHideCancellation(
+#pragma warning disable CS0618 // allow use of obsolete method
                        () => fake.Subscriber.StopAsync(CancellationToken.None));
+#pragma warning restore CS0618
 
                     Assert.Equal(1, fake.Subscribers.Count);
                     Assert.Empty(fake.Subscribers[0].Acks);
@@ -812,8 +818,10 @@ public void RecvManyMsgsNoErrors_Obsolete(
                         return SubscriberClient.Reply.Ack;
                     });
                     await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromSeconds(stopAfterSeconds) + TimeSpan.FromSeconds(0.5), CancellationToken.None));
+#pragma warning disable CS0618 // allow use of obsolete method
                     var isCancelled = await fake.TaskHelper.ConfigureAwaitHideCancellation(
                         () => fake.Subscriber.StopAsync(new CancellationToken(hardStop)));
+#pragma warning restore CS0618
                     Assert.Equal(hardStop, isCancelled);
                     Assert.Equal(clientCount, fake.Subscribers.Count);
                     Assert.Equal(expectedMsgCount, handledMsgs.Locked(() => handledMsgs.Count));
@@ -986,7 +994,9 @@ public void FlowControl_Obsolete(
                         return SubscriberClient.Reply.Ack;
                     });
                     await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromSeconds(stopAfterSeconds) + TimeSpan.FromSeconds(0.5), CancellationToken.None));
+#pragma warning disable CS0618 // allow use of obsolete method
                     await fake.TaskHelper.ConfigureAwaitHideCancellation(() => fake.Subscriber.StopAsync(new CancellationToken(hardStop)));
+#pragma warning restore CS0618
                     Assert.Equal(expectedMsgCount, handledMsgs.Count);
                 });
             }
@@ -2275,7 +2285,9 @@ public void Shutdown_SoftStop_NacksMessages()
 
                 await fake.TaskHelper.ConfigureAwait(fake.Scheduler.Delay(TimeSpan.FromSeconds(1), CancellationToken.None));
                 Assert.Equivalent(new[] { "msg0", "msg1", "msg2", "msg3", "msg4" }, givenToMessageHandler, strict: true);
+#pragma warning disable CS0618 // allow use of obsolete method
                 await fake.TaskHelper.ConfigureAwait(fake.Subscriber.StopAsync(CancellationToken.None));
+#pragma warning restore CS0618
 
                 // msg0-2 were handled/acked; msg3-5 were pulled but nacked on shutdown.
                 Assert.Equivalent(new[] { "msg0", "msg1", "msg2", "msg3", "msg4" }, fake.Subscribers[0].Acks.Select(x => x.Id), strict: true);

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClient.cs

@@ -235,6 +235,7 @@ public virtual Task StartAsync(Func<PubsubMessage, CancellationToken, Task<Reply
     /// <param name="hardStopToken">Cancel this <see cref="CancellationToken"/> to abort handlers and acknowledgement.</param>
     /// <returns>A <see cref="Task"/> that completes when all handled messages have been acknowledged;
     /// faults on unrecoverable service errors; or cancels if <paramref name="hardStopToken"/> is cancelled.</returns>
+    [Obsolete("Use StopAsync(SubscriberShutdownSetting, TimeSpan?, CancellationToken) instead.")]
     public virtual Task StopAsync(CancellationToken hardStopToken) => throw new NotImplementedException();
 
     /// <summary>
@@ -247,7 +248,8 @@ public virtual Task StartAsync(Func<PubsubMessage, CancellationToken, Task<Reply
     /// <param name="timeout">After this period, abort handling and acknowledging messages.</param>
     /// <returns>A <see cref="Task"/> that completes when all handled messages have been acknowledged;
     /// faults on unrecoverable service errors; or cancels if <paramref name="timeout"/> expires.</returns>
-    public virtual Task StopAsync(TimeSpan timeout) => StopAsync(new CancellationTokenSource(timeout).Token);
+    [Obsolete("Use StopAsync(SubscriberShutdownSetting, TimeSpan?, CancellationToken) instead.")]
+    public virtual Task StopAsync(TimeSpan timeout) => StopAsync(SubscriberShutdownSetting.WaitForProcessing, timeout);
 
     /// <summary>
     /// Stop this <see cref="SubscriberClient"/>.

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.cs

@@ -200,10 +200,13 @@ public override ValueTask DisposeAsync()
                 return new ValueTask(Task.CompletedTask);
             }
         }
-        return new ValueTask(StopAsync(_disposeTimeout));
+#pragma warning disable CS0618 // allow use of obsolete method
+        return new ValueTask(StopAsync( _disposeTimeout));
+#pragma warning restore CS0618
     }
 
     /// <inheritdoc />
+    [Obsolete("Use StopAsync(SubscriberShutdownSetting, TimeSpan?, CancellationToken) instead.")]
     public override Task StopAsync(CancellationToken hardStopToken)
     {
         lock (_lock)

apis/Google.Cloud.PubSub.V1/docs/index.md

@@ -172,6 +172,15 @@ Below is an example implementation of a console application that utilizes the de
 
 {{sample:SubscriberClient.UseSubscriberServiceInConsoleApp}}
 
+## Subscriber shutdown
+
+When shutting down a `SubscriberClient`, two different shutdown flows are available via the `StopAsync(SubscriberShutdownSetting, TimeSpan?, CancellationToken)` method:
+
+- **NackImmediately**: This immediately stops streaming new messages and then actively sends "Nack" (Negative Acknowledgement) responses for any messages that have been received but have not yet finished being handled. This allows those messages to be quickly redelivered to other active subscribers.
+- **WaitForProcessing**: This immediately stops streaming new messages from the server but continues to process all messages that have already been received. If processing does not complete 30s before the specified timeout, the client will switch to NackImmediately to release any remaining messages.
+
+By default, a 1-hour timeout is used for the shutdown process, which can be customized as needed. When the timeout is reached or if the `CancellationToken` is invoked this will trigger an immediate hard stop, aborting all outstanding tasks.
+
 ## Disposing of the publisher and subscriber clients
 
 Both `PublisherClient` and `SubscriberClient` implement the `IAsyncDisposable` interface,