Add stream_by_subject

.github/workflows/test.yml

@@ -164,7 +164,7 @@ jobs:
 
  - name: Install msrv Rust on ubuntu-latest
  id: install-rust
- uses: dtolnay/rust-toolchain@1.67.0
+ uses: dtolnay/rust-toolchain@1.70.0
  - name: Cache the build artifacts
  uses: Swatinem/rust-cache@v2
  with:

async-nats/Cargo.toml

@@ -30,7 +30,7 @@ tokio-rustls = { version = "0.26", default-features = false }
 rustls-pemfile = "2"
 nuid = "0.5"
 serde_nanos = "0.1.3"
-time = { version = "0.3.24", features = ["parsing", "formatting", "serde", "serde-well-known"] }
+time = { version = "0.3.36", features = ["parsing", "formatting", "serde", "serde-well-known"] }
 rustls-native-certs = "0.7"
 tracing = "0.1"
 thiserror = "1.0"

async-nats/src/jetstream/context.rs

@@ -19,10 +19,12 @@ use crate::jetstream::account::Account;
 use crate::jetstream::publish::PublishAck;
 use crate::jetstream::response::Response;
 use crate::subject::ToSubject;
-use crate::{header, Client, Command, HeaderMap, HeaderValue, Message, StatusCode};
+use crate::{
+ header, is_valid_subject, Client, Command, HeaderMap, HeaderValue, Message, StatusCode,
+};
 use bytes::Bytes;
 use futures::future::BoxFuture;
-use futures::{Future, TryFutureExt};
+use futures::{Future, StreamExt, TryFutureExt};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use serde_json::{self, json};
@@ -498,6 +500,52 @@ impl Context {
  }
  }
 
+ /// Looks up Stream that contains provided subject.
+ ///
+ /// # Examples
+ ///
+ /// ```no_run
+ /// # #[tokio::main]
+ /// # async fn main() -> Result<(), async_nats::Error> {
+ /// use futures::TryStreamExt;
+ /// let client = async_nats::connect("demo.nats.io:4222").await?;
+ /// let jetstream = async_nats::jetstream::new(client);
+ /// let stream_name = jetstream.stream_by_subject("foo.>");
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub async fn stream_by_subject<T: Into<String>>(
+ &self,
+ subject: T,
+ ) -> Result<String, GetStreamByNameError> {
+ let subject = subject.into();
+ if !is_valid_subject(subject.as_str()) {
+ return Err(GetStreamByNameError::new(
+ GetStreamByNameErrorKind::InvalidSubject,
+ ));
+ }
+ let mut names = StreamNames {
+ context: self.clone(),
+ offset: 0,
+ page_request: None,
+ streams: Vec::new(),
+ subject: Some(subject),
+ done: false,
+ };
+ match names.next().await {
+ Some(name) => match name {
+ Ok(name) => Ok(name),
+ Err(err) => Err(GetStreamByNameError::with_source(
+ GetStreamByNameErrorKind::Request,
+ err,
+ )),
+ },
+ None => Err(GetStreamByNameError::new(
+ GetStreamByNameErrorKind::NotFound,
+ )),
+ }
+ }
+
  /// Lists names of all streams for current context.
  ///
  /// # Examples
@@ -521,6 +569,7 @@ impl Context {
  offset: 0,
  page_request: None,
  streams: Vec::new(),
+ subject: None,
  done: false,
  }
  }
@@ -1297,6 +1346,7 @@ pub struct StreamNames {
  context: Context,
  offset: usize,
  page_request: Option<PageRequest>,
+ subject: Option<String>,
  streams: Vec<String>,
  done: bool,
 }
@@ -1339,12 +1389,14 @@ impl futures::Stream for StreamNames {
  }
  let context = self.context.clone();
  let offset = self.offset;
+ let subject = self.subject.clone();
  self.page_request = Some(Box::pin(async move {
  match context
  .request(
  "STREAM.NAMES",
  &json!({
  "offset": offset,
+ "subject": subject
  }),
  )
  .await?
@@ -1611,7 +1663,27 @@ impl Display for GetStreamErrorKind {
  }
 }
 
+#[derive(Clone, Debug, PartialEq)]
+pub enum GetStreamByNameErrorKind {
+ Request,
+ NotFound,
+ InvalidSubject,
+ JetStream(super::errors::Error),
+}
+
+impl Display for GetStreamByNameErrorKind {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Request => write!(f, "request error"),
+ Self::NotFound => write!(f, "stream not found"),
+ Self::InvalidSubject => write!(f, "invalid subject"),
+ Self::JetStream(err) => write!(f, "jetstream error: {}", err),
+ }
+ }
+}
+
 pub type GetStreamError = Error<GetStreamErrorKind>;
+pub type GetStreamByNameError = Error<GetStreamByNameErrorKind>;
 
 pub type UpdateStreamError = CreateStreamError;
 pub type UpdateStreamErrorKind = CreateStreamErrorKind;

async-nats/src/lib.rs

@@ -1555,9 +1555,11 @@ impl<T: ToServerAddrs + ?Sized> ToServerAddrs for &T {
 }
 
 pub(crate) fn is_valid_subject<T: AsRef<str>>(subject: T) -> bool {
- !subject.as_ref().contains([' ', '.', '\r', '\n'])
+ let subject_str = subject.as_ref();
+ !subject_str.starts_with('.')
+ && !subject_str.ends_with('.')
+ && subject_str.bytes().all(|c| !c.is_ascii_whitespace())
 }
-
 macro_rules! from_with_timeout {
  ($t:ty, $k:ty, $origin: ty, $origin_kind: ty) => {
  impl From<$origin> for $t {

async-nats/tests/jetstream_tests.rs

@@ -36,7 +36,7 @@ mod jetstream {
  self, push, AckPolicy, DeliverPolicy, Info, OrderedPullConsumer, OrderedPushConsumer,
  PullConsumer, PushConsumer, ReplayPolicy,
  };
- use async_nats::jetstream::context::{Publish, PublishErrorKind};
+ use async_nats::jetstream::context::{GetStreamByNameErrorKind, Publish, PublishErrorKind};
  use async_nats::jetstream::response::Response;
  use async_nats::jetstream::stream::{
  self, ConsumerCreateStrictErrorKind, ConsumerUpdateErrorKind, DiscardPolicy, StorageType,
@@ -3654,4 +3654,27 @@ mod jetstream {
  .await
  .expect_err("should fail but not panic because of lack of server info");
  }
+
+ #[tokio::test]
+ async fn test_stream_by_subject() {
+ let server = nats_server::run_server("tests/configs/jetstream.conf");
+ let client = async_nats::connect(server.client_url()).await.unwrap();
+
+ let jetstream = async_nats::jetstream::new(client);
+
+ let stream = jetstream
+ .create_stream(stream::Config {
+ name: "events".to_string(),
+ subjects: vec!["events.>".to_string()],
+ ..Default::default()
+ })
+ .await
+ .unwrap();
+
+ let stream_name = jetstream.stream_by_subject("events.>").await.unwrap();
+ assert_eq!(stream_name, stream.cached_info().config.name);
+
+ let err = jetstream.stream_by_subject("foo").await.unwrap_err();
+ assert_eq!(err.kind(), GetStreamByNameErrorKind::NotFound);
+ }
 }