[pubsub] Remove cancellation token / Avoid silent stop (#394)

Author: yoshidan

pubsub/Cargo.toml

@@ -20,22 +20,21 @@ tokio = "1.32"
 async-channel = "1.9"
 async-stream = "0.3"
 thiserror = "1.0"
-tokio-util = "0.7"
 
 token-source = "1.0"
 google-cloud-gax = { package = "gcloud-gax", version = "1.3.0", path = "../foundation/gax" }
 google-cloud-googleapis = { package = "gcloud-googleapis", version = "1.3.0", path = "../googleapis", features = ["pubsub"]}
-
 google-cloud-auth = { package = "gcloud-auth", optional = true, version = "1.1.0", path="../foundation/auth", default-features=false }
 
 [dev-dependencies]
 tokio = { version="1.32", features=["rt-multi-thread"] }
 rand = "0.8.5"
-tracing-subscriber = "0.3"
+tracing-subscriber = { version="0.3", features=["env-filter"] }
 serial_test = "3.1"
 uuid = { version="1.4", features=["v4"] }
 ctor = "0.1.26"
 futures-util = "0.3"
+tokio-util = "0.7"
 
 [features]
 default = ["auth", "default-tls"]

pubsub/README.md

@@ -153,87 +153,30 @@ async fn run(config: ClientConfig) -> Result<(), Status> {
 
     // Token for cancel.
     let cancel = CancellationToken::new();
-    let cancel2 = cancel.clone();
+    let cancel_for_task = cancel.clone();
     tokio::spawn(async move {
         // Cancel after 10 seconds.
         tokio::time::sleep(Duration::from_secs(10)).await;
-        cancel2.cancel();
+        cancel_for_task.cancel();
     });
 
-    // Receive blocks until the ctx is cancelled or an error occurs.
-    // Or simply use the `subscription.subscribe` method.
-    subscription.receive(|mut message, cancel| async move {
-        // Handle data.
-        println!("Got Message: {:?}", message.message.data);
-
-        // Ack or Nack message.
+    // Start receiving messages from the subscription.
+    let mut iter = subscription.subscribe(None).await?;
+    // Get buffered messages.
+    // To close safely, use a CancellationToken or to signal shutdown.
+    while let Some(message) = tokio::select!{
+        v = iter.next() => v,
+        _ = ctx.cancelled() => None,
+    }.await {
         let _ = message.ack().await;
-    }, cancel.clone(), None).await?;
-
+    }
+    // Wait for all the unprocessed messages to be Nack.
+    // If you don't call dispose, the unprocessed messages will be Nack when the iterator is dropped.
+    iter.dispose().await;    
+        
     // Delete subscription if needed.
     subscription.delete(None).await?;
 
     Ok(())
 }
-```
-
-### Subscribe Message (Alternative Way)
-
-After canceling, wait until all pulled messages are processed.
-```rust
-use std::time::Duration;
-use futures_util::StreamExt;
-use google_cloud_pubsub::client::{Client, ClientConfig};
-use google_cloud_googleapis::pubsub::v1::PubsubMessage;
-use google_cloud_pubsub::subscription::{SubscribeConfig, SubscriptionConfig};
-use google_cloud_gax::grpc::Status;
-
-async fn run(config: ClientConfig) -> Result<(), Status> {
-     // Creating Client, Topic and Subscription...
-     let client = Client::new(config).await.unwrap();
-     let subscription = client.subscription("test-subscription");
-
-     // Read the messages as a stream
-     let mut stream = subscription.subscribe(None).await.unwrap();
-     let cancellable = stream.cancellable();
-     let task = tokio::spawn(async move {
-         // None if the stream is cancelled
-         while let Some(message) = stream.next().await {
-             message.ack().await.unwrap();
-         }
-     });
-     tokio::time::sleep(Duration::from_secs(60)).await;
-     cancellable.cancel();
-     let _ = task.await;
-     Ok(())
-}
- ```
-
-Unprocessed messages are nack after cancellation.
-```rust
-use std::time::Duration;
-use google_cloud_pubsub::client::{Client, ClientConfig};
-use google_cloud_googleapis::pubsub::v1::PubsubMessage;
-use google_cloud_pubsub::subscription::{SubscribeConfig, SubscriptionConfig};
-use google_cloud_gax::grpc::Status;
-
-async fn run(config: ClientConfig) -> Result<(), Status> {
-     // Creating Client, Topic and Subscription...
-     let client = Client::new(config).await.unwrap();
-     let subscription = client.subscription("test-subscription");
-
-     // Read the messages as a stream
-     let mut stream = subscription.subscribe(None).await.unwrap();
-     let cancellable = stream.cancellable();
-     let task = tokio::spawn(async move {
-         // None if the stream is cancelled
-         while let Some(message) = stream.read().await {
-             message.ack().await.unwrap();
-         }
-     });
-     tokio::time::sleep(Duration::from_secs(60)).await;
-     cancellable.cancel();
-     let _ = task.await;
-     Ok(())
-}
 ```