enabling naming for super_stream consumers and setting up sac properties internally

Author: DanielePalaia

examples/single_active_consumer/single_active_consumer_super_stream.rs

@@ -34,23 +34,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         super_stream
     );
 
-    let mut properties = HashMap::new();
-
-    properties.insert("single-active-consumer".to_string(), "true".to_string());
-    properties.insert("name".to_string(), "consumer-group-1".to_string());
-    properties.insert("super-stream".to_string(), "hello-rust-super-stream".to_string());
-
     let mut super_stream_consumer: SuperStreamConsumer = environment
         .super_stream_consumer()
+        // Mandatory if sac is enabled
+        .name("consumer-group-1")
         .offset(OffsetSpecification::First)
+        .enable_single_active_consumer(true)
         .client_provided_name("my super stream consumer for hello rust")
-        /*We can decide a strategy to manage Offset specification in single active consumer based on is_active flag
-        By default if this clousure is not present the default strategy OffsetSpecification::NEXT will be set.*/
         .consumer_update(move |active, message_context| {
             println!("single active consumer: is active: {} on stream {}", active, message_context.get_stream());
             OffsetSpecification::First
         })
-        .properties(properties)
         .build(super_stream)
         .await
         .unwrap();

src/consumer.rs

@@ -113,13 +113,20 @@ pub struct ConsumerBuilder {
     pub(crate) consumer_update_listener: Option<ConsumerUpdateListener>,
     pub(crate) client_provided_name: String,
     pub(crate) properties: HashMap<String, String>,
+    pub(crate) is_single_active_consumer: bool,
 }
 
 impl ConsumerBuilder {
     pub async fn build(mut self, stream: &str) -> Result<Consumer, ConsumerCreateError> {
+        if (self.is_single_active_consumer
+            || self.properties.contains_key("single-active-consumer"))
+            && self.consumer_name.is_none()
+        {
+            return Err(ConsumerCreateError::SingleActiveConsumerNotSupported);
+        }
+
         // Connect to the user specified node first, then look for a random replica to connect to instead.
         // This is recommended for load balancing purposes
-
         let mut opt_with_client_provided_name = self.environment.options.client_options.clone();
         opt_with_client_provided_name.client_provided_name = self.client_provided_name.clone();
 
@@ -203,6 +210,13 @@ impl ConsumerBuilder {
             );
         }
 
+        if self.is_single_active_consumer {
+            self.properties
+                .insert("single-active-consumer".to_string(), "true".to_string());
+            self.properties
+                .insert("name".to_string(), self.consumer_name.clone().unwrap());
+        }
+
         let response = client
             .subscribe(
                 subscription_id,
@@ -242,6 +256,16 @@ impl ConsumerBuilder {
         self
     }
 
+    pub fn name_optional(mut self, consumer_name: Option<String>) -> Self {
+        self.consumer_name = consumer_name;
+        self
+    }
+
+    pub fn enable_single_active_consumer(mut self, is_single_active_consumer: bool) -> Self {
+        self.is_single_active_consumer = is_single_active_consumer;
+        self
+    }
+
     pub fn filter_input(mut self, filter_configuration: Option<FilterConfiguration>) -> Self {
         self.filter_configuration = filter_configuration;
         self

src/environment.rs

@@ -77,17 +77,20 @@ impl Environment {
             consumer_update_listener: None,
             client_provided_name: String::from("rust-stream-consumer"),
             properties: HashMap::new(),
+            is_single_active_consumer: false,
         }
     }
 
     pub fn super_stream_consumer(&self) -> SuperStreamConsumerBuilder {
         SuperStreamConsumerBuilder {
+            super_stream_consumer_name: None,
             environment: self.clone(),
             offset_specification: OffsetSpecification::Next,
             filter_configuration: None,
             consumer_update_listener: None,
             client_provided_name: String::from("rust-super-stream-consumer"),
             properties: HashMap::new(),
+            is_single_active_consumer: false,
         }
     }
 

src/error.rs

@@ -159,6 +159,9 @@ pub enum ConsumerCreateError {
 
     #[error("Filtering is not supported by the broker (requires RabbitMQ 3.13+ and stream_filtering feature flag activated)")]
     FilteringNotSupport,
+
+    #[error("if you set single active consumer a consumer and super_stream consumer name need to be setup")]
+    SingleActiveConsumerNotSupported,
 }
 
 #[derive(Error, Debug)]

src/superstream_consumer.rs

@@ -32,21 +32,30 @@ struct SuperStreamConsumerInternal {
 
 /// Builder for [`Consumer`]
 pub struct SuperStreamConsumerBuilder {
+    pub(crate) super_stream_consumer_name: Option<String>,
     pub(crate) environment: Environment,
     pub(crate) offset_specification: OffsetSpecification,
     pub(crate) filter_configuration: Option<FilterConfiguration>,
     pub(crate) consumer_update_listener: Option<ConsumerUpdateListener>,
     pub(crate) client_provided_name: String,
+    pub(crate) is_single_active_consumer: bool,
     pub(crate) properties: HashMap<String, String>,
 }
 
 impl SuperStreamConsumerBuilder {
     pub async fn build(
-        self,
+        &mut self,
         super_stream: &str,
     ) -> Result<SuperStreamConsumer, ConsumerCreateError> {
         // Connect to the user specified node first, then look for a random replica to connect to instead.
         // This is recommended for load balancing purposes.
+        if (self.is_single_active_consumer
+            || self.properties.contains_key("single-active-consumer"))
+            && self.super_stream_consumer_name.is_none()
+        {
+            return Err(ConsumerCreateError::SingleActiveConsumerNotSupported);
+        }
+
         let client = self.environment.create_client().await?;
         let (tx, rx) = channel(10000);
 
@@ -58,17 +67,24 @@ impl SuperStreamConsumerBuilder {
         };
         let partitions = super_stream_metadata.partitions().await;
 
+        if self.is_single_active_consumer {
+            self.properties
+                .insert("super-stream".to_string(), super_stream.to_string());
+        }
+
         let mut handlers = Vec::<ConsumerHandle>::new();
         for partition in partitions.into_iter() {
             let tx_cloned = tx.clone();
             let mut consumer = self
                 .environment
                 .consumer()
+                .name_optional(self.super_stream_consumer_name.clone())
                 .offset(self.offset_specification.clone())
                 .client_provided_name(self.client_provided_name.as_str())
                 .filter_input(self.filter_configuration.clone())
                 .consumer_update_arc(self.consumer_update_listener.clone())
                 .properties(self.properties.clone())
+                .enable_single_active_consumer(self.is_single_active_consumer)
                 .build(partition.as_str())
                 .await
                 .unwrap();
@@ -100,6 +116,16 @@ impl SuperStreamConsumerBuilder {
         self
     }
 
+    pub fn name(mut self, consumer_name: &str) -> Self {
+        self.super_stream_consumer_name = Some(String::from(consumer_name));
+        self
+    }
+
+    pub fn enable_single_active_consumer(mut self, is_single_active_consumer: bool) -> Self {
+        self.is_single_active_consumer = is_single_active_consumer;
+        self
+    }
+
     pub fn filter_input(mut self, filter_configuration: Option<FilterConfiguration>) -> Self {
         self.filter_configuration = filter_configuration;
         self

tests/integration/consumer_test.rs

@@ -687,36 +687,34 @@ async fn super_stream_single_active_consumer_test() {
         .await
         .unwrap();
 
-    let mut properties = HashMap::new();
     let notify_received_messages = Arc::new(Notify::new());
 
-    properties.insert("single-active-consumer".to_string(), "true".to_string());
-    properties.insert("name".to_string(), "consumer-group-1".to_string());
-    properties.insert("super-stream".to_string(), env.super_stream.clone());
-
     let mut super_stream_consumer: SuperStreamConsumer = env
         .env
         .super_stream_consumer()
+        .name("super-stream-with-sac-enabled")
+        .enable_single_active_consumer(true)
         .offset(OffsetSpecification::First)
-        .properties(properties.clone())
         .build(&env.super_stream)
         .await
         .unwrap();
 
     let mut super_stream_consumer_2: SuperStreamConsumer = env
         .env
         .super_stream_consumer()
+        .name("super-stream-with-sac-enabled")
+        .enable_single_active_consumer(true)
         .offset(OffsetSpecification::First)
-        .properties(properties.clone())
         .build(&env.super_stream)
         .await
         .unwrap();
 
     let mut super_stream_consumer_3: SuperStreamConsumer = env
         .env
         .super_stream_consumer()
+        .name("super-stream-with-sac-enabled")
+        .enable_single_active_consumer(true)
         .offset(OffsetSpecification::First)
-        .properties(properties.clone())
         .build(&env.super_stream)
         .await
         .unwrap();
@@ -803,60 +801,58 @@ async fn super_stream_single_active_consumer_test_with_callback() {
         .await
         .unwrap();
 
-    let mut properties = HashMap::new();
     let notify_received_messages = Arc::new(Notify::new());
 
     let mut result_stream_name_1 = Arc::new(Mutex::new(String::from("")));
     let mut result_stream_name_2 = Arc::new(Mutex::new(String::from("")));
     let mut result_stream_name_3 = Arc::new(Mutex::new(String::from("")));
 
-    properties.insert("single-active-consumer".to_string(), "true".to_string());
-    properties.insert("name".to_string(), "consumer-group-1".to_string());
-    properties.insert("super-stream".to_string(), env.super_stream.clone());
-
     let mut result_stream_name_outer = result_stream_name_1.clone();
     let mut result_stream_name_2_outer = result_stream_name_2.clone();
     let mut result_stream_name_3_outer = result_stream_name_3.clone();
 
     let mut super_stream_consumer: SuperStreamConsumer = env
         .env
         .super_stream_consumer()
+        .name("super-stream-with-sac-enabled")
+        .enable_single_active_consumer(true)
         .offset(OffsetSpecification::First)
         .consumer_update(move |active, message_context| {
             let mut result_consumer_name_int = result_stream_name_outer.clone();
             *result_consumer_name_int.lock().unwrap() = message_context.get_stream().clone();
 
             OffsetSpecification::First
         })
-        .properties(properties.clone())
         .build(&env.super_stream)
         .await
         .unwrap();
 
     let mut super_stream_consumer_2: SuperStreamConsumer = env
         .env
         .super_stream_consumer()
+        .name("super-stream-with-sac-enabled")
+        .enable_single_active_consumer(true)
         .offset(OffsetSpecification::First)
         .consumer_update(move |active, message_context| {
             let mut result_consumer_name_int = result_stream_name_2_outer.clone();
             *result_consumer_name_int.lock().unwrap() = message_context.get_stream().clone();
             OffsetSpecification::First
         })
-        .properties(properties.clone())
         .build(&env.super_stream)
         .await
         .unwrap();
 
     let mut super_stream_consumer_3: SuperStreamConsumer = env
         .env
         .super_stream_consumer()
+        .name("super-stream-with-sac-enabled")
+        .enable_single_active_consumer(true)
         .offset(OffsetSpecification::First)
         .consumer_update(move |active, message_context| {
             let mut result_consumer_name_int = result_stream_name_3_outer.clone();
             *result_consumer_name_int.lock().unwrap() = message_context.get_stream().clone();
             OffsetSpecification::First
         })
-        .properties(properties.clone())
         .build(&env.super_stream)
         .await
         .unwrap();