Add Context::get_consumer_from_stream

Signed-off-by: Tomasz Pietrek <[email protected]>

async-nats/src/jetstream/context.rs

@@ -33,6 +33,7 @@ use std::task::Poll;
 use std::time::Duration;
 use tracing::debug;
 
+use super::consumer::{Consumer, FromConsumer, IntoConsumerConfig};
 use super::kv::{Store, MAX_HISTORY};
 use super::object_store::{is_valid_bucket_name, ObjectStore};
 use super::stream::{self, Config, DeleteStatus, DiscardPolicy, External, Info, Stream};
@@ -751,6 +752,58 @@ impl Context {
     //         .and_then(|info| Ok(()))
     // }
 
+    /// Get a [crate::jetstream::consumer::Consumer] straight from [Context], without binding to a [Stream] first.
+    ///
+    /// It has one less interaction with the server when binding to only one
+    /// [crate::jetstream::consumer::Consumer].
+    ///
+    /// # Examples:
+    ///
+    /// ```no_run
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), async_nats::Error> {
+    /// use async_nats::jetstream::consumer::PullConsumer;
+    ///
+    /// let client = async_nats::connect("localhost:4222").await?;
+    /// let jetstream = async_nats::jetstream::new(client);
+    ///
+    /// let consumer: PullConsumer = jetstream.get_consumer_from_stream("consumer", "stream").await?;
+    ///
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn get_consumer_from_stream<T, C, S>(
+        &self,
+        consumer: C,
+        stream: S,
+    ) -> Result<Consumer<T>, Error>
+    where
+        T: FromConsumer + IntoConsumerConfig,
+        S: AsRef<str>,
+        C: AsRef<str>,
+    {
+        let subject = format!("CONSUMER.INFO.{}.{}", stream.as_ref(), consumer.as_ref());
+
+        let info: super::consumer::Info = match self.request(subject, &json!({})).await? {
+            Response::Ok(info) => info,
+            Response::Err { error } => {
+                return Err(Box::new(std::io::Error::new(
+                    ErrorKind::Other,
+                    format!(
+                        "nats: error while getting consumer info: {}, {}, {}",
+                        error.code, error.status, error.description
+                    ),
+                )))
+            }
+        };
+
+        Ok(Consumer::new(
+            T::try_from_consumer_config(info.config.clone())?,
+            info,
+            self.clone(),
+        ))
+    }
+
     /// Send a request to the jetstream JSON API.
     ///
     /// This is a low level API used mostly internally, that should be used only in

async-nats/tests/jetstream_tests.rs

@@ -1073,6 +1073,34 @@ mod jetstream {
         let consumer = stream.get_consumer("pull").await.unwrap();
         consumer.fetch().max_messages(10).messages().await.unwrap();
     }
+    #[tokio::test]
+    async fn get_consumer_from_stream() {
+        let server = nats_server::run_server("tests/configs/jetstream.conf");
+        let client = async_nats::connect(server.client_url()).await.unwrap();
+        let context = async_nats::jetstream::new(client);
+
+        let stream = context.get_or_create_stream("stream").await.unwrap();
+        stream
+            .create_consumer(consumer::pull::Config {
+                durable_name: Some("pull".to_string()),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        stream
+            .create_consumer(consumer::push::Config {
+                durable_name: Some("push".to_string()),
+                deliver_subject: "subject".to_string(),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        let _consumer: PullConsumer = context
+            .get_consumer_from_stream("pull", "stream")
+            .await
+            .unwrap();
+    }
 
     #[tokio::test]
     async fn get_or_create_consumer() {