Updated AMQP builders to match Rust SDK guidelines (Azure#1857)

* Updated AMQP builders to match Rust SDK guidelines

* PR feedback

Author: LarryOsterman

sdk/core/azure_core_amqp/src/fe2o3/messaging/message_fields.rs

@@ -340,43 +340,54 @@ impl From<fe2o3_amqp_types::messaging::Properties> for AmqpMessageProperties {
         let mut amqp_message_properties_builder = AmqpMessageProperties::builder();
 
         if let Some(message_id) = properties.message_id {
-            amqp_message_properties_builder.with_message_id(message_id);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_message_id(message_id);
         }
         if let Some(user_id) = properties.user_id {
-            amqp_message_properties_builder.with_user_id(user_id.to_vec());
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_user_id(user_id.to_vec());
         }
         if let Some(to) = properties.to {
-            amqp_message_properties_builder.with_to(to);
+            amqp_message_properties_builder = amqp_message_properties_builder.with_to(to);
         }
         if let Some(subject) = properties.subject {
-            amqp_message_properties_builder.with_subject(subject);
+            amqp_message_properties_builder = amqp_message_properties_builder.with_subject(subject);
         }
         if let Some(reply_to) = properties.reply_to {
-            amqp_message_properties_builder.with_reply_to(reply_to);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_reply_to(reply_to);
         }
         if let Some(correlation_id) = properties.correlation_id {
-            amqp_message_properties_builder.with_correlation_id(correlation_id);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_correlation_id(correlation_id);
         }
         if let Some(content_type) = properties.content_type {
-            amqp_message_properties_builder.with_content_type(content_type);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_content_type(content_type);
         }
         if let Some(content_encoding) = properties.content_encoding {
-            amqp_message_properties_builder.with_content_encoding(content_encoding);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_content_encoding(content_encoding);
         }
         if let Some(absolute_expiry_time) = properties.absolute_expiry_time {
-            amqp_message_properties_builder.with_absolute_expiry_time(absolute_expiry_time);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_absolute_expiry_time(absolute_expiry_time);
         }
         if let Some(creation_time) = properties.creation_time {
-            amqp_message_properties_builder.with_creation_time(creation_time);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_creation_time(creation_time);
         }
         if let Some(group_id) = properties.group_id {
-            amqp_message_properties_builder.with_group_id(group_id);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_group_id(group_id);
         }
         if let Some(group_sequence) = properties.group_sequence {
-            amqp_message_properties_builder.with_group_sequence(group_sequence);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_group_sequence(group_sequence);
         }
         if let Some(reply_to_group_id) = properties.reply_to_group_id {
-            amqp_message_properties_builder.with_reply_to_group_id(reply_to_group_id);
+            amqp_message_properties_builder =
+                amqp_message_properties_builder.with_reply_to_group_id(reply_to_group_id);
         }
         amqp_message_properties_builder.build()
     }

sdk/core/azure_core_amqp/src/fe2o3/messaging/message_source.rs

@@ -77,9 +77,9 @@ impl From<fe2o3_amqp_types::messaging::Source> for AmqpSource {
         let mut amqp_source_builder = AmqpSource::builder();
 
         if let Some(address) = source.address {
-            amqp_source_builder.with_address(address);
+            amqp_source_builder = amqp_source_builder.with_address(address);
         }
-        amqp_source_builder
+        amqp_source_builder = amqp_source_builder
             .with_durable(source.durable.into())
             .with_expiry_policy(source.expiry_policy.into())
             .with_timeout(source.timeout)
@@ -92,26 +92,29 @@ impl From<fe2o3_amqp_types::messaging::Source> for AmqpSource {
                     .map(|(k, v)| (k.into(), v.into()))
                     .collect();
 
-            amqp_source_builder.with_dynamic_node_properties(dynamic_node_properties);
+            amqp_source_builder =
+                amqp_source_builder.with_dynamic_node_properties(dynamic_node_properties);
         }
         if let Some(distribution_mode) = source.distribution_mode {
-            amqp_source_builder.with_distribution_mode(distribution_mode.into());
+            amqp_source_builder =
+                amqp_source_builder.with_distribution_mode(distribution_mode.into());
         }
         if let Some(filter) = source.filter {
             let filter: AmqpOrderedMap<AmqpSymbol, AmqpValue> = filter
                 .into_iter()
                 .map(|(k, v)| (k.into(), v.into()))
                 .collect();
-            amqp_source_builder.with_filter(filter);
+            amqp_source_builder = amqp_source_builder.with_filter(filter);
         }
         if let Some(default_outcome) = source.default_outcome {
-            amqp_source_builder.with_default_outcome(default_outcome.into());
+            amqp_source_builder = amqp_source_builder.with_default_outcome(default_outcome.into());
         }
         if let Some(outcomes) = source.outcomes {
-            amqp_source_builder.with_outcomes(outcomes.into_iter().map(|o| o.into()).collect());
+            amqp_source_builder =
+                amqp_source_builder.with_outcomes(outcomes.into_iter().map(|o| o.into()).collect());
         }
         if let Some(capabilities) = source.capabilities {
-            amqp_source_builder
+            amqp_source_builder = amqp_source_builder
                 .with_capabilities(capabilities.into_iter().map(|c| c.into()).collect());
         }
         amqp_source_builder.build()

sdk/core/azure_core_amqp/src/fe2o3/messaging/mod.rs

@@ -77,13 +77,14 @@ where
     let mut amqp_message_builder = AmqpMessage::builder();
 
     if let Some(application_properties) = message.application_properties {
-        amqp_message_builder.with_application_properties(application_properties.into());
+        amqp_message_builder =
+            amqp_message_builder.with_application_properties(application_properties.into());
     }
 
     let body = message.body;
     if body.is_empty() {
         let body = AmqpMessageBody::Empty;
-        amqp_message_builder.with_body(body);
+        amqp_message_builder = amqp_message_builder.with_body(body);
     } else if body.is_data() {
         let data = body.try_into_data().map_err(|_| {
             Error::message(
@@ -92,7 +93,7 @@ where
             )
         })?;
         let body = AmqpMessageBody::Binary(data.map(|x| x.to_vec()).collect());
-        amqp_message_builder.with_body(body);
+        amqp_message_builder = amqp_message_builder.with_body(body);
     } else if body.is_value() {
         let value = body.try_into_value().map_err(|_| {
             Error::message(
@@ -103,7 +104,7 @@ where
         // Because a conversion exists between fe2o3 values and AmqpValue types,
         // this try_into will always succeed.
         let value = value.try_into().unwrap();
-        amqp_message_builder.with_body(AmqpMessageBody::Value(value));
+        amqp_message_builder = amqp_message_builder.with_body(AmqpMessageBody::Value(value));
     } else if body.is_sequence() {
         let sequence = body.try_into_sequence().map_err(|_| {
             Error::message(
@@ -125,27 +126,29 @@ where
                 })
                 .collect(),
         );
-        amqp_message_builder.with_body(body);
+        amqp_message_builder = amqp_message_builder.with_body(body);
     }
 
     if let Some(header) = message.header {
-        amqp_message_builder.with_header(header.into());
+        amqp_message_builder = amqp_message_builder.with_header(header.into());
     }
 
     if let Some(properties) = message.properties {
-        amqp_message_builder.with_properties(properties);
+        amqp_message_builder = amqp_message_builder.with_properties(properties);
     }
 
     if let Some(delivery_annotations) = message.delivery_annotations {
-        amqp_message_builder.with_delivery_annotations(delivery_annotations.0.into());
+        amqp_message_builder =
+            amqp_message_builder.with_delivery_annotations(delivery_annotations.0.into());
     }
 
     if let Some(message_annotations) = message.message_annotations {
-        amqp_message_builder.with_message_annotations(message_annotations.0.into());
+        amqp_message_builder =
+            amqp_message_builder.with_message_annotations(message_annotations.0.into());
     }
 
     if let Some(footer) = message.footer {
-        amqp_message_builder.with_footer(footer.0.into());
+        amqp_message_builder = amqp_message_builder.with_footer(footer.0.into());
     }
 
     Ok(amqp_message_builder.build())
@@ -211,114 +214,134 @@ impl
     ) -> Self {
         let mut amqp_message_builder = AmqpMessage::builder();
 
-        amqp_message_builder.with_body(AmqpMessageBody::Empty);
+        amqp_message_builder = amqp_message_builder.with_body(AmqpMessageBody::Empty);
 
         if let Some(application_properties) = message.application_properties {
-            amqp_message_builder.with_application_properties(application_properties.into());
+            amqp_message_builder =
+                amqp_message_builder.with_application_properties(application_properties.into());
         }
 
         if let Some(header) = message.header {
-            amqp_message_builder.with_header(header.into());
+            amqp_message_builder = amqp_message_builder.with_header(header.into());
         }
 
         if let Some(properties) = message.properties {
             info!("Converting properties to AmqpMessageProperties");
-            amqp_message_builder.with_properties(properties);
+            amqp_message_builder = amqp_message_builder.with_properties(properties);
         }
 
         if let Some(delivery_annotations) = message.delivery_annotations {
-            amqp_message_builder.with_delivery_annotations(delivery_annotations.0.into());
+            amqp_message_builder =
+                amqp_message_builder.with_delivery_annotations(delivery_annotations.0.into());
         }
 
         if let Some(message_annotations) = message.message_annotations {
-            amqp_message_builder.with_message_annotations(message_annotations.0.into());
+            amqp_message_builder =
+                amqp_message_builder.with_message_annotations(message_annotations.0.into());
         }
 
         if let Some(footer) = message.footer {
-            amqp_message_builder.with_footer(footer.0.into());
+            amqp_message_builder = amqp_message_builder.with_footer(footer.0.into());
         }
 
         amqp_message_builder.build()
     }
 }
 
+fn fe2o3_message_from_amqp_message(
+    message: &AmqpMessage,
+) -> fe2o3_amqp_types::messaging::Message<
+    fe2o3_amqp_types::messaging::Body<fe2o3_amqp_types::primitives::Value>,
+> {
+    let message_builder = fe2o3_amqp_types::messaging::Message::builder()
+        .application_properties(message.application_properties().map(|x| x.clone().into()))
+        .properties(message.properties().map(|p| p.clone().into()))
+        .header(message.header().map(|x| x.clone().into()))
+        .delivery_annotations(message.delivery_annotations().map(|x| x.clone().into()))
+        .message_annotations(message.message_annotations().map(|x| x.clone().into()))
+        .footer(message.footer().map(|x| x.clone().into()));
+
+    match message.body() {
+        AmqpMessageBody::Value(value) => {
+            let value: fe2o3_amqp_types::primitives::Value = value.clone().into();
+            let value = fe2o3_amqp_types::messaging::Body::Value(value.into_body());
+            let message_builder = message_builder.body(value);
+            message_builder.build()
+        }
+        AmqpMessageBody::Binary(data) => {
+            let data: Vec<serde_bytes::ByteBuf> = data
+                .clone()
+                .into_iter()
+                .map(serde_bytes::ByteBuf::from)
+                .collect();
+            let message_builder = message_builder.body(fe2o3_amqp_types::messaging::Body::Data(
+                data.into_iter().map(|x| x.into()).collect(),
+            ));
+            message_builder.build()
+        }
+        AmqpMessageBody::Empty => message_builder
+            .body(fe2o3_amqp_types::messaging::Body::Empty)
+            .build(),
+        AmqpMessageBody::Sequence(sequence) => {
+            let sequence: TransparentVec<
+                fe2o3_amqp_types::primitives::List<fe2o3_amqp_types::primitives::Value>,
+            > = sequence
+                .iter()
+                .map(|x| {
+                    let mut l = fe2o3_amqp_types::primitives::List::new();
+                    let c = x
+                        .clone()
+                        .0
+                        .into_iter()
+                        .map(|v| Into::<fe2o3_amqp_types::primitives::Value>::into(v.clone()));
+                    for v in c {
+                        l.push(v);
+                    }
+                    l
+                })
+                .collect();
+            let amqp_sequence = TransparentVec::<
+                fe2o3_amqp_types::messaging::AmqpSequence<fe2o3_amqp_types::primitives::Value>,
+            >::new(
+                sequence
+                    .into_iter()
+                    .map(|x| {
+                        x.into_iter()
+                        .collect::<fe2o3_amqp_types::primitives::List<
+                            fe2o3_amqp_types::primitives::Value,
+                        >>()
+                        .into()
+                    })
+                    .collect::<Vec<
+                        fe2o3_amqp_types::messaging::AmqpSequence<
+                            fe2o3_amqp_types::primitives::Value,
+                        >,
+                    >>(),
+            );
+
+            let message_builder =
+                message_builder.body(fe2o3_amqp_types::messaging::Body::Sequence(amqp_sequence));
+            message_builder.build()
+        }
+    }
+}
+
 impl From<AmqpMessage>
     for fe2o3_amqp_types::messaging::Message<
         fe2o3_amqp_types::messaging::Body<fe2o3_amqp_types::primitives::Value>,
     >
 {
     fn from(message: AmqpMessage) -> Self {
-        let message_builder = fe2o3_amqp_types::messaging::Message::builder()
-            .application_properties(message.application_properties().map(|x| x.clone().into()))
-            .properties(message.properties().map(|p| p.clone().into()))
-            .header(message.header().map(|x| x.clone().into()))
-            .delivery_annotations(message.delivery_annotations().map(|x| x.clone().into()))
-            .message_annotations(message.message_annotations().map(|x| x.clone().into()))
-            .footer(message.footer().map(|x| x.clone().into()));
-
-        match message.body() {
-            AmqpMessageBody::Value(value) => {
-                let value: fe2o3_amqp_types::primitives::Value = value.clone().into();
-                let value = fe2o3_amqp_types::messaging::Body::Value(value.into_body());
-                let message_builder = message_builder.body(value);
-                message_builder.build()
-            }
-            AmqpMessageBody::Binary(data) => {
-                let data: Vec<serde_bytes::ByteBuf> = data
-                    .clone()
-                    .into_iter()
-                    .map(serde_bytes::ByteBuf::from)
-                    .collect();
-                let message_builder =
-                    message_builder.body(fe2o3_amqp_types::messaging::Body::Data(
-                        data.into_iter().map(|x| x.into()).collect(),
-                    ));
-                message_builder.build()
-            }
-            AmqpMessageBody::Empty => message_builder
-                .body(fe2o3_amqp_types::messaging::Body::Empty)
-                .build(),
-            AmqpMessageBody::Sequence(sequence) => {
-                let sequence: TransparentVec<
-                    fe2o3_amqp_types::primitives::List<fe2o3_amqp_types::primitives::Value>,
-                > = sequence
-                    .iter()
-                    .map(|x| {
-                        let mut l = fe2o3_amqp_types::primitives::List::new();
-                        let c =
-                            x.clone().0.into_iter().map(|v| {
-                                Into::<fe2o3_amqp_types::primitives::Value>::into(v.clone())
-                            });
-                        for v in c {
-                            l.push(v);
-                        }
-                        l
-                    })
-                    .collect();
-                let amqp_sequence = TransparentVec::<
-                    fe2o3_amqp_types::messaging::AmqpSequence<fe2o3_amqp_types::primitives::Value>,
-                >::new(
-                    sequence
-                        .into_iter()
-                        .map(|x| {
-                            x.into_iter()
-                                .collect::<fe2o3_amqp_types::primitives::List<
-                                    fe2o3_amqp_types::primitives::Value,
-                                >>()
-                                .into()
-                        })
-                        .collect::<Vec<
-                            fe2o3_amqp_types::messaging::AmqpSequence<
-                                fe2o3_amqp_types::primitives::Value,
-                            >,
-                        >>(),
-                );
-
-                let message_builder = message_builder
-                    .body(fe2o3_amqp_types::messaging::Body::Sequence(amqp_sequence));
-                message_builder.build()
-            }
-        }
+        fe2o3_message_from_amqp_message(&message)
+    }
+}
+impl From<&AmqpMessage>
+    for fe2o3_amqp_types::messaging::Message<
+        fe2o3_amqp_types::messaging::Body<fe2o3_amqp_types::primitives::Value>,
+    >
+{
+    fn from(message: &AmqpMessage) -> Self {
+        fe2o3_message_from_amqp_message(message)
     }
 }