Add double_ack_with method to JetStream Messages

This will allow consumers of the library to obtain exactly-once
delivery semantics across a wider range of valid usecases that include
NAKing entire ranges of Messages.

Author: OtaK

async-nats/src/jetstream/message.rs

@@ -338,7 +338,7 @@ impl Message {
         }
     }
 
-    /// Acknowledges a message delivery by sending `+ACK` to the server
+    /// Acknowledges a message delivery by sending a chosen [AckKind] to the server
     /// and awaits for confirmation for the server that it received the message.
     /// Useful if user wants to ensure `exactly once` semantics.
     ///
@@ -350,6 +350,7 @@ impl Message {
     /// ```no_run
     /// # #[tokio::main]
     /// # async fn main() -> Result<(), async_nats::Error> {
+    /// use async_nats::jetstream::AckKind;
     /// use futures_util::StreamExt;
     /// let client = async_nats::connect("localhost:4222").await?;
     /// let jetstream = async_nats::jetstream::new(client);
@@ -363,18 +364,18 @@ impl Message {
     /// let mut messages = consumer.fetch().max_messages(100).messages().await?;
     ///
     /// while let Some(message) = messages.next().await {
-    ///     message?.double_ack().await?;
+    ///     message?.double_ack_with(AckKind::Ack).await?;
     /// }
     /// # Ok(())
     /// # }
     /// ```
-    pub async fn double_ack(&self) -> Result<(), Error> {
+    pub async fn double_ack_with(&self, ack_kind: AckKind) -> Result<(), Error> {
         if let Some(ref reply) = self.reply {
             let inbox = self.context.client.new_inbox();
             let mut subscription = self.context.client.subscribe(inbox.clone()).await?;
             self.context
                 .client
-                .publish_with_reply(reply.clone(), inbox, AckKind::Ack.into())
+                .publish_with_reply(reply.clone(), inbox, ack_kind.into())
                 .await?;
             match tokio::time::timeout(self.context.timeout, subscription.next())
                 .await
@@ -394,6 +395,40 @@ impl Message {
         }
     }
 
+    /// Acknowledges a message delivery by sending `+ACK` to the server
+    /// and awaits for confirmation for the server that it received the message.
+    /// Useful if user wants to ensure `exactly once` semantics.
+    ///
+    /// If [AckPolicy][crate::jetstream::consumer::AckPolicy] is set to `All` or `Explicit`, messages has to be acked.
+    /// Otherwise redeliveries will occur and [Consumer][crate::jetstream::consumer::Consumer] will not be able to advance.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), async_nats::Error> {
+    /// use futures_util::StreamExt;
+    /// let client = async_nats::connect("localhost:4222").await?;
+    /// let jetstream = async_nats::jetstream::new(client);
+    ///
+    /// let consumer = jetstream
+    ///     .get_stream("events")
+    ///     .await?
+    ///     .get_consumer("pull")
+    ///     .await?;
+    ///
+    /// let mut messages = consumer.fetch().max_messages(100).messages().await?;
+    ///
+    /// while let Some(message) = messages.next().await {
+    ///     message?.double_ack().await?;
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn double_ack(&self) -> Result<(), Error> {
+        self.double_ack_with(AckKind::Ack).await
+    }
+
     /// Returns the `JetStream` message ID
     /// if this is a `JetStream` message.
     #[allow(clippy::mixed_read_write_in_expression)]
@@ -535,8 +570,7 @@ impl Acker {
     /// ```no_run
     /// # #[tokio::main]
     /// # async fn main() -> Result<(), async_nats::Error> {
-    /// use async_nats::jetstream::consumer::PullConsumer;
-    /// use async_nats::jetstream::Message;
+    /// use async_nats::jetstream::{consumer::PullConsumer, Message};
     /// use futures_util::StreamExt;
     /// let client = async_nats::connect("localhost:4222").await?;
     /// let jetstream = async_nats::jetstream::new(client);
@@ -581,9 +615,7 @@ impl Acker {
     /// ```no_run
     /// # #[tokio::main]
     /// # async fn main() -> Result<(), async_nats::Error> {
-    /// use async_nats::jetstream::consumer::PullConsumer;
-    /// use async_nats::jetstream::AckKind;
-    /// use async_nats::jetstream::Message;
+    /// use async_nats::jetstream::{consumer::PullConsumer, AckKind, Message};
     /// use futures_util::StreamExt;
     /// let client = async_nats::connect("localhost:4222").await?;
     /// let jetstream = async_nats::jetstream::new(client);
@@ -621,7 +653,7 @@ impl Acker {
         }
     }
 
-    /// Acknowledges a message delivery by sending `+ACK` to the server
+    /// Acknowledges a message delivery by sending the chosen [AckKind] to the server
     /// and awaits for confirmation for the server that it received the message.
     /// Useful if user wants to ensure `exactly once` semantics.
     ///
@@ -633,7 +665,7 @@ impl Acker {
     /// ```no_run
     /// # #[tokio::main]
     /// # async fn main() -> Result<(), async_nats::Error> {
-    /// use async_nats::jetstream::Message;
+    /// use async_nats::jetstream::{AckKind, Message};
     /// use futures_util::StreamExt;
     /// let client = async_nats::connect("localhost:4222").await?;
     /// let jetstream = async_nats::jetstream::new(client);
@@ -652,18 +684,18 @@ impl Acker {
     ///     // while retaining ability to ack later.
     ///     println!("message: {:?}", message);
     ///     // Ack it. `Message` may be dropped already.
-    ///     acker.double_ack().await?;
+    ///     acker.double_ack_with(AckKind::Ack).await?;
     /// }
     /// # Ok(())
     /// # }
     /// ```
-    pub async fn double_ack(&self) -> Result<(), Error> {
+    pub async fn double_ack_with(&self, ack_kind: AckKind) -> Result<(), Error> {
         if let Some(ref reply) = self.reply {
             let inbox = self.context.client.new_inbox();
             let mut subscription = self.context.client.subscribe(inbox.to_owned()).await?;
             self.context
                 .client
-                .publish_with_reply(reply.to_owned(), inbox, AckKind::Ack.into())
+                .publish_with_reply(reply.to_owned(), inbox, ack_kind.into())
                 .await?;
             match tokio::time::timeout(self.context.timeout, subscription.next())
                 .await
@@ -682,6 +714,46 @@ impl Acker {
             )))
         }
     }
+
+    /// Acknowledges a message delivery by sending `+ACK` to the server
+    /// and awaits for confirmation for the server that it received the message.
+    /// Useful if user wants to ensure `exactly once` semantics.
+    ///
+    /// If [AckPolicy][crate::jetstream::consumer::AckPolicy] is set to `All` or `Explicit`, messages has to be acked.
+    /// Otherwise redeliveries will occur and [Consumer][crate::jetstream::consumer::Consumer] will not be able to advance.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), async_nats::Error> {
+    /// use async_nats::jetstream::Message;
+    /// use futures_util::StreamExt;
+    /// let client = async_nats::connect("localhost:4222").await?;
+    /// let jetstream = async_nats::jetstream::new(client);
+    ///
+    /// let consumer = jetstream
+    ///     .get_stream("events")
+    ///     .await?
+    ///     .get_consumer("pull")
+    ///     .await?;
+    ///
+    /// let mut messages = consumer.fetch().max_messages(100).messages().await?;
+    ///
+    /// while let Some(message) = messages.next().await {
+    ///     let (message, acker) = message.map(Message::split)?;
+    ///     // Do something with the message. Ownership can be taken over `Message`.
+    ///     // while retaining ability to ack later.
+    ///     println!("message: {:?}", message);
+    ///     // Ack it. `Message` may be dropped already.
+    ///     acker.double_ack().await?;
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn double_ack(&self) -> Result<(), Error> {
+        self.double_ack_with(AckKind::Ack).await
+    }
 }
 /// The kinds of response used for acknowledging a processed message.
 #[derive(Debug, Clone, Copy)]

async-nats/tests/jetstream_tests.rs

@@ -2457,9 +2457,9 @@ mod jetstream {
             .unwrap();
         let mut consumer = stream.get_consumer("pull").await.unwrap();
 
-        for _ in 0..10 {
+        for i in 0..10 {
             context
-                .publish("events", "dat".into())
+                .publish("events", format!("dat-{i}").into())
                 .await
                 .unwrap()
                 .await
@@ -2523,10 +2523,11 @@ mod jetstream {
         if let Some(message) = iter.next().await {
             message
                 .unwrap()
-                .ack_with(async_nats::jetstream::AckKind::Nak(None))
+                .double_ack_with(async_nats::jetstream::AckKind::Nak(None))
                 .await
                 .unwrap();
         }
+
         client.flush().await.unwrap();
 
         tokio::time::sleep(Duration::from_millis(100)).await;