Updated JetstreamMessage controls.

Author: s3rius

python/natsrpy/_natsrpy_rs/js/__init__.pyi

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from typing import Any
 
 from .managers import KVManager, StreamsManager
@@ -26,4 +27,57 @@ class JetStreamMessage:
     def payload(self) -> bytes: ...
     @property
     def headers(self) -> dict[str, Any]: ...
-    async def ack(self) -> None: ...
+    async def ack(self, double: bool = False) -> None:
+        """
+        Acknowledge that a message was handled.
+
+        :param double: whether to wait for server response, defaults to False
+        """
+
+    async def nack(
+        self,
+        delay: float | timedelta | None = None,
+        double: bool = False,
+    ) -> None:
+        """
+        Negative acknowledgement.
+
+        Signals that the message will not be processed now
+        and processing can move onto the next message, NAK'd
+        message will be retried.
+
+        :param duration: time, defaults to None
+        :param double: whether to wait for server response, defaults to False
+        """
+
+    async def progress(self, double: bool = False) -> None:
+        """
+        Progress acknowledgement.
+
+        Singnals that the mesasge is being handled right now.
+        Sending this request before the AckWait will extend wait period
+        before redelivering a message.
+
+        :param double: whether to wait for server response, defaults to False
+        """
+
+    async def next(self, double: bool = False) -> None:
+        """
+        Next acknowledgement.
+
+        Only applies to pull consumers!
+        Acknowledges message processing and instructs server to send
+        delivery of the next message to the reply subject.
+
+        :param double: whether to wait for server response, defaults to False
+        """
+
+    async def term(self, double: bool = False) -> None:
+        """
+        Term acknowledgement.
+
+        Instructs server to stop redelivering message.
+        Useful to stop redelivering a message after multiple NACKs.
+
+        :param double: whether to wait for server response, defaults to False
+        """

src/js/consumers/pull/consumer.rs

@@ -6,7 +6,7 @@ use tokio::sync::RwLock;
 
 use crate::{
     exceptions::rust_err::NatsrpyResult,
-    utils::{futures::natsrpy_future_with_timeout, py_types::TimeoutValue},
+    utils::{futures::natsrpy_future_with_timeout, py_types::TimeValue},
 };
 
 type NatsPullConsumer =
@@ -51,7 +51,7 @@ impl PullConsumer {
         expires: Option<Duration>,
         min_pending: Option<usize>,
         min_ack_pending: Option<usize>,
-        timeout: Option<TimeoutValue>,
+        timeout: Option<TimeValue>,
     ) -> NatsrpyResult<Bound<'py, PyAny>> {
         let ctx = self.consumer.clone();
 

src/js/message.rs

@@ -4,7 +4,7 @@ use tokio::sync::RwLock;
 
 use crate::{
     exceptions::rust_err::NatsrpyResult,
-    utils::{headers::NatsrpyHeadermapExt, natsrpy_future},
+    utils::{headers::NatsrpyHeadermapExt, natsrpy_future, py_types::TimeValue},
 };
 
 #[pyo3::pyclass]
@@ -25,6 +25,25 @@ impl From<async_nats::jetstream::Message> for JetStreamMessage {
     }
 }
 
+impl JetStreamMessage {
+    pub fn inner_ack<'py>(
+        &self,
+        py: Python<'py>,
+        kind: async_nats::jetstream::message::AckKind,
+        double: bool,
+    ) -> NatsrpyResult<Bound<'py, PyAny>> {
+        let acker_guard = self.acker.clone();
+        natsrpy_future(py, async move {
+            if double {
+                acker_guard.read().await.double_ack_with(kind).await?;
+            } else {
+                acker_guard.read().await.ack_with(kind).await?;
+            }
+            Ok(())
+        })
+    }
+}
+
 #[pyo3::pymethods]
 impl JetStreamMessage {
     #[getter]
@@ -51,11 +70,41 @@ impl JetStreamMessage {
         }
     }
 
-    pub fn ack<'py>(&self, py: Python<'py>) -> NatsrpyResult<Bound<'py, PyAny>> {
-        let acker_guard = self.acker.clone();
-        natsrpy_future(py, async move {
-            acker_guard.read().await.ack().await?;
-            Ok(())
-        })
+    #[pyo3(signature=(double=false))]
+    pub fn ack<'py>(&self, py: Python<'py>, double: bool) -> NatsrpyResult<Bound<'py, PyAny>> {
+        self.inner_ack(py, async_nats::jetstream::message::AckKind::Ack, double)
+    }
+
+    #[pyo3(signature=(delay=None, double=false))]
+    pub fn nack<'py>(
+        &self,
+        py: Python<'py>,
+        delay: Option<TimeValue>,
+        double: bool,
+    ) -> NatsrpyResult<Bound<'py, PyAny>> {
+        self.inner_ack(
+            py,
+            async_nats::jetstream::message::AckKind::Nak(delay.map(Into::into)),
+            double,
+        )
+    }
+
+    #[pyo3(signature=(double=false))]
+    pub fn progress<'py>(&self, py: Python<'py>, double: bool) -> NatsrpyResult<Bound<'py, PyAny>> {
+        self.inner_ack(
+            py,
+            async_nats::jetstream::message::AckKind::Progress,
+            double,
+        )
+    }
+
+    #[pyo3(signature=(double=false))]
+    pub fn next<'py>(&self, py: Python<'py>, double: bool) -> NatsrpyResult<Bound<'py, PyAny>> {
+        self.inner_ack(py, async_nats::jetstream::message::AckKind::Next, double)
+    }
+
+    #[pyo3(signature=(double=false))]
+    pub fn term<'py>(&self, py: Python<'py>, double: bool) -> NatsrpyResult<Bound<'py, PyAny>> {
+        self.inner_ack(py, async_nats::jetstream::message::AckKind::Term, double)
     }
 }

src/nats_cls.rs

@@ -13,7 +13,7 @@ use crate::{
         futures::natsrpy_future_with_timeout,
         headers::NatsrpyHeadermapExt,
         natsrpy_future,
-        py_types::{SendableValue, TimeoutValue},
+        py_types::{SendableValue, TimeValue},
     },
 };
 
@@ -28,8 +28,8 @@ pub struct NatsCls {
     read_buffer_capacity: u16,
     sender_capacity: usize,
     max_reconnects: Option<usize>,
-    connection_timeout: TimeoutValue,
-    request_timeout: Option<TimeoutValue>,
+    connection_timeout: TimeValue,
+    request_timeout: Option<TimeValue>,
 }
 
 #[pyo3::pymethods]
@@ -45,8 +45,8 @@ impl NatsCls {
         read_buffer_capacity=65535,
         sender_capacity=128,
         max_reconnects=None,
-        connection_timeout=TimeoutValue::FloatSecs(5.0),
-        request_timeout=TimeoutValue::FloatSecs(10.0),
+        connection_timeout=TimeValue::FloatSecs(5.0),
+        request_timeout=TimeValue::FloatSecs(10.0),
     ))]
     fn __new__(
         addrs: Vec<String>,
@@ -57,8 +57,8 @@ impl NatsCls {
         read_buffer_capacity: u16,
         sender_capacity: usize,
         max_reconnects: Option<usize>,
-        connection_timeout: TimeoutValue,
-        request_timeout: Option<TimeoutValue>,
+        connection_timeout: TimeValue,
+        request_timeout: Option<TimeValue>,
     ) -> Self {
         Self {
             nats_session: Arc::new(RwLock::new(None)),

src/utils/py_types.rs

@@ -44,21 +44,21 @@ impl From<SendableValue> for bytes::Bytes {
 }
 
 #[derive(Clone, Debug, Copy, PartialEq, PartialOrd)]
-pub enum TimeoutValue {
+pub enum TimeValue {
     Duration(Duration),
     FloatSecs(f32),
 }
 
-impl From<TimeoutValue> for Duration {
-    fn from(value: TimeoutValue) -> Self {
+impl From<TimeValue> for Duration {
+    fn from(value: TimeValue) -> Self {
         match value {
-            TimeoutValue::Duration(duration) => duration,
-            TimeoutValue::FloatSecs(fsecs) => Self::from_secs_f32(fsecs),
+            TimeValue::Duration(duration) => duration,
+            TimeValue::FloatSecs(fsecs) => Self::from_secs_f32(fsecs),
         }
     }
 }
 
-impl<'py> FromPyObject<'_, 'py> for TimeoutValue {
+impl<'py> FromPyObject<'_, 'py> for TimeValue {
     type Error = NatsrpyError;
 
     fn extract(obj: pyo3::Borrowed<'_, 'py, pyo3::PyAny>) -> Result<Self, Self::Error> {