Direct get

async-nats/src/jetstream/stream.rs

@@ -14,19 +14,21 @@
 //! Manage operations on a [Stream], create/delete/update [Consumer][crate::jetstream::consumer::Consumer].
 
 use std::{
+    fmt::Debug,
     io::{self, ErrorKind},
     time::Duration,
 };
 
 use crate::Error;
+use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use time::serde::rfc3339;
 
 use super::{
     consumer::{self, Consumer, FromConsumer, IntoConsumerConfig},
     response::Response,
-    Context,
+    Context, Message,
 };
 
 /// Handle to operations that can be performed on a `Stream`.
@@ -95,6 +97,130 @@ impl Stream {
     pub fn cached_info(&self) -> &Info {
         &self.info
     }
+
+    pub async fn direct_get_next_for_subject_after_sequence(
+        &self,
+        subject: String,
+        sequence: u64,
+    ) -> Result<Message, Error> {
+        let request_subject = format!(
+            "{}.DIRECT.GET.{}",
+            &self.context.prefix, &self.info.config.name
+        );
+        let payload = json!({
+            "seq": sequence,
+            "next_by_subj": subject,
+        });
+
+        let response = self
+            .context
+            .client
+            .request(
+                request_subject,
+                serde_json::to_vec(&payload).map(Bytes::from)?,
+            )
+            .await
+            .map(|message| Message {
+                message,
+                context: self.context.clone(),
+            })?;
+        if let Some(status) = response.status {
+            if let Some(ref description) = response.description {
+                return Err(Box::from(std::io::Error::new(
+                    ErrorKind::Other,
+                    format!("{} {}", status, description),
+                )));
+            }
+        }
+        Ok(response)
+    }
+    pub async fn direct_get_next_for_subject(&self, subject: String) -> Result<Message, Error> {
+        let request_subject = format!(
+            "{}.DIRECT.GET.{}",
+            &self.context.prefix, &self.info.config.name
+        );
+        let payload = json!({
+            "next_by_subj": subject,
+        });
+
+        let response = self
+            .context
+            .client
+            .request(
+                request_subject,
+                serde_json::to_vec(&payload).map(Bytes::from)?,
+            )
+            .await
+            .map(|message| Message {
+                message,
+                context: self.context.clone(),
+            })?;
+        if let Some(status) = response.status {
+            if let Some(ref description) = response.description {
+                return Err(Box::from(std::io::Error::new(
+                    ErrorKind::Other,
+                    format!("{} {}", status, description),
+                )));
+            }
+        }
+        Ok(response)
+    }
+
+    pub async fn direct_get_by_sequence(&self, sequence: u64) -> Result<Message, Error> {
+        let subject = format!(
+            "{}.DIRECT.GET.{}",
+            &self.context.prefix, &self.info.config.name
+        );
+        let payload = json!({
+            "seq": sequence,
+        });
+
+        let response = self
+            .context
+            .client
+            .request(subject, serde_json::to_vec(&payload).map(Bytes::from)?)
+            .await
+            .map(|message| Message {
+                context: self.context.clone(),
+                message,
+            })?;
+
+        if let Some(status) = response.status {
+            if let Some(ref description) = response.description {
+                return Err(Box::from(std::io::Error::new(
+                    ErrorKind::Other,
+                    format!("{} {}", status, description),
+                )));
+            }
+        }
+        Ok(response)
+    }
+
+    pub async fn direct_get_last_for_subject(&self, subject: String) -> Result<Message, Error> {
+        let subject = format!(
+            "{}.DIRECT.GET.{}.{}",
+            &self.context.prefix, &self.info.config.name, subject
+        );
+
+        let response = self
+            .context
+            .client
+            .request(subject, "".into())
+            .await
+            .map(|message| Message {
+                context: self.context.clone(),
+                message,
+            })?;
+        if let Some(status) = response.status {
+            if let Some(ref description) = response.description {
+                return Err(Box::from(std::io::Error::new(
+                    ErrorKind::Other,
+                    format!("{} {}", status, description),
+                )));
+            }
+        }
+        Ok(response)
+    }
     /// Get a raw message from the stream.
     ///
     /// # Examples
@@ -564,6 +690,9 @@ pub struct Config {
 
     #[serde(default, skip_serializing_if = "is_default")]
     pub republish: Option<Republish>,
+
+    #[serde(default, skip_serializing_if = "is_default")]
+    pub allow_direct: bool,
 }
 
 impl From<&Config> for Config {