refactor!: rename JetstreamError2 to JetstreamError

paolobarbolini · paolobarbolini · commit ef550cbe1098 · 2025-07-15T10:47:29.000+02:00
diff --git a/watermelon/src/client/jetstream/commands/consumer_batch.rs b/watermelon/src/client/jetstream/commands/consumer_batch.rs
@@ -11,7 +11,7 @@ use tokio::time::{Sleep, sleep};
 use watermelon_proto::{ServerMessage, StatusCode, error::ServerError};
 
 use crate::{
-    client::{Consumer, JetstreamClient, JetstreamError2},
+    client::{Consumer, JetstreamClient, JetstreamError},
     subscription::Subscription,
 };
 
@@ -43,15 +43,15 @@ impl ConsumerBatch {
         client: JetstreamClient,
         expires: Duration,
         max_msgs: usize,
-    ) -> impl Future<Output = Result<Self, JetstreamError2>> + use<> {
+    ) -> impl Future<Output = Result<Self, JetstreamError>> + use<> {
         let subject = format!(
             "{}.CONSUMER.MSG.NEXT.{}.{}",
             client.prefix, consumer.stream_name, consumer.config.name
         )
         .try_into();
 
         async move {
-            let subject = subject.map_err(JetstreamError2::Subject)?;
+            let subject = subject.map_err(JetstreamError::Subject)?;
             let incoming_subject = client.client.create_inbox_subject();
             let payload = serde_json::to_vec(&if expires.is_zero() {
                 json!({
@@ -65,20 +65,20 @@ impl ConsumerBatch {
                     "no_wait": true
                 })
             })
-            .map_err(JetstreamError2::Json)?;
+            .map_err(JetstreamError::Json)?;
 
             let subscription = client
                 .client
                 .subscribe(incoming_subject.clone(), None)
                 .await
-                .map_err(JetstreamError2::ClientClosed)?;
+                .map_err(JetstreamError::ClientClosed)?;
             client
                 .client
                 .publish(subject)
                 .reply_subject(Some(incoming_subject.clone()))
                 .payload(payload.into())
                 .await
-                .map_err(JetstreamError2::ClientClosed)?;
+                .map_err(JetstreamError::ClientClosed)?;
 
             let timeout = sleep(expires.saturating_add(client.request_timeout));
             Ok(Self {
diff --git a/watermelon/src/client/jetstream/commands/consumer_list.rs b/watermelon/src/client/jetstream/commands/consumer_list.rs
@@ -12,7 +12,7 @@ use serde_json::json;
 use watermelon_proto::Subject;
 
 use crate::{
-    client::{self, JetstreamClient, jetstream::JetstreamError2},
+    client::{self, JetstreamClient, jetstream::JetstreamError},
     util::BoxFuture,
 };
 
@@ -24,7 +24,7 @@ pub struct Consumers {
     client: JetstreamClient,
     offset: u32,
     partial_subject: Subject,
-    fetch: Option<BoxFuture<'static, Result<ConsumersResponse, JetstreamError2>>>,
+    fetch: Option<BoxFuture<'static, Result<ConsumersResponse, JetstreamError>>>,
     buffer: VecDeque<client::Consumer>,
     exhausted: bool,
 }
@@ -52,7 +52,7 @@ impl Consumers {
 }
 
 impl Stream for Consumers {
-    type Item = Result<client::Consumer, JetstreamError2>;
+    type Item = Result<client::Consumer, JetstreamError>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
@@ -83,10 +83,10 @@ impl Stream for Consumers {
                         .into(),
                     )
                     .await
-                    .map_err(JetstreamError2::ClientClosed)?;
-                let response = response_fut.await.map_err(JetstreamError2::ResponseError)?;
-                let payload = serde_json::from_slice(&response.base.payload)
-                    .map_err(JetstreamError2::Json)?;
+                    .map_err(JetstreamError::ClientClosed)?;
+                let response = response_fut.await.map_err(JetstreamError::ResponseError)?;
+                let payload =
+                    serde_json::from_slice(&response.base.payload).map_err(JetstreamError::Json)?;
                 Ok(payload)
             })
         });
diff --git a/watermelon/src/client/jetstream/commands/consumer_stream.rs b/watermelon/src/client/jetstream/commands/consumer_stream.rs
@@ -10,7 +10,7 @@ use pin_project_lite::pin_project;
 use watermelon_proto::ServerMessage;
 
 use crate::{
-    client::{Consumer, JetstreamClient, JetstreamError2},
+    client::{Consumer, JetstreamClient, JetstreamError},
     util::BoxFuture,
 };
 
@@ -36,7 +36,7 @@ pin_project! {
     #[project = ConsumerStreamStatusProj]
     enum ConsumerStreamStatus {
         Polling {
-            future: BoxFuture<'static, Result<ConsumerBatch, JetstreamError2>>,
+            future: BoxFuture<'static, Result<ConsumerBatch, JetstreamError>>,
         },
         RunningBatch {
             #[pin]
@@ -51,7 +51,7 @@ pub enum ConsumerStreamError {
     #[error("consumer batch error")]
     BatchError(#[source] ConsumerBatchError),
     #[error("jetstream error")]
-    Jetstream(#[source] JetstreamError2),
+    Jetstream(#[source] JetstreamError),
 }
 
 impl ConsumerStream {
diff --git a/watermelon/src/client/jetstream/commands/stream_list.rs b/watermelon/src/client/jetstream/commands/stream_list.rs
@@ -11,7 +11,7 @@ use serde_json::json;
 use watermelon_proto::Subject;
 
 use crate::{
-    client::{self, JetstreamClient, jetstream::JetstreamError2},
+    client::{self, JetstreamClient, jetstream::JetstreamError},
     util::BoxFuture,
 };
 
@@ -22,7 +22,7 @@ use crate::{
 pub struct Streams {
     client: JetstreamClient,
     offset: u32,
-    fetch: Option<BoxFuture<'static, Result<StreamsResponse, JetstreamError2>>>,
+    fetch: Option<BoxFuture<'static, Result<StreamsResponse, JetstreamError>>>,
     buffer: VecDeque<client::Stream>,
     exhausted: bool,
 }
@@ -46,7 +46,7 @@ impl Streams {
 }
 
 impl Stream for Streams {
-    type Item = Result<client::Stream, JetstreamError2>;
+    type Item = Result<client::Stream, JetstreamError>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
@@ -76,10 +76,10 @@ impl Stream for Streams {
                         .into(),
                     )
                     .await
-                    .map_err(JetstreamError2::ClientClosed)?;
-                let response = response_fut.await.map_err(JetstreamError2::ResponseError)?;
-                let payload = serde_json::from_slice(&response.base.payload)
-                    .map_err(JetstreamError2::Json)?;
+                    .map_err(JetstreamError::ClientClosed)?;
+                let response = response_fut.await.map_err(JetstreamError::ResponseError)?;
+                let payload =
+                    serde_json::from_slice(&response.base.payload).map_err(JetstreamError::Json)?;
                 Ok(payload)
             })
         });
diff --git a/watermelon/src/client/jetstream/mod.rs b/watermelon/src/client/jetstream/mod.rs
@@ -53,7 +53,7 @@ pub struct JetstreamErrorCode(u16);
 
 /// An error encountered while making a Jetstream request
 #[derive(Debug, thiserror::Error)]
-pub enum JetstreamError2 {
+pub enum JetstreamError {
     #[error("invalid subject")]
     Subject(#[source] SubjectValidateError),
     #[error("client closed")]
@@ -102,24 +102,24 @@ impl JetstreamClient {
     ///
     /// It returns an error if the stream name produces an invalid subject or if an error occurs
     /// while creating the stream.
-    pub async fn create_stream(&self, config: &StreamConfig) -> Result<Stream, JetstreamError2> {
+    pub async fn create_stream(&self, config: &StreamConfig) -> Result<Stream, JetstreamError> {
         let subject = format!("{}.STREAM.CREATE.{}", self.prefix, config.name)
             .try_into()
-            .map_err(JetstreamError2::Subject)?;
+            .map_err(JetstreamError::Subject)?;
 
-        let payload = serde_json::to_vec(config).map_err(JetstreamError2::Json)?;
+        let payload = serde_json::to_vec(config).map_err(JetstreamError::Json)?;
         let resp = self
             .make_request(subject)
             .payload(payload.into())
             .await
-            .map_err(JetstreamError2::ClientClosed)?;
-        let resp = resp.await.map_err(JetstreamError2::ResponseError)?;
+            .map_err(JetstreamError::ClientClosed)?;
+        let resp = resp.await.map_err(JetstreamError::ResponseError)?;
 
         let json = serde_json::from_slice::<Response<Stream>>(&resp.base.payload)
-            .map_err(JetstreamError2::Json)?;
+            .map_err(JetstreamError::Json)?;
         match json {
             Response::Response(stream) => Ok(stream),
-            Response::Error { error } => Err(JetstreamError2::Api(error)),
+            Response::Error { error } => Err(JetstreamError::Api(error)),
         }
     }
 
@@ -134,25 +134,25 @@ impl JetstreamClient {
     ///
     /// It returns an error if the given `name` produces an invalid subject or if an error occurs
     /// while creating the stream.
-    pub async fn stream(&self, name: impl Display) -> Result<Option<Stream>, JetstreamError2> {
+    pub async fn stream(&self, name: impl Display) -> Result<Option<Stream>, JetstreamError> {
         let subject = format!("{}.STREAM.INFO.{}", self.prefix, name)
             .try_into()
-            .map_err(JetstreamError2::Subject)?;
+            .map_err(JetstreamError::Subject)?;
         let resp = self
             .make_request(subject)
             .payload(Bytes::new())
             .await
-            .map_err(JetstreamError2::ClientClosed)?;
-        let resp = resp.await.map_err(JetstreamError2::ResponseError)?;
+            .map_err(JetstreamError::ClientClosed)?;
+        let resp = resp.await.map_err(JetstreamError::ResponseError)?;
 
         let json = serde_json::from_slice::<Response<Stream>>(&resp.base.payload)
-            .map_err(JetstreamError2::Json)?;
+            .map_err(JetstreamError::Json)?;
         match json {
             Response::Response(stream) => Ok(Some(stream)),
             Response::Error { error } if error.code == JetstreamErrorCode::STREAM_NOT_FOUND => {
                 Ok(None)
             }
-            Response::Error { error } => Err(JetstreamError2::Api(error)),
+            Response::Error { error } => Err(JetstreamError::Api(error)),
         }
     }
 
@@ -166,7 +166,7 @@ impl JetstreamClient {
         &self,
         stream_name: &str,
         config: &ConsumerConfig,
-    ) -> Result<Consumer, JetstreamError2> {
+    ) -> Result<Consumer, JetstreamError> {
         let mut subject = format!(
             "{}.CONSUMER.CREATE.{}.{}",
             self.prefix, stream_name, config.name
@@ -176,27 +176,27 @@ impl JetstreamClient {
             subject.push_str(filter_subject);
         }
 
-        let subject = subject.try_into().map_err(JetstreamError2::Subject)?;
+        let subject = subject.try_into().map_err(JetstreamError::Subject)?;
 
         let payload = serde_json::to_vec(&json!({
             "stream_name": stream_name,
             "config": config,
             "action": "create",
             "pedantic": true,
         }))
-        .map_err(JetstreamError2::Json)?;
+        .map_err(JetstreamError::Json)?;
         let resp = self
             .make_request(subject)
             .payload(payload.into())
             .await
-            .map_err(JetstreamError2::ClientClosed)?;
-        let resp = resp.await.map_err(JetstreamError2::ResponseError)?;
+            .map_err(JetstreamError::ClientClosed)?;
+        let resp = resp.await.map_err(JetstreamError::ResponseError)?;
 
         let json = serde_json::from_slice::<Response<Consumer>>(&resp.base.payload)
-            .map_err(JetstreamError2::Json)?;
+            .map_err(JetstreamError::Json)?;
         match json {
             Response::Response(consumer) => Ok(consumer),
-            Response::Error { error } => Err(JetstreamError2::Api(error)),
+            Response::Error { error } => Err(JetstreamError::Api(error)),
         }
     }
 
@@ -215,28 +215,28 @@ impl JetstreamClient {
         &self,
         stream_name: impl Display,
         consumer_name: impl Display,
-    ) -> Result<Option<Consumer>, JetstreamError2> {
+    ) -> Result<Option<Consumer>, JetstreamError> {
         let subject = format!(
             "{}.CONSUMER.INFO.{}.{}",
             self.prefix, stream_name, consumer_name
         )
         .try_into()
-        .map_err(JetstreamError2::Subject)?;
+        .map_err(JetstreamError::Subject)?;
         let resp = self
             .make_request(subject)
             .payload(Bytes::new())
             .await
-            .map_err(JetstreamError2::ClientClosed)?;
-        let resp = resp.await.map_err(JetstreamError2::ResponseError)?;
+            .map_err(JetstreamError::ClientClosed)?;
+        let resp = resp.await.map_err(JetstreamError::ResponseError)?;
 
         let json = serde_json::from_slice::<Response<Consumer>>(&resp.base.payload)
-            .map_err(JetstreamError2::Json)?;
+            .map_err(JetstreamError::Json)?;
         match json {
             Response::Response(stream) => Ok(Some(stream)),
             Response::Error { error } if error.code == JetstreamErrorCode::CONSUMER_NOT_FOUND => {
                 Ok(None)
             }
-            Response::Error { error } => Err(JetstreamError2::Api(error)),
+            Response::Error { error } => Err(JetstreamError::Api(error)),
         }
     }
 
@@ -250,7 +250,7 @@ impl JetstreamClient {
         consumer: &Consumer,
         expires: Duration,
         max_msgs: usize,
-    ) -> Result<ConsumerBatch, JetstreamError2> {
+    ) -> Result<ConsumerBatch, JetstreamError> {
         ConsumerBatch::new(consumer, self.clone(), expires, max_msgs).await
     }
 
diff --git a/watermelon/src/client/mod.rs b/watermelon/src/client/mod.rs
@@ -28,7 +28,7 @@ pub use self::commands::{
 pub use self::jetstream::{
     AckPolicy, Compression, Consumer, ConsumerBatch, ConsumerConfig, ConsumerDurability,
     ConsumerSpecificConfig, ConsumerStorage, ConsumerStream, ConsumerStreamError, Consumers,
-    DeliverPolicy, DiscardPolicy, JetstreamApiError, JetstreamClient, JetstreamError2,
+    DeliverPolicy, DiscardPolicy, JetstreamApiError, JetstreamClient, JetstreamError,
     JetstreamErrorCode, ReplayPolicy, RetentionPolicy, Storage, Stream, StreamConfig, StreamState,
     Streams,
 };
diff --git a/watermelon/src/lib.rs b/watermelon/src/lib.rs
@@ -63,7 +63,7 @@ pub mod jetstream {
         //! NATS Jetstream specific errors
 
         pub use crate::client::{
-            ConsumerStreamError, JetstreamApiError, JetstreamError2, JetstreamErrorCode,
+            ConsumerStreamError, JetstreamApiError, JetstreamError, JetstreamErrorCode,
         };
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ pub mod jetstream {`
`63`	`63`	`//! NATS Jetstream specific errors`
`64`	`64`
`65`	`65`	`pub use crate::client::{`
`66`		`- ConsumerStreamError, JetstreamApiError, JetstreamError2, JetstreamErrorCode,`
	`66`	`+ ConsumerStreamError, JetstreamApiError, JetstreamError, JetstreamErrorCode,`
`67`	`67`	`};`
`68`	`68`	`}`
`69`	`69`	`}`