Deserialize ClientOption with serde (#271)

* Move TlsConfiguration under client::options
* Rework TlsConfiguration
* ClientOption is deserializable
* TlsConfigurationBuilder enables tls on every method
* Add example. Implement ByteCapacity deserialization
* Remove serde from default
* Simplify ByteCapacity deserialization

Author: allevo

Cargo.toml

@@ -10,7 +10,8 @@ keywords = ["AMQP", "IoT", "messaging", "streams"]
 categories = ["network-programming"]
 readme = "README.md"
 
-
+[package.metadata."docs.rs"]
+all-features = true
 
 [workspace]
 members = [
@@ -37,11 +38,17 @@ async-trait = "0.1.51"
 rand = "0.8"
 dashmap = "6.1.0"
 murmur3 = "0.5.2"
+serde = { version = "1.0", features = ["derive"], optional = true }
 
 [dev-dependencies]
 tracing-subscriber = "0.3.1"
 fake = { version = "3.0.0", features = ['derive'] }
 chrono = "0.4.26"
+serde_json = "1.0"
+
+[features]
+default = []
+serde = ["dep:serde"]
 
 [[example]]
 name="receive_super_stream"
@@ -55,3 +62,7 @@ path="examples/superstreams/send_super_stream_routing_key.rs"
 [[example]]
 name="send_super_stream"
 path="examples/superstreams/send_super_stream.rs"
+[[example]]
+name="environment_deserialization"
+path="examples/environment_deserialization.rs"
+

Makefile

@@ -5,13 +5,14 @@ fmt:
 	cargo fmt --all -- --check
 
 clippy:
-	cargo clippy --all -- -D warnings
+	cargo clippy --all --all-features -- -D warnings
 
 build: fmt clippy
+	cargo build --all-features
 	cargo build
 
 test: build
-	cargo test --all -- --nocapture
+	cargo test --all --all-features -- --nocapture
 
 watch: build
 	cargo watch -x 'test --all -- --nocapture'

examples/environment_deserialization.rs

@@ -0,0 +1,116 @@
+#[cfg(feature = "serde")]
+mod example {
+    use tracing::{info, Level};
+    use tracing_subscriber::FmtSubscriber;
+
+    use futures::StreamExt;
+    use rabbitmq_stream_client::{
+        types::{ByteCapacity, Message, OffsetSpecification},
+        ClientOptions, Environment,
+    };
+    use serde::*;
+
+    #[derive(Deserialize)]
+    struct QueueConfig {
+        #[serde(flatten)]
+        rabbitmq: ClientOptions,
+        stream_name: String,
+        capabity: ByteCapacity,
+        producer_client_name: String,
+        consumer_client_name: String,
+    }
+
+    #[derive(Deserialize)]
+    struct MyConfig {
+        queue_config: QueueConfig,
+    }
+
+    pub async fn run() -> Result<(), Box<dyn std::error::Error>> {
+        let subscriber = FmtSubscriber::builder()
+            .with_max_level(Level::TRACE)
+            .finish();
+
+        tracing::subscriber::set_global_default(subscriber)
+            .expect("setting default subscriber failed");
+
+        // The configuration is loadable from a file.
+        // Here, just for semplification, we use a JSON string.
+        // NB: we use `serde` internally, so you can use any format supported by serde.
+        let j = r#"
+{
+    "queue_config": {
+        "host": "localhost",
+        "tls": {
+            "enabled": false
+        },
+        "stream_name": "test",
+        "capabity": "5GB",
+        "producer_client_name": "producer",
+        "consumer_client_name": "consumer"
+    }
+}
+        "#;
+        let my_config: MyConfig = serde_json::from_str(j).unwrap();
+        let environment = Environment::from_client_option(my_config.queue_config.rabbitmq).await?;
+
+        let message_count = 10;
+        environment
+            .stream_creator()
+            .max_length(my_config.queue_config.capabity)
+            .create(&my_config.queue_config.stream_name)
+            .await?;
+
+        let producer = environment
+            .producer()
+            .client_provided_name(&my_config.queue_config.producer_client_name)
+            .build(&my_config.queue_config.stream_name)
+            .await?;
+
+        for i in 0..message_count {
+            producer
+                .send_with_confirm(Message::builder().body(format!("message{}", i)).build())
+                .await?;
+        }
+
+        producer.close().await?;
+
+        let mut consumer = environment
+            .consumer()
+            .client_provided_name(&my_config.queue_config.consumer_client_name)
+            .offset(OffsetSpecification::First)
+            .build(&my_config.queue_config.stream_name)
+            .await
+            .unwrap();
+
+        for _ in 0..message_count {
+            let delivery = consumer.next().await.unwrap()?;
+            info!(
+                "Got message : {:?} with offset {}",
+                delivery
+                    .message()
+                    .data()
+                    .map(|data| String::from_utf8(data.to_vec())),
+                delivery.offset()
+            );
+        }
+
+        consumer.handle().close().await.unwrap();
+
+        environment.delete_stream("test").await?;
+
+        Ok(())
+    }
+}
+#[cfg(not(feature = "serde"))]
+mod example {
+    pub async fn run() -> Result<(), Box<dyn std::error::Error>> {
+        panic!("This example requires the 'serde' feature to be enabled. Add `--features serde` to the cargo command.");
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    example::run().await?;
+
+    Ok(())
+}

examples/simple-consumer.rs

@@ -33,8 +33,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await
         .unwrap();
 
-    while let delivery = consumer.next().await.unwrap() {
-        let delivery = delivery.unwrap();
+    while let Some(Ok(delivery)) = consumer.next().await {
         println!(
             "Got message: {:#?} from stream: {} with offset: {}",
             delivery

examples/tls_producer.rs

@@ -17,7 +17,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // it will start a rabbitmq server with compatible TLS certificates
     let tls_configuration: TlsConfiguration = TlsConfiguration::builder()
         .add_root_certificates(String::from(".ci/certs/ca_certificate.pem"))
-        .build();
+        .build()?;
 
     // Use this configuration if you want to trust the certificates
     // without providing the root certificate and the client certificates

src/byte_capacity.rs

@@ -22,3 +22,96 @@ impl ByteCapacity {
         }
     }
 }
+
+#[cfg(feature = "serde")]
+impl<'de> serde::Deserialize<'de> for ByteCapacity {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        use serde::Deserialize;
+
+        #[derive(Deserialize)]
+        #[serde(untagged)]
+        enum StringOrNumber {
+            String(String),
+            Number(u64),
+        }
+
+        macro_rules! match_suffix {
+            ($str: ident, $suf: expr, $variant: expr) => {
+                if $str.ends_with($suf) {
+                    let num = $str
+                        .trim_end_matches($suf)
+                        .parse()
+                        .map_err(serde::de::Error::custom)?;
+                    return Ok(($variant)(num));
+                }
+            };
+        }
+
+        let s_or_n = StringOrNumber::deserialize(deserializer)?;
+
+        match s_or_n {
+            StringOrNumber::String(str) => {
+                match_suffix!(str, "TB", ByteCapacity::TB);
+                match_suffix!(str, "GB", ByteCapacity::GB);
+                match_suffix!(str, "MB", ByteCapacity::MB);
+                match_suffix!(str, "KB", ByteCapacity::KB);
+                match_suffix!(str, "B", ByteCapacity::B);
+
+                let num = str.parse().map_err(|_| {
+                    serde::de::Error::custom(
+                        "Expect a number or a string with a TB|GB|MB|KB|B suffix",
+                    )
+                })?;
+                Ok(ByteCapacity::B(num))
+            }
+            StringOrNumber::Number(num) => Ok(ByteCapacity::B(num)),
+        }
+    }
+}
+
+#[cfg(feature = "serde")]
+mod tests {
+    #[test]
+    fn test_deserilize_byte_capacity() {
+        use crate::types::ByteCapacity;
+
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("\"5GB\""),
+            Ok(ByteCapacity::GB(5))
+        ));
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("\"5TB\""),
+            Ok(ByteCapacity::TB(5))
+        ));
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("\"5MB\""),
+            Ok(ByteCapacity::MB(5))
+        ));
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("\"5KB\""),
+            Ok(ByteCapacity::KB(5))
+        ));
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("\"5B\""),
+            Ok(ByteCapacity::B(5))
+        ));
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("\"5\""),
+            Ok(ByteCapacity::B(5))
+        ));
+        assert!(matches!(
+            serde_json::from_str::<ByteCapacity>("5"),
+            Ok(ByteCapacity::B(5))
+        ));
+        let err = serde_json::from_str::<ByteCapacity>("\"Wrong string format\"")
+            .err()
+            .expect("Expect an error");
+        assert_eq!(
+            err.to_string(),
+            "Expect a number or a string with a TB|GB|MB|KB|B suffix"
+        );
+    }
+}