error handling improvement in super_stream send

Author: DanielePalaia

src/error.rs

@@ -110,6 +110,26 @@ pub enum ProducerPublishError {
     #[error(transparent)]
     Client(#[from] ClientError),
 }
+
+#[derive(Error, Debug)]
+pub enum SuperStreamProducerPublishError {
+    #[error("Failed to send message to stream")]
+    ProducerPublishError(),
+    #[error("Failed to create a producer")]
+    ProducerCreateError(),
+}
+
+impl From<ProducerPublishError> for SuperStreamProducerPublishError {
+    fn from(_err: ProducerPublishError) -> Self {
+        SuperStreamProducerPublishError::ProducerPublishError()
+    }
+}
+impl From<ProducerCreateError> for SuperStreamProducerPublishError {
+    fn from(_err: ProducerCreateError) -> Self {
+        SuperStreamProducerPublishError::ProducerCreateError()
+    }
+}
+
 #[derive(Error, Debug)]
 pub enum ProducerCloseError {
     #[error("Failed to close producer for stream {stream} status {status:?}")]

src/superstream_consumer.rs

@@ -23,7 +23,6 @@ struct SuperStreamConsumerInternal {
     closed: Arc<AtomicBool>,
     handlers: Vec<ConsumerHandle>,
     waker: AtomicWaker,
-    //filter_configuration: Option<FilterConfiguration>,
 }
 
 /// Builder for [`Consumer`]
@@ -76,7 +75,6 @@ impl SuperStreamConsumerBuilder {
             closed: Arc::new(AtomicBool::new(false)),
             handlers,
             waker: AtomicWaker::new(),
-            //filter_configuration: self.filter_configuration.clone(),
         };
 
         Ok(SuperStreamConsumer {

src/superstream_producer.rs

@@ -2,7 +2,7 @@ use crate::error::ProducerCloseError;
 use crate::{
     client::Client,
     environment::Environment,
-    error::{ProducerCreateError, ProducerPublishError},
+    error::{ProducerCreateError, ProducerPublishError, SuperStreamProducerPublishError},
     producer::{ConfirmationStatus, NoDedup, Producer},
     superstream::{DefaultSuperStreamMetadata, RoutingStrategy},
 };
@@ -48,7 +48,7 @@ impl SuperStreamProducer<NoDedup> {
             + Sync
             + 'static
             + Clone,
-    ) -> Result<(), ProducerPublishError>
+    ) -> Result<(), SuperStreamProducerPublishError>
     where
         Fut: Future<Output = ()> + Send + Sync + 'static,
     {
@@ -61,6 +61,10 @@ impl SuperStreamProducer<NoDedup> {
             }
         };
 
+        if routes.is_empty() {
+            return Err(crate::error::SuperStreamProducerPublishError::ProducerCreateError());
+        }
+
         for route in routes.into_iter() {
             if !self.1.contains_key(route.as_str()) {
                 let producer = self
@@ -69,8 +73,8 @@ impl SuperStreamProducer<NoDedup> {
                     .producer()
                     .filter_value_extractor_arc(self.0.filter_value_extractor.clone())
                     .build(route.as_str())
-                    .await;
-                self.1.insert(route.clone(), producer.unwrap());
+                    .await?;
+                self.1.insert(route.clone(), producer);
             }
 
             let producer = self.1.get(route.as_str()).unwrap();

tests/integration/producer_test.rs

@@ -440,6 +440,38 @@ async fn key_super_steam_producer_test() {
     _ = super_stream_producer.close();
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn key_super_steam_non_existing_producer_test() {
+    let env = TestEnvironment::create_super_stream().await;
+
+    let mut super_stream_producer = env
+        .env
+        .super_stream_producer(RoutingStrategy::RoutingKeyStrategy(
+            RoutingKeyRoutingStrategy {
+                routing_extractor: &routing_key_strategy_value_extractor,
+            },
+        ))
+        .build("non-existing-stream")
+        .await
+        .unwrap();
+
+    let msg = Message::builder().body(format!("message{}", 0)).build();
+    let result = super_stream_producer
+        .send(msg, |_| async move {})
+        .await
+        .unwrap_err();
+
+    assert_eq!(
+        matches!(
+            result,
+            rabbitmq_stream_client::error::SuperStreamProducerPublishError::ProducerCreateError()
+        ),
+        true
+    );
+
+    _ = super_stream_producer.close();
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn hash_super_steam_producer_test() {
     let env = TestEnvironment::create_super_stream().await;