Create EventHubs processor (Azure#2413)

Author: LarryOsterman

Cargo.lock

@@ -1,5 +1,20 @@
 # Release History
 
+## 0.2.0 (Unreleased)
+
+### Features Added
+
+- Added the ability to compare an `AmqpAnnotationKey` with a string and string slice.
+
+### Breaking Changes
+
+- APIs which used to return `Option<String>`, and `Option<Vec<T>>` now return `Option<&str>`, and `Option<&[T]>`.
+- APIs which take ownership of string parameters now take a `String` parameter instead of a `&str` parameter.
+
+### Bugs Fixed
+
+### Other Changes
+
 ## 0.1.0 (2025-02-18)
 
 ### Features Added

sdk/core/azure_core_amqp/CHANGELOG.md

@@ -55,8 +55,8 @@ impl AmqpDescribedError {
     pub fn condition(&self) -> &AmqpSymbol {
         &self.condition
     }
-    pub fn description(&self) -> Option<&String> {
-        self.description.as_ref()
+    pub fn description(&self) -> Option<&str> {
+        self.description.as_deref()
     }
     pub fn info(&self) -> &AmqpOrderedMap<AmqpSymbol, AmqpValue> {
         &self.info

sdk/core/azure_core_amqp/src/error.rs

@@ -80,7 +80,7 @@ impl AmqpConnectionApis for Fe2o3AmqpConnection {
             }
             if let Some(incoming_locales) = options.incoming_locales {
                 for locale in incoming_locales {
-                    builder = builder.add_incoming_locales(locale.as_str());
+                    builder = builder.add_incoming_locales(locale);
                 }
             }
             if let Some(offered_capabilities) = options.offered_capabilities.as_ref() {

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

@@ -1,6 +1,8 @@
 // Copyright (c) Microsoft Corporation. All Rights reserved
 // Licensed under the MIT license.
 
+use std::time::Duration;
+
 use crate::{
     messaging::{
         AmqpAnnotationKey, AmqpAnnotations, AmqpApplicationProperties, AmqpMessageHeader,
@@ -136,9 +138,7 @@ impl From<fe2o3_amqp_types::messaging::Header> for AmqpMessageHeader {
         AmqpMessageHeader {
             durable: header.durable,
             priority: header.priority.into(),
-            time_to_live: header
-                .ttl
-                .map(|t| std::time::Duration::from_millis(t as u64)),
+            time_to_live: header.ttl.map(|t| Duration::from_millis(t as u64)),
             first_acquirer: (header.first_acquirer),
             delivery_count: (header.delivery_count),
         }
@@ -426,6 +426,8 @@ impl From<AmqpMessageProperties> for fe2o3_amqp_types::messaging::Properties {
 
 #[test]
 fn test_properties_conversion() {
+    use std::time::{SystemTime, UNIX_EPOCH};
+
     {
         let properties = fe2o3_amqp_types::messaging::Properties {
             message_id: Some(fe2o3_amqp_types::messaging::MessageId::String(
@@ -451,14 +453,13 @@ fn test_properties_conversion() {
     }
 
     {
-        let time_now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
+        let time_now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
             .unwrap()
             .as_millis() as i64;
 
         // Round trip time_now through milliseconds to round down from nanoseconds.
-        let time_now: std::time::SystemTime =
-            std::time::UNIX_EPOCH + std::time::Duration::from_millis(time_now as u64);
+        let time_now: SystemTime = UNIX_EPOCH + Duration::from_millis(time_now as u64);
 
         let properties = AmqpMessageProperties {
             absolute_expiry_time: Some(time_now.into()),

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

@@ -428,6 +428,8 @@ impl From<AmqpMessage>
 #[cfg(test)]
 mod tests {
 
+    use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
     use super::*;
     use crate::messaging::{
         AmqpAnnotationKey, AmqpAnnotations, AmqpMessageHeader, AmqpMessageProperties,
@@ -510,14 +512,13 @@ mod tests {
         }
         // Amqp->Fe2o3
         {
-            let timestamp = std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
+            let timestamp = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
                 .unwrap()
                 .as_millis() as i64;
 
             // Round trip timestamp through milliseconds to round down from nanoseconds.
-            let timestamp: std::time::SystemTime =
-                std::time::UNIX_EPOCH + std::time::Duration::from_millis(timestamp as u64);
+            let timestamp: SystemTime = UNIX_EPOCH + Duration::from_millis(timestamp as u64);
 
             let amqp_message = AmqpMessage::builder()
                 .add_application_property("abc".to_string(), "23 skiddoo")
@@ -542,7 +543,7 @@ mod tests {
                     delivery_count: 95,
                     first_acquirer: true,
                     durable: true,
-                    time_to_live: Some(std::time::Duration::from_millis(1000)),
+                    time_to_live: Some(Duration::from_millis(1000)),
                     priority: 3,
                 })
                 .with_delivery_annotations(AmqpAnnotations::from(vec![

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

@@ -11,7 +11,7 @@ use crate::{
 };
 use serde_amqp::primitives::Timestamp;
 use serde_bytes::ByteBuf;
-use std::time::UNIX_EPOCH;
+use std::time::{Duration, UNIX_EPOCH};
 use typespec::error::ErrorKind;
 
 use super::error::Fe2o3SerializationError;
@@ -57,9 +57,7 @@ impl From<fe2o3_amqp_types::primitives::Timestamp> for AmqpTimestamp {
             return AmqpTimestamp(None);
         }
         AmqpTimestamp(
-            std::time::UNIX_EPOCH.checked_add(std::time::Duration::from_millis(
-                timestamp.milliseconds() as u64,
-            )),
+            UNIX_EPOCH.checked_add(Duration::from_millis(timestamp.milliseconds() as u64)),
         )
     }
 }

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

@@ -9,6 +9,7 @@ use crate::Deserializable;
 #[cfg(feature = "cplusplus")]
 use azure_core::error::ErrorKind;
 use azure_core::{Result, Uuid};
+use std::time::Duration;
 
 #[cfg(all(feature = "fe2o3_amqp", not(target_arch = "wasm32")))]
 type DeliveryImplementation = super::fe2o3::messaging::messaging_types::Fe2o3AmqpDelivery;
@@ -233,8 +234,8 @@ impl AmqpTarget {
         builders::AmqpTargetBuilder::new()
     }
 
-    pub fn address(&self) -> Option<&String> {
-        self.address.as_ref()
+    pub fn address(&self) -> Option<&str> {
+        self.address.as_deref()
     }
 
     pub fn durable(&self) -> Option<&TerminusDurability> {
@@ -257,8 +258,8 @@ impl AmqpTarget {
         self.dynamic_node_properties.as_ref()
     }
 
-    pub fn capabilities(&self) -> Option<&Vec<AmqpValue>> {
-        self.capabilities.as_ref()
+    pub fn capabilities(&self) -> Option<&[AmqpValue]> {
+        self.capabilities.as_deref()
     }
 }
 
@@ -577,7 +578,7 @@ impl From<AmqpSource> for AmqpList {
 pub struct AmqpMessageHeader {
     pub durable: bool,
     pub priority: u8,
-    pub time_to_live: Option<std::time::Duration>,
+    pub time_to_live: Option<Duration>,
     pub first_acquirer: bool,
     pub delivery_count: u32,
 }
@@ -980,6 +981,24 @@ impl From<u64> for AmqpAnnotationKey {
     }
 }
 
+impl PartialEq<String> for AmqpAnnotationKey {
+    fn eq(&self, other: &String) -> bool {
+        match self {
+            AmqpAnnotationKey::Symbol(symbol) => symbol.0 == *other,
+            AmqpAnnotationKey::Ulong(_) => false,
+        }
+    }
+}
+
+impl PartialEq<&str> for AmqpAnnotationKey {
+    fn eq(&self, other: &&str) -> bool {
+        match self {
+            AmqpAnnotationKey::Symbol(symbol) => symbol.0 == *other,
+            AmqpAnnotationKey::Ulong(_) => false,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Default)]
 pub struct AmqpAnnotations(pub AmqpOrderedMap<AmqpAnnotationKey, AmqpValue>);
 

sdk/core/azure_core_amqp/src/messaging.rs

@@ -182,8 +182,8 @@ mod tests {
 
         assert!(receiver_options.target.is_some());
         assert_eq!(
-            receiver_options.target.unwrap().address().unwrap().clone(),
-            "test_address".to_string()
+            receiver_options.target.unwrap().address().unwrap(),
+            "test_address"
         );
     }
 
@@ -249,8 +249,8 @@ mod tests {
         );
         assert!(receiver_options.target.is_some());
         assert_eq!(
-            receiver_options.target.unwrap().address().unwrap().clone(),
-            "combo_address".to_string()
+            receiver_options.target.unwrap().address().unwrap(),
+            "combo_address"
         );
         assert_eq!(receiver_options.name.unwrap(), "combo_name".to_string());
         assert!(receiver_options.properties.is_some());

sdk/core/azure_core_amqp/src/receiver.rs

@@ -43,12 +43,12 @@ impl AmqpSessionOptions {
         self.handle_max
     }
 
-    pub fn offered_capabilities(&self) -> Option<&Vec<AmqpSymbol>> {
-        self.offered_capabilities.as_ref()
+    pub fn offered_capabilities(&self) -> Option<&[AmqpSymbol]> {
+        self.offered_capabilities.as_deref()
     }
 
-    pub fn desired_capabilities(&self) -> Option<&Vec<AmqpSymbol>> {
-        self.desired_capabilities.as_ref()
+    pub fn desired_capabilities(&self) -> Option<&[AmqpSymbol]> {
+        self.desired_capabilities.as_deref()
     }
 
     pub fn properties(&self) -> Option<&AmqpOrderedMap<AmqpSymbol, AmqpValue>> {