feat(pubsub): add APIs to shutdown subscriber (#5257)

Most of the work for #5024 (the rest are cleanups)

Provide a way for applications to both signal and await a shutdown of a
subscriber message stream.

---

Docs look like:

<img width="1277" height="557" alt="Screenshot From 2026-04-03 12-52-40"
src="https://github.com/user-attachments/assets/e018716d-270e-4864-a02c-36f07bef2151"
/>

<img width="1287" height="965" alt="Screenshot From 2026-04-03 12-54-36"
src="https://github.com/user-attachments/assets/4711d483-b2e1-46d6-83e6-8be4ce34a61c"
/>

Author: dbolduc

src/pubsub/src/subscriber.rs

@@ -16,6 +16,7 @@ pub mod handler;
 
 pub use message_stream::MessageStream;
 pub use shutdown_behavior::ShutdownBehavior;
+pub use shutdown_token::ShutdownToken;
 
 pub(super) mod builder;
 pub(super) mod client;
@@ -27,7 +28,6 @@ mod leaser;
 mod message_stream;
 mod retry_policy;
 mod shutdown_behavior;
-#[allow(dead_code)] // TODO(#5024) - implementation in progress...
 mod shutdown_token;
 mod stream;
 mod stub;

src/pubsub/src/subscriber/message_stream.rs

@@ -18,7 +18,6 @@ use super::lease_loop::LeaseLoop;
 use super::lease_state::{AtLeastOnceInfo, ExactlyOnceInfo, LeaseInfo, LeaseOptions, NewMessage};
 use super::leaser::DefaultLeaser;
 use super::retry_policy::StreamRetryPolicy;
-#[cfg(test)]
 use super::shutdown_token::ShutdownToken;
 use super::stream::Stream;
 use super::stub::TonicStreaming as _;
@@ -65,7 +64,6 @@ pub struct MessageStream {
     /// the same time.
     inner: MessageStreamImpl,
 
-    #[allow(dead_code)] // TODO(#5024) - implementation in progress...
     /// This future is ready when the lease loop shutdown completes.
     lease_loop: Shared<BoxFuture<'static, ()>>,
 
@@ -254,7 +252,28 @@ impl MessageStream {
         }))
     }
 
-    #[cfg(test)] // TODO(#5024) - document and make public
+    /// Returns a shutdown token for the stream.
+    ///
+    /// # Example
+    /// ```
+    /// # use google_cloud_pubsub::subscriber::MessageStream;
+    /// # async fn sample(mut stream: MessageStream) {
+    /// // Get a shutdown token for the stream.
+    /// let shutdown_token = stream.shutdown_token();
+    ///
+    /// // Signal and await a shutdown of the stream.
+    /// shutdown_token.shutdown().await;
+    ///
+    /// // The stream stops yielding messages after a cancel.
+    /// assert!(stream.next().await.is_none());
+    /// # }
+    /// ```
+    ///
+    /// Use this token to signal and/or await shutdown of the stream.
+    ///
+    /// Awaiting a stream shutdown gives the subscriber time to flush its
+    /// pending acknowledgements, and schedule other messages for redelivery to
+    /// another client as soon as possible.
     pub fn shutdown_token(&self) -> ShutdownToken {
         ShutdownToken {
             inner: self.shutdown.clone(),

src/pubsub/src/subscriber/shutdown_token.rs

@@ -18,7 +18,7 @@ use tokio_util::sync::CancellationToken;
 /// A token to signal and await shutdown of a stream.
 ///
 /// # Example
-/// ```no_rust
+/// ```
 /// use google_cloud_pubsub::subscriber::MessageStream;
 /// async fn sample(stream: MessageStream) {
 ///   // Get a shutdown token for the stream.