add client_provided_name

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

examples/environment.rs

@@ -28,7 +28,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut producer = environment
         .producer()
-        .name("test_producer")
+        .client_provided_name("my producer")
         .build("test")
         .await?;
 
@@ -42,6 +42,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut consumer = environment
         .consumer()
+        .client_provided_name("my consumer")
         .offset(OffsetSpecification::First)
         .build("test")
         .await

src/client/mod.rs

@@ -231,9 +231,14 @@ impl Client {
             client_properties: HashMap::new(),
         };
 
+        const VERSION: &str = env!("CARGO_PKG_VERSION");
+
         client
             .client_properties
             .insert(String::from("product"), String::from("RabbitMQ"));
+        client
+            .client_properties
+            .insert(String::from("version"), String::from(VERSION));
         client
             .client_properties
             .insert(String::from("platform"), String::from("Rust"));
@@ -259,6 +264,7 @@ impl Client {
             client.filtering_supported = true
         }
 
+        println!("Connect {}", client.opts.client_provided_name.clone());
         Ok(client)
     }
 

src/client/options.rs

@@ -72,6 +72,10 @@ impl ClientOptions {
     pub fn set_port(&mut self, port: u16) {
         self.port = port;
     }
+
+    pub fn set_client_provided_name(&mut self, name: &str) {
+        self.client_provided_name = name.to_owned();
+    }
 }
 
 pub struct ClientOptionsBuilder(ClientOptions);

src/consumer.rs

@@ -88,13 +88,21 @@ pub struct ConsumerBuilder {
     pub(crate) environment: Environment,
     pub(crate) offset_specification: OffsetSpecification,
     pub(crate) filter_configuration: Option<FilterConfiguration>,
+    pub(crate) client_provided_name: String,
 }
 
 impl ConsumerBuilder {
     pub async fn build(self, stream: &str) -> Result<Consumer, 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.
-        let mut client = self.environment.create_client().await?;
+        // 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();
+
+        let mut client = self
+            .environment
+            .create_client_with_options(opt_with_client_provided_name)
+            .await?;
         let collector = self.environment.options.client_options.collector.clone();
         if let Some(metadata) = client.metadata(vec![stream.to_string()]).await?.get(stream) {
             // If there are no replicas we do not reassign client, meaning we just keep reading from the leader.
@@ -197,6 +205,11 @@ impl ConsumerBuilder {
         self
     }
 
+    pub fn client_provided_name(mut self, name: &str) -> Self {
+        self.client_provided_name = String::from(name);
+        self
+    }
+
     pub fn name(mut self, consumer_name: &str) -> Self {
         self.consumer_name = Some(String::from(consumer_name));
         self

src/environment.rs

@@ -46,6 +46,7 @@ impl Environment {
             batch_publishing_delay: Duration::from_millis(100),
             data: PhantomData,
             filter_value_extractor: None,
+            client_provided_name: String::from("rust-stream-producer"),
         }
     }
 
@@ -56,11 +57,18 @@ impl Environment {
             environment: self.clone(),
             offset_specification: OffsetSpecification::Next,
             filter_configuration: None,
+            client_provided_name: String::from("rust-stream-consumer"),
         }
     }
     pub(crate) async fn create_client(&self) -> RabbitMQStreamResult<Client> {
         Client::connect(self.options.client_options.clone()).await
     }
+    pub(crate) async fn create_client_with_options(
+        &self,
+        opts: impl Into<ClientOptions>,
+    ) -> RabbitMQStreamResult<Client> {
+        Client::connect(opts).await
+    }
 
     /// Delete a stream
     pub async fn delete_stream(&self, stream: &str) -> Result<(), StreamDeleteError> {
@@ -148,6 +156,11 @@ impl EnvironmentBuilder {
         self.0.client_options.load_balancer_mode = load_balancer_mode;
         self
     }
+
+    pub fn client_provided_name(mut self, name: &str) -> EnvironmentBuilder {
+        self.0.client_options.client_provided_name = name.to_owned();
+        self
+    }
 }
 #[derive(Clone, Default)]
 pub struct EnvironmentOptions {

src/producer.rs

@@ -105,6 +105,7 @@ pub struct ProducerBuilder<T> {
     pub batch_publishing_delay: Duration,
     pub(crate) data: PhantomData<T>,
     pub filter_value_extractor: Option<FilterValueExtractor>,
+    pub(crate) client_provided_name: String,
 }
 
 #[derive(Clone)]
@@ -117,7 +118,14 @@ impl<T> ProducerBuilder<T> {
         // Connect to the user specified node first, then look for the stream leader.
         // The leader is the recommended node for writing, because writing to a replica will redundantly pass these messages
         // to the leader anyway - it is the only one capable of writing.
-        let mut client = self.environment.create_client().await?;
+
+        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();
+
+        let mut client = self
+            .environment
+            .create_client_with_options(opt_with_client_provided_name.clone())
+            .await?;
 
         let mut publish_version = 1;
 
@@ -155,7 +163,7 @@ impl<T> ProducerBuilder<T> {
                 client = Client::connect(ClientOptions {
                     host: metadata.leader.host.clone(),
                     port: metadata.leader.port as u16,
-                    ..self.environment.options.client_options
+                    ..opt_with_client_provided_name.clone()
                 })
                 .await?
             };
@@ -225,6 +233,12 @@ impl<T> ProducerBuilder<T> {
         self.batch_publishing_delay = delay;
         self
     }
+
+    pub fn client_provided_name(mut self, name: &str) -> Self {
+        self.client_provided_name = String::from(name);
+        self
+    }
+
     pub fn name(mut self, name: &str) -> ProducerBuilder<Dedup> {
         self.name = Some(name.to_owned());
         ProducerBuilder {
@@ -234,6 +248,7 @@ impl<T> ProducerBuilder<T> {
             batch_publishing_delay: self.batch_publishing_delay,
             data: PhantomData,
             filter_value_extractor: None,
+            client_provided_name: String::from("rust-stream-producer"),
         }
     }
 

tests/integration/client_test.rs

@@ -17,7 +17,15 @@ use crate::common::TestClient;
 #[tokio::test]
 async fn client_connection_test() {
     let client = Client::connect(ClientOptions::default()).await.unwrap();
+    assert_ne!(client.server_properties().await.len(), 0);
+    assert_ne!(client.connection_properties().await.len(), 0);
+}
 
+#[tokio::test]
+async fn client_connection_with_properties_test() {
+    let mut opts = ClientOptions::default();
+    opts.set_client_provided_name("my_connection_name");
+    let client = Client::connect(opts).await.unwrap();
     assert_ne!(client.server_properties().await.len(), 0);
     assert_ne!(client.connection_properties().await.len(), 0);
 }

tests/integration/consumer_test.rs

@@ -32,6 +32,7 @@ async fn consumer_test() {
     let mut consumer = env
         .env
         .consumer()
+        .client_provided_name("my consumer")
         .offset(OffsetSpecification::Next)
         .build(&env.stream)
         .await

tests/integration/producer_test.rs

@@ -53,6 +53,7 @@ async fn producer_send_name_deduplication_unique_ids() {
         let mut producer = env
             .producer()
             .name("my_producer")
+            .client_provided_name("my producer")
             .build(&stream)
             .await
             .unwrap();