impl(pubsub): add shutdown token (#5125)

Part of the work for #5024 

Add a type that can signal and await shutdown.

I think applications will typically signal and await together, but we
separate the functions to allow for graceful shutdown without signaling
shutdown (e.g. from an external error):

```rs
let shutdown_token = stream.shutdown_token();
while let Some((m, h)) = stream.next().await { ... }

// There must have been an external error. Wait for shutdown before returning.
shutdown_token.shutdown().await;
```

Author: dbolduc

src/pubsub/src/subscriber.rs

@@ -27,6 +27,8 @@ 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;
 mod transport;

src/pubsub/src/subscriber/shutdown_token.rs

@@ -0,0 +1,120 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use futures::future::{BoxFuture, Shared};
+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.
+///   let token = stream.shutdown_token();
+///
+///   // Signal a shutdown of the stream.
+///   token.cancel();
+///
+///   // Await a shutdown of the stream.
+///   token.shutdown().await;
+/// }
+/// ```
+#[derive(Clone, Debug)]
+pub struct ShutdownToken {
+    pub(super) inner: CancellationToken,
+    pub(super) fut: Shared<BoxFuture<'static, ()>>,
+}
+
+impl ShutdownToken {
+    /// Signal a stream shutdown.
+    ///
+    /// The stream will stop yielding messages.
+    pub fn cancel(&self) {
+        self.inner.cancel();
+    }
+
+    /// Await a stream shutdown.
+    ///
+    /// Applications should call this to ensure all pending ack/nack RPCs have
+    /// time to complete before a process exits.
+    ///
+    /// See [`Subscribe::set_shutdown_behavior`][setter] to configure the exact
+    /// behavior on shutdown.
+    ///
+    /// [setter]: crate::builder::subscriber::Subscribe::set_shutdown_behavior
+    pub async fn shutdown(&self) {
+        self.fut.clone().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::FutureExt;
+    use tokio::sync::oneshot::channel;
+
+    #[tokio::test(start_paused = true)]
+    async fn cancel() {
+        let token = ShutdownToken {
+            inner: CancellationToken::new(),
+            fut: std::future::pending().boxed().shared(),
+        };
+        assert!(!token.inner.is_cancelled(), "{token:?}");
+
+        let token_clone = token.clone();
+        assert!(!token_clone.inner.is_cancelled(), "{token_clone:?}");
+
+        token.cancel();
+        assert!(token.inner.is_cancelled(), "{token:?}");
+        assert!(token_clone.inner.is_cancelled(), "{token_clone:?}");
+
+        // A second cancel is a no-op.
+        token.cancel();
+        assert!(token.inner.is_cancelled(), "{token:?}");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn shutdown() -> anyhow::Result<()> {
+        let (tx, rx) = channel();
+        let fut = rx.map(|_| ()).boxed().shared();
+
+        let token = ShutdownToken {
+            inner: CancellationToken::new(),
+            fut,
+        };
+        assert!(token.fut.peek().is_none(), "future should be pending");
+
+        let token_clone = token.clone();
+        assert!(token_clone.fut.peek().is_none(), "future should be pending");
+
+        let handle = tokio::spawn(async move {
+            token_clone.shutdown().await;
+        });
+        tokio::task::yield_now().await;
+
+        assert!(token.fut.peek().is_none(), "future should be pending");
+
+        // Satisfy the future
+        let _ = tx.send(());
+        handle.await?;
+        assert!(token.fut.peek().is_some(), "future should be satisfied");
+
+        // A second shutdown is a no-op.
+        token.shutdown().await;
+        assert!(token.fut.peek().is_some(), "future should be satisfied");
+
+        Ok(())
+    }
+}