feat: add pause consumer

Signed-off-by: Yordis Prieto <[email protected]>

async-nats/src/jetstream/consumer/mod.rs

@@ -162,6 +162,10 @@ pub struct Info {
     /// Indicates if any client is connected and receiving messages from a push consumer
     #[serde(default, skip_serializing_if = "is_default")]
     pub push_bound: bool,
+    /// Indicates if the consumer is paused
+    pub paused: bool,
+    /// The remaining time the consumer is paused
+    pub pause_remaining: Option<Duration>,
 }
 
 /// Information about a consumer and the stream it is consuming

async-nats/src/jetstream/stream.rs

@@ -850,6 +850,64 @@ impl Stream {
         }
     }
 
+  /// Pause a [Consumer] until the given time.
+  ///
+  /// # Examples
+  ///
+  /// ```no_run
+  /// # #[tokio::main]
+  /// # async fn main() -> Result<(), async_nats::Error> {
+  /// use async_nats::jetstream::consumer;
+  /// use futures::StreamExt;
+  /// let client = async_nats::connect("localhost:4222").await?;
+  /// let jetstream = async_nats::jetstream::new(client);
+  /// let pause_until = time::OffsetDateTime::now_utc() +  time::Duration::from_secs(60);
+  ///
+  /// jetstream
+  ///     .get_stream("events")
+  ///     .await?
+  ///     .pause_consumer("my_consumer",  pause_until)
+  ///     .await?;
+  /// # Ok(())
+  /// # }
+  /// ```
+  pub async fn pause_consumer(&self, name: &str, pause_until: OffsetDateTime) -> Result<PauseResponse, ConsumerError> {
+    self.request_pause_consumer(name, Some(pause_until)).await
+  }
+
+  /// Resume a paused [Consumer].
+  ///
+  /// # Examples
+  ///
+  /// ```no_run
+  /// # #[tokio::main]
+  /// # async fn main() -> Result<(), async_nats::Error> {
+  /// use async_nats::jetstream::consumer;
+  /// use futures::StreamExt;
+  /// let client = async_nats::connect("localhost:4222").await?;
+  /// let jetstream = async_nats::jetstream::new(client);
+  ///
+  /// jetstream
+  ///     .get_stream("events")
+  ///     .await?
+  ///     .resume_consumer("my_consumer")
+  ///     .await?;
+  /// # Ok(())
+  /// # }
+  /// ```
+  pub async fn resume_consumer(&self, name: &str) -> Result<PauseResponse, ConsumerError> {
+    self.request_pause_consumer(name, None).await
+  }
+
+  async fn request_pause_consumer(&self, name: &str, pause_until: Option<OffsetDateTime>) -> Result<PauseResponse, ConsumerError> {
+    let subject = format!("CONSUMER.PAUSE.{}.{}", self.info.config.name, name);
+    let payload = &PauseConsumerRequest{ pause_until };
+    match self.context.request(subject, payload).await? {
+      Response::Ok::<PauseResponse>(resp) => { Ok(resp) }
+      Response::Err { error } => Err(error.into()),
+    }
+  }
+
     /// Lists names of all consumers for current stream.
     ///
     /// # Examples
@@ -1167,6 +1225,19 @@ pub struct DeleteStatus {
     pub success: bool,
 }
 
+#[derive(Deserialize)]
+pub struct PauseResponse {
+  pub paused: bool,
+  pub pause_until: Option<OffsetDateTime>,
+  pub pause_remaining: Option<Duration>,
+}
+
+#[derive(Serialize)]
+struct PauseConsumerRequest {
+  #[serde(with = "rfc3339", skip_serializing_if = "Option::is_none")]
+  pause_until: Option<OffsetDateTime>,
+}
+
 /// information about the given stream.
 #[derive(Debug, Deserialize, Clone, Copy)]
 pub struct State {