Enable Custom Endpoints for EventHubs (#2245)

* Added ability to use AMQP proxy server; Refactored Eventhubs to consolidate common infrastructure


* Removed uuid and url direct dependencies; fixed up AMQP cargo.toml file to properly declare the cplusplus feature

Author: LarryOsterman

.vscode/cspell.json

@@ -17,7 +17,7 @@
     ".vscode/tasks.json",
     "NOTICE.txt",
     "eng/",
-    "*.dict.txt",
+    "**/.dict.txt",
     "rust-toolchain.toml"
   ],
   "words": [

Cargo.lock

@@ -26,18 +26,17 @@ fe2o3-amqp-management = { workspace = true, optional = true }
 fe2o3-amqp-types = { workspace = true, optional = true }
 serde_amqp = { workspace = true, optional = true }
 serde_bytes = { workspace = true, optional = true }
-typespec = { workspace = true, features = ["amqp"] }
 time.workspace = true
 tokio.workspace = true
 tracing.workspace = true
-uuid = { workspace = true }
-url = { workspace = true }
+typespec = { workspace = true, features = ["amqp"] }
 
 [dev-dependencies]
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 
 [features]
 default = ["fe2o3-amqp"]
+cplusplus = []
 fe2o3-amqp = [
   "dep:fe2o3-amqp",
   "fe2o3-amqp-types",
@@ -47,7 +46,6 @@ fe2o3-amqp = [
   "serde_amqp",
   "serde_bytes",
 ]
-cplusplus = []
 
 [package.metadata.docs.rs]
 features = ["fe2o3-amqp"]

sdk/core/azure_core_amqp/Cargo.toml

@@ -12,17 +12,29 @@ type ConnectionImplementation = super::fe2o3::connection::Fe2o3AmqpConnection;
 #[cfg(any(not(feature = "fe2o3-amqp"), target_arch = "wasm32"))]
 type ConnectionImplementation = super::noop::NoopAmqpConnection;
 
+/// Options for configuring an AMQP connection.
 #[derive(Debug, Default, Clone)]
 pub struct AmqpConnectionOptions {
+    /// Maximum frame size for the connection in bytes.
     pub max_frame_size: Option<u32>,
+    /// Maximum number of channels for the connection.
     pub channel_max: Option<u16>,
+    /// Idle timeout for the connection.
     pub idle_timeout: Option<Duration>,
+    /// List of outgoing locales for the connection.
     pub outgoing_locales: Option<Vec<String>>,
+    /// List of incoming locales for the connection.
     pub incoming_locales: Option<Vec<String>>,
+    /// List of offered capabilities for the connection.
     pub offered_capabilities: Option<Vec<AmqpSymbol>>,
+    /// List of desired capabilities for the connection.
     pub desired_capabilities: Option<Vec<AmqpSymbol>>,
+    /// Properties for the connection.
     pub properties: Option<AmqpOrderedMap<AmqpSymbol, AmqpValue>>,
+    /// Buffer size for the connection.
     pub buffer_size: Option<usize>,
+    /// Custom endpoint for the connection. Used to connect to a local AMQP proxy server.
+    pub custom_endpoint: Option<Url>,
 }
 
 impl AmqpConnectionOptions {}
@@ -205,6 +217,7 @@ mod tests {
             incoming_locales: Some(vec!["en-US".to_string()]),
             offered_capabilities: Some(vec!["capability".into()]),
             desired_capabilities: Some(vec!["capability".into()]),
+            custom_endpoint: Some(Url::parse("http://localhost:8080").unwrap()),
             properties: Some(
                 vec![("key", "value")]
                     .into_iter()
@@ -242,6 +255,10 @@ mod tests {
                 vec![("key".into(), "value".into())].into_iter().collect() // convert to AmqpOrderedMap
             )
         );
+        assert_eq!(
+            connection_options.custom_endpoint,
+            Some(Url::parse("http://localhost:8080").unwrap())
+        );
     }
 
     #[tokio::test]

sdk/core/azure_core_amqp/src/connection.rs

@@ -71,7 +71,7 @@ impl AmqpClaimsBasedSecurityApis for Fe2o3ClaimsBasedSecurity<'_> {
     }
 
     async fn detach(mut self) -> Result<()> {
-        let cbs = self.cbs.take().ok_or(Self::cbs_not_set())?;
+        let cbs = self.cbs.take().ok_or_else(Self::cbs_not_set)?;
         let cbs = cbs.into_inner();
         cbs.close().await.map_err(AmqpError::from)?;
         Ok(())

sdk/core/azure_core_amqp/src/fe2o3/cbs.rs

@@ -53,64 +53,71 @@ impl AmqpConnectionApis for Fe2o3AmqpConnection {
         options: Option<AmqpConnectionOptions>,
     ) -> Result<()> {
         {
+            let options = options.unwrap_or_default();
+            let mut endpoint = url.clone();
+
             // All AMQP clients have a similar set of options.
             let mut builder = fe2o3_amqp::Connection::builder()
                 .sasl_profile(fe2o3_amqp::sasl_profile::SaslProfile::Anonymous)
                 .alt_tls_establishment(true)
                 .container_id(id)
                 .max_frame_size(65536);
 
-            if let Some(options) = options {
-                if let Some(frame_size) = options.max_frame_size {
-                    builder = builder.max_frame_size(frame_size);
-                }
+            if let Some(frame_size) = options.max_frame_size {
+                builder = builder.max_frame_size(frame_size);
+            }
 
-                if let Some(channel_max) = options.channel_max {
-                    builder = builder.channel_max(channel_max);
-                }
-                if let Some(idle_timeout) = options.idle_timeout {
-                    builder = builder.idle_time_out(idle_timeout.whole_milliseconds() as u32);
-                }
-                if let Some(outgoing_locales) = options.outgoing_locales.as_ref() {
-                    for locale in outgoing_locales {
-                        builder = builder.add_outgoing_locales(locale.as_str());
-                    }
-                }
-                if let Some(incoming_locales) = options.incoming_locales {
-                    for locale in incoming_locales {
-                        builder = builder.add_incoming_locales(locale.as_str());
-                    }
+            if let Some(channel_max) = options.channel_max {
+                builder = builder.channel_max(channel_max);
+            }
+            if let Some(idle_timeout) = options.idle_timeout {
+                builder = builder.idle_time_out(idle_timeout.whole_milliseconds() as u32);
+            }
+            if let Some(outgoing_locales) = options.outgoing_locales.as_ref() {
+                for locale in outgoing_locales {
+                    builder = builder.add_outgoing_locales(locale.as_str());
                 }
-                if let Some(offered_capabilities) = options.offered_capabilities.as_ref() {
-                    for capability in offered_capabilities {
-                        let capability: fe2o3_amqp_types::primitives::Symbol =
-                            capability.clone().into();
-                        builder = builder.add_offered_capabilities(capability);
-                    }
+            }
+            if let Some(incoming_locales) = options.incoming_locales {
+                for locale in incoming_locales {
+                    builder = builder.add_incoming_locales(locale.as_str());
                 }
-                if let Some(desired_capabilities) = options.desired_capabilities.as_ref() {
-                    for capability in desired_capabilities {
-                        let capability: fe2o3_amqp_types::primitives::Symbol =
-                            capability.clone().into();
-                        builder = builder.add_desired_capabilities(capability);
-                    }
+            }
+            if let Some(offered_capabilities) = options.offered_capabilities.as_ref() {
+                for capability in offered_capabilities {
+                    let capability: fe2o3_amqp_types::primitives::Symbol =
+                        capability.clone().into();
+                    builder = builder.add_offered_capabilities(capability);
                 }
-                if let Some(properties) = options.properties.as_ref() {
-                    let mut fields = fe2o3_amqp::types::definitions::Fields::new();
-                    for property in properties.iter() {
-                        let k = fe2o3_amqp_types::primitives::Symbol::from(property.0);
-                        let v = fe2o3_amqp_types::primitives::Value::from(property.1);
-
-                        fields.insert(k, v);
-                    }
-                    builder = builder.properties(fields);
+            }
+            if let Some(desired_capabilities) = options.desired_capabilities.as_ref() {
+                for capability in desired_capabilities {
+                    let capability: fe2o3_amqp_types::primitives::Symbol =
+                        capability.clone().into();
+                    builder = builder.add_desired_capabilities(capability);
                 }
-                if let Some(buffer_size) = options.buffer_size {
-                    builder = builder.buffer_size(buffer_size);
+            }
+            if let Some(properties) = options.properties.as_ref() {
+                let mut fields = fe2o3_amqp::types::definitions::Fields::new();
+                for property in properties.iter() {
+                    let k = fe2o3_amqp_types::primitives::Symbol::from(property.0);
+                    let v = fe2o3_amqp_types::primitives::Value::from(property.1);
+
+                    fields.insert(k, v);
                 }
+                builder = builder.properties(fields);
             }
+            if let Some(buffer_size) = options.buffer_size {
+                builder = builder.buffer_size(buffer_size);
+            }
+
+            if let Some(custom_endpoint) = options.custom_endpoint {
+                endpoint = custom_endpoint;
+                builder = builder.hostname(url.host_str());
+            }
+
             self.connection
-                .set(Mutex::new(builder.open(url).await.map_err(|e| {
+                .set(Mutex::new(builder.open(endpoint).await.map_err(|e| {
                     azure_core::Error::from(Fe2o3ConnectionOpenError(e))
                 })?))
                 .map_err(|_| Self::connection_already_set())?;
@@ -142,7 +149,7 @@ impl AmqpConnectionApis for Fe2o3AmqpConnection {
         let mut connection = self
             .connection
             .get()
-            .ok_or(Self::connection_not_set())?
+            .ok_or_else(Self::connection_not_set)?
             .lock()
             .await;
         let res = connection

sdk/core/azure_core_amqp/src/fe2o3/connection.rs

@@ -82,7 +82,7 @@ impl AmqpManagementApis for Fe2o3AmqpManagement {
         let management = self
             .management
             .take()
-            .ok_or(Self::amqp_management_not_attached())?;
+            .ok_or_else(Self::amqp_management_not_attached)?;
         let management = management.into_inner();
         management.close().await.map_err(AmqpError::from)?;
         Ok(())
@@ -96,7 +96,7 @@ impl AmqpManagementApis for Fe2o3AmqpManagement {
         let mut management = self
             .management
             .get()
-            .ok_or(Self::amqp_management_not_attached())?
+            .ok_or_else(Self::amqp_management_not_attached)?
             .lock()
             .await;
 

sdk/core/azure_core_amqp/src/fe2o3/management.rs

@@ -49,8 +49,7 @@ impl From<AmqpMessageId> for fe2o3_amqp_types::messaging::MessageId {
 
 #[test]
 fn test_message_id_conversion() {
-    use crate::Uuid;
-
+    use azure_core::Uuid;
     {
         let message_id = fe2o3_amqp_types::messaging::MessageId::String("test".into());
         let amqp_message_id: AmqpMessageId = message_id.clone().into();

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

@@ -78,7 +78,7 @@ impl AmqpReceiverApis for Fe2o3AmqpReceiver {
     }
 
     async fn detach(mut self) -> Result<()> {
-        let receiver = self.receiver.take().ok_or(Self::receiver_not_set())?;
+        let receiver = self.receiver.take().ok_or_else(Self::receiver_not_set)?;
         let res = receiver
             .into_inner()
             .detach()
@@ -100,21 +100,21 @@ impl AmqpReceiverApis for Fe2o3AmqpReceiver {
     }
 
     async fn set_credit_mode(&self, credit_mode: ReceiverCreditMode) -> Result<()> {
-        let receiver = self.receiver.get().ok_or(Self::receiver_not_set())?;
+        let receiver = self.receiver.get().ok_or_else(Self::receiver_not_set)?;
         receiver.lock().await.set_credit_mode(credit_mode.into());
         Ok(())
     }
 
     async fn credit_mode(&self) -> Result<ReceiverCreditMode> {
-        let receiver = self.receiver.get().ok_or(Self::receiver_not_set())?;
+        let receiver = self.receiver.get().ok_or_else(Self::receiver_not_set)?;
         Ok(receiver.lock().await.credit_mode().into())
     }
 
     async fn receive_delivery(&self) -> Result<AmqpDelivery> {
         let mut receiver = self
             .receiver
             .get()
-            .ok_or(Self::receiver_not_set())?
+            .ok_or_else(Self::receiver_not_set)?
             .lock()
             .await;
 
@@ -132,7 +132,7 @@ impl AmqpReceiverApis for Fe2o3AmqpReceiver {
         let receiver = self
             .receiver
             .get()
-            .ok_or(Self::receiver_not_set())?
+            .ok_or_else(Self::receiver_not_set)?
             .lock()
             .await;
 
@@ -150,7 +150,7 @@ impl AmqpReceiverApis for Fe2o3AmqpReceiver {
         let receiver = self
             .receiver
             .get()
-            .ok_or(Self::receiver_not_set())?
+            .ok_or_else(Self::receiver_not_set)?
             .lock()
             .await;
 
@@ -168,7 +168,7 @@ impl AmqpReceiverApis for Fe2o3AmqpReceiver {
         let receiver = self
             .receiver
             .get()
-            .ok_or(Self::receiver_not_set())?
+            .ok_or_else(Self::receiver_not_set)?
             .lock()
             .await;
 

sdk/core/azure_core_amqp/src/fe2o3/receiver.rs

@@ -94,7 +94,7 @@ impl AmqpSenderApis for Fe2o3AmqpSender {
         let sender = self
             .sender
             .take()
-            .ok_or(Self::could_not_get_message_sender())?;
+            .ok_or_else(Self::could_not_get_message_sender)?;
         let res = sender
             .into_inner()
             .detach()
@@ -119,7 +119,7 @@ impl AmqpSenderApis for Fe2o3AmqpSender {
         Ok(self
             .sender
             .get()
-            .ok_or(Self::could_not_get_message_sender())?
+            .ok_or_else(Self::could_not_get_message_sender)?
             .lock()
             .await
             .max_message_size())
@@ -149,7 +149,7 @@ impl AmqpSenderApis for Fe2o3AmqpSender {
         let outcome = self
             .sender
             .get()
-            .ok_or(Self::could_not_get_message_sender())?
+            .ok_or_else(Self::could_not_get_message_sender)?
             .lock()
             .await
             .borrow_mut()