cleanup(pubsub)!: remove explicit nack (#4549)

Simplify the pubsub subscriber API by removing the explicit `nack`
method and replacing it with an idiomatic `Drop` implementation. Update
unit tests and documentation accordingly.

Fixes #4514

---------

Co-authored-by: Carlos O'Ryan <coryan@google.com>

Author: haphungw

src/pubsub/src/subscriber/builder.rs

@@ -64,8 +64,9 @@ impl StreamingPull {
 
     /// Sets the ack deadline to use for the stream.
     ///
-    /// This value represents how long the application has to ack or nack an
-    /// incoming message. Note that this value is independent of the deadline
+    /// This value represents how long the application has to ack an
+    /// incoming message. If the handler is dropped without being acked,
+    /// it is nacked. Note that this value is independent of the
     /// configured on the server-side subscription.
     ///
     /// If the server does not hear back from the client within this deadline
@@ -95,7 +96,7 @@ impl StreamingPull {
     /// Flow control settings for the maximum number of outstanding messages.
     ///
     /// The server will stop sending messages to a client when this many
-    /// messages are outstanding (i.e. that have not been acked or nacked).
+    /// messages are outstanding (i.e. that have not been acked).
     ///
     /// The server resumes sending messages when the outstanding message count
     /// drops below this value.
@@ -122,7 +123,7 @@ impl StreamingPull {
     /// Flow control settings for the maximum number of outstanding bytes.
     ///
     /// The server will stop sending messages to a client when this many bytes
-    /// of messages are outstanding (i.e. that have not been acked or nacked).
+    /// of messages are outstanding (i.e. that have not been acked).
     ///
     /// The server resumes sending messages when the outstanding byte count
     /// drops below this value.

src/pubsub/src/subscriber/handler.rs

@@ -40,16 +40,6 @@ impl Handler {
             Handler::AtLeastOnce(h) => h.ack(),
         }
     }
-
-    /// Rejects the message associated with this handler.
-    ///
-    /// The message will be removed from this `Subscriber`'s lease management.
-    /// The service will redeliver this message, possibly to another client.
-    pub fn nack(self) {
-        match self {
-            Handler::AtLeastOnce(h) => h.nack(),
-        }
-    }
 }
 
 #[derive(Debug)]
@@ -91,16 +81,6 @@ impl AtLeastOnce {
         }
     }
 
-    /// Rejects the message associated with this handler.
-    ///
-    /// The message will be removed from this `Subscriber`'s lease management.
-    /// The service will redeliver this message, possibly to another client.
-    pub fn nack(mut self) {
-        if let Some(inner) = self.inner.take() {
-            inner.nack();
-        }
-    }
-
     #[cfg(test)]
     pub(crate) fn ack_id(&self) -> &str {
         self.inner
@@ -148,7 +128,7 @@ mod tests {
         let h = Handler::AtLeastOnce(AtLeastOnce::new(test_id(1), ack_tx));
         assert_eq!(ack_rx.try_recv(), Err(TryRecvError::Empty));
 
-        h.nack();
+        drop(h);
         let ack = ack_rx.try_recv()?;
         assert_eq!(ack, AckResult::Nack(test_id(1)));
 
@@ -174,7 +154,7 @@ mod tests {
         let h = AtLeastOnce::new(test_id(1), ack_tx);
         assert_eq!(ack_rx.try_recv(), Err(TryRecvError::Empty));
 
-        h.nack();
+        drop(h);
         let ack = ack_rx.try_recv()?;
         assert_eq!(ack, AckResult::Nack(test_id(1)));
 

src/pubsub/src/subscriber/session.rs

@@ -132,8 +132,8 @@ impl Session {
 
     /// Returns the next message received on this subscription.
     ///
-    /// The message data is returned along with a [Handler] for acknowledging
-    /// (ack) or rejecting (nack) the message.
+    /// Returns the message data along with a [Handler] for acknowledging (ack) the message.
+    /// Dropping the [Handler] without acknowledging it will reject (nack) the message.
     ///
     /// If the underlying stream encounters a permanent error, an `Error` is
     /// returned instead.
@@ -469,7 +469,7 @@ mod tests {
             let Some((_, Handler::AtLeastOnce(h))) = session.next().await.transpose()? else {
                 anyhow::bail!("expected message {i}")
             };
-            h.nack();
+            drop(h);
         }
         // Take a long time to process some messages
         let mut hold = Vec::new();