refactor(eventhubs)!: harden public API for 1.0 (non_exhaustive + TryFrom)

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md

@@ -11,6 +11,8 @@
 ### Breaking Changes
 
 - On the receive path, the `amqp:link:stolen` AMQP condition is no longer auto-retried. A receiver displaced by a higher-or-equal-epoch attacher now surfaces the error (translated to `EventHubsError::ConsumerDisconnected` by `EventReceiver::stream_events`) instead of silently re-attaching. Sender, CBS, and management operations retain the historical retry-on-stolen behavior.
+- Marked the crate's public data, options, and error types `#[non_exhaustive]` so fields and variants can be added after the 1.0 API freeze without a further major version bump. Affected types: the `MessageId`, `StartLocation`, and `ProcessorStrategy` enums; the `SendBatchOptions`, `SendEventOptions`, `SendMessageOptions`, `AddEventDataOptions`, `EventDataBatchOptions`, `OpenReceiverOptions`, `StartPosition`, and `RetryOptions` options structs; and the `EventHubProperties`, `EventHubPartitionProperties`, `Checkpoint`, `Ownership`, `StartPositions`, and `EventHubsError` data structs. External code can no longer build these with a struct literal or exhaustively match the enums; construct the structs from `Default::default()` and set the public fields, and add a wildcard arm when matching the enums.
+- Changed the fallible `MessageId` conversions from `From` to `TryFrom` to remove panics from the public API before 1.0. `Uuid`, `Vec<u8>`, `String`, and `u64` now implement `TryFrom<MessageId>` (returning `Result<_, EventHubsError>`) instead of `From<MessageId>`, which previously panicked when the `MessageId` variant did not match the target type. The infallible conversions into `MessageId` and between `MessageId` and `AmqpMessageId` are unchanged.
 
 ### Bugs Fixed
 

sdk/eventhubs/azure_messaging_eventhubs/README.md

@@ -218,12 +218,12 @@ async fn receive_events(client: &ConsumerClient) -> Result<(), Box<dyn std::erro
     let message_receiver = client
         .open_receiver_on_partition(
             "0".to_string(),
-            Some(OpenReceiverOptions {
-                start_position: Some(StartPosition {
-                    location: StartLocation::Earliest,
-                    ..Default::default()
-                }),
-                ..Default::default()
+            Some({
+                let mut start_position = StartPosition::default();
+                start_position.location = StartLocation::Earliest;
+                let mut options = OpenReceiverOptions::default();
+                options.start_position = Some(start_position);
+                options
             }),
         )
         .await?;

sdk/eventhubs/azure_messaging_eventhubs/benches/benchmarks.rs

@@ -210,8 +210,10 @@ fn send_benchmark(c: &mut Criterion) {
                 client
                     .send_event(
                         event_data,
-                        Some(SendEventOptions {
-                            partition_id: Some("0".to_string()),
+                        Some({
+                            let mut options = SendEventOptions::default();
+                            options.partition_id = Some("0".to_string());
+                            options
                         }),
                     )
                     .await

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_batch_produce_events.rs

@@ -17,9 +17,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let credential = DeveloperToolsCredential::new(None)?;
 
     let client = ProducerClient::builder()
-        .with_retry_options(RetryOptions {
-            initial_delay: Duration::milliseconds(100),
-            ..Default::default()
+        .with_retry_options({
+            let mut retry_options = RetryOptions::default();
+            retry_options.initial_delay = Duration::milliseconds(100);
+            retry_options
         })
         .open(
             eventhub_namespace.as_str(),
@@ -38,9 +39,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Send the message to each partition using a batch sender.
     for partition_id in properties.partition_ids {
         let batch = client
-            .create_batch(Some(EventDataBatchOptions {
-                partition_id: Some(partition_id.clone()),
-                ..Default::default()
+            .create_batch(Some({
+                let mut options = EventDataBatchOptions::default();
+                options.partition_id = Some(partition_id.clone());
+                options
             }))
             .await?;
         if batch.try_add_event_data(message, None)? {

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_connection_string.rs

@@ -40,8 +40,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     producer
         .send_event(
             marker.clone(),
-            Some(SendEventOptions {
-                partition_id: Some(partition_id.clone()),
+            Some({
+                let mut options = SendEventOptions::default();
+                options.partition_id = Some(partition_id.clone());
+                options
             }),
         )
         .await?;
@@ -55,13 +57,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let receiver = consumer
         .open_receiver_on_partition(
             partition_id.clone(),
-            Some(OpenReceiverOptions {
-                start_position: Some(StartPosition {
-                    location: StartLocation::SequenceNumber(start_sequence),
-                    inclusive: false,
-                }),
-                receive_timeout: Some(Duration::seconds(30)),
-                ..Default::default()
+            Some({
+                let mut start_position = StartPosition::default();
+                start_position.location = StartLocation::SequenceNumber(start_sequence);
+                start_position.inclusive = false;
+                let mut options = OpenReceiverOptions::default();
+                options.start_position = Some(start_position);
+                options.receive_timeout = Some(Duration::seconds(30));
+                options
             }),
         )
         .await?;

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_consume_events.rs

@@ -35,13 +35,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let receiver = consumer
         .open_receiver_on_partition(
             properties.partition_ids[0].clone(),
-            Some(OpenReceiverOptions {
-                start_position: Some(StartPosition {
-                    location: StartLocation::Earliest,
-                    ..Default::default()
-                }),
-                receive_timeout: Some(Duration::seconds(5)),
-                ..Default::default()
+            Some({
+                let mut start_position = StartPosition::default();
+                start_position.location = StartLocation::Earliest;
+                let mut options = OpenReceiverOptions::default();
+                options.start_position = Some(start_position);
+                options.receive_timeout = Some(Duration::seconds(5));
+                options
             }),
         )
         .await?;

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_produce_events.rs

@@ -34,8 +34,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     client
         .send_event(
             vec![2, 4, 8, 16],
-            Some(SendEventOptions {
-                partition_id: Some("0".to_string()),
+            Some({
+                let mut options = SendEventOptions::default();
+                options.partition_id = Some("0".to_string());
+                options
             }),
         )
         .await?;

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_tracing_subscriber.rs

@@ -56,8 +56,10 @@ impl EventHubsLayer {
                 if let Err(err) = producer_clone
                     .send_event(
                         event,
-                        Some(SendEventOptions {
-                            partition_id: Some("0".to_string()),
+                        Some({
+                            let mut options = SendEventOptions::default();
+                            options.partition_id = Some("0".to_string());
+                            options
                         }),
                     )
                     .await

sdk/eventhubs/azure_messaging_eventhubs/src/common/retry.rs

@@ -33,6 +33,7 @@ pub(crate) enum ErrorRecoveryAction {
 
 /// Options for configuring exponential backoff retry behavior.
 #[derive(Debug, Clone)]
+#[non_exhaustive]
 pub struct RetryOptions {
     /// The initial backoff delay (Default is 100ms).
     pub initial_delay: Duration,

sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs

@@ -428,6 +428,7 @@ impl ConsumerClient {
 
 /// Represents the options for receiving events from an Event Hub.
 #[derive(Debug, Clone, Default)]
+#[non_exhaustive]
 pub struct OpenReceiverOptions {
     /// The owner level for messages being retrieved.
     pub owner_level: Option<i64>,
@@ -447,6 +448,7 @@ impl OpenReceiverOptions {}
 
 /// Represents the starting position of a consumer when receiving events from an Event Hub.
 #[derive(Debug, Default, PartialEq, Clone)]
+#[non_exhaustive]
 pub enum StartLocation {
     /// The starting position is specified by an offset.
     Offset(String),
@@ -479,45 +481,36 @@ pub(crate) const SEQUENCE_NUMBER_ANNOTATION: &str = "amqp.annotation.x-opt-seque
 /// ```
 /// use azure_messaging_eventhubs::{StartPosition, StartLocation};
 ///
-/// let start_position = StartPosition{
-///   location: StartLocation::SequenceNumber(12345),
-///    ..Default::default()};;
+/// let mut start_position = StartPosition::default();
+/// start_position.location = StartLocation::SequenceNumber(12345);
 /// ```
 ///
 /// ```
 /// use azure_messaging_eventhubs::{StartPosition, StartLocation};
 ///
-/// let start_position = StartPosition{
-///  location: StartLocation::EnqueuedTime(std::time::SystemTime::now()),
-///  ..Default::default()
-/// };
+/// let mut start_position = StartPosition::default();
+/// start_position.location = StartLocation::EnqueuedTime(std::time::SystemTime::now());
 /// ```
 ///
 /// ```
 /// use azure_messaging_eventhubs::{StartPosition, StartLocation};
 ///
-/// let start_position = StartPosition{
-///   location: StartLocation::Offset("12345".to_string()),
-///   ..Default::default()
-/// };
+/// let mut start_position = StartPosition::default();
+/// start_position.location = StartLocation::Offset("12345".to_string());
 /// ```
 ///
 /// ```
 /// use azure_messaging_eventhubs::{StartPosition, StartLocation};
 ///
-/// let start_position = StartPosition{
-///   location: StartLocation::Earliest,
-///   ..Default::default()
-/// };
+/// let mut start_position = StartPosition::default();
+/// start_position.location = StartLocation::Earliest;
 /// ```
 ///
 /// ```
 /// use azure_messaging_eventhubs::{StartPosition, StartLocation};
 ///
-/// let start_position = StartPosition{
-///   location: StartLocation::Latest,
-///   ..Default::default()
-/// };
+/// let mut start_position = StartPosition::default();
+/// start_position.location = StartLocation::Latest;
 /// ```
 ///
 /// ```
@@ -527,6 +520,7 @@ pub(crate) const SEQUENCE_NUMBER_ANNOTATION: &str = "amqp.annotation.x-opt-seque
 /// ```
 ///
 #[derive(Debug, PartialEq, Clone, Default)]
+#[non_exhaustive]
 pub struct StartPosition {
     /// The location of the starting position.
     pub location: StartLocation,

sdk/eventhubs/azure_messaging_eventhubs/src/error.rs

@@ -37,6 +37,7 @@ pub enum ErrorKind {
 }
 
 /// Represents an error that can occur in the Event Hubs module.
+#[non_exhaustive]
 pub struct EventHubsError {
     /// The kind of error that occurred.
     pub kind: ErrorKind,

sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/mod.rs

@@ -88,6 +88,7 @@ pub trait CheckpointStore: Send + Sync {
 }
 
 #[derive(Clone, Debug, Copy)]
+#[non_exhaustive]
 /// Represents the strategy for load balancing event processing.
 ///
 /// The choice of strategy can impact the performance and efficiency

sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/models.rs

@@ -15,6 +15,7 @@ use std::collections::HashMap;
 /// for a specific partition. It helps in resuming event processing from
 /// the correct position in case of failures or restarts.
 #[derive(Debug, Default, Clone)]
+#[non_exhaustive]
 pub struct Checkpoint {
     /// The fully qualified namespace of the Event Hub.
     pub fully_qualified_namespace: String,
@@ -80,6 +81,7 @@ impl Checkpoint {
 /// by different consumers in a consumer group. It helps in load balancing
 /// and ensuring that each partition is processed by only one consumer at a time.
 #[derive(Debug, Default, Clone)]
+#[non_exhaustive]
 pub struct Ownership {
     /// The fully qualified namespace of the Event Hub.
     pub fully_qualified_namespace: String,
@@ -142,6 +144,7 @@ impl Ownership {
 /// position based on various criteria, such as the latest event, a specific
 /// offset, or a specific sequence number.
 #[derive(Debug, Default)]
+#[non_exhaustive]
 pub struct StartPositions {
     /// The starting position for each partition in the Event Hub.
     /// The key is the partition ID, and the value is the starting position.

sdk/eventhubs/azure_messaging_eventhubs/src/models/mod.rs

@@ -29,6 +29,7 @@ pub mod builders {
 /// Event sent to an Event Hub.
 pub use event_data::EventData;
 
+use crate::error::EventHubsError;
 use azure_core::Uuid;
 use azure_core_amqp::message::AmqpMessageId;
 use std::fmt::Debug;
@@ -65,6 +66,7 @@ use std::time::SystemTime;
 /// ```
 ///
 #[derive(Debug)]
+#[non_exhaustive]
 pub struct EventHubProperties {
     /// The name of the Event Hubs instance.
     pub name: String,
@@ -107,6 +109,7 @@ pub struct EventHubProperties {
 /// ```
 ///
 #[derive(Debug)]
+#[non_exhaustive]
 pub struct EventHubPartitionProperties {
     /// The unique identifier of the partition.
     pub id: String,
@@ -138,6 +141,7 @@ pub struct EventHubPartitionProperties {
 /// assured to be globally unique.
 ///
 #[derive(Debug, PartialEq, Clone)]
+#[non_exhaustive]
 pub enum MessageId {
     /// A binary representation of the message identifier.
     Binary(Vec<u8>),
@@ -182,38 +186,54 @@ impl From<String> for MessageId {
     }
 }
 
-impl From<MessageId> for Uuid {
-    fn from(message_id: MessageId) -> Self {
+impl TryFrom<MessageId> for Uuid {
+    type Error = EventHubsError;
+
+    fn try_from(message_id: MessageId) -> std::result::Result<Self, Self::Error> {
         match message_id {
-            MessageId::Uuid(uuid) => uuid,
-            _ => panic!("Cannot convert MessageId to Uuid"),
+            MessageId::Uuid(uuid) => Ok(uuid),
+            _ => Err(EventHubsError::with_message(
+                "Cannot convert MessageId to Uuid",
+            )),
         }
     }
 }
 
-impl From<MessageId> for Vec<u8> {
-    fn from(message_id: MessageId) -> Self {
+impl TryFrom<MessageId> for Vec<u8> {
+    type Error = EventHubsError;
+
+    fn try_from(message_id: MessageId) -> std::result::Result<Self, Self::Error> {
         match message_id {
-            MessageId::Binary(binary) => binary,
-            _ => panic!("Cannot convert MessageId to Vec<u8>"),
+            MessageId::Binary(binary) => Ok(binary),
+            _ => Err(EventHubsError::with_message(
+                "Cannot convert MessageId to Vec<u8>",
+            )),
         }
     }
 }
 
-impl From<MessageId> for String {
-    fn from(message_id: MessageId) -> Self {
+impl TryFrom<MessageId> for String {
+    type Error = EventHubsError;
+
+    fn try_from(message_id: MessageId) -> std::result::Result<Self, Self::Error> {
         match message_id {
-            MessageId::String(string) => string,
-            _ => panic!("Cannot convert MessageId to String"),
+            MessageId::String(string) => Ok(string),
+            _ => Err(EventHubsError::with_message(
+                "Cannot convert MessageId to String",
+            )),
         }
     }
 }
 
-impl From<MessageId> for u64 {
-    fn from(message_id: MessageId) -> Self {
+impl TryFrom<MessageId> for u64 {
+    type Error = EventHubsError;
+
+    fn try_from(message_id: MessageId) -> std::result::Result<Self, Self::Error> {
         match message_id {
-            MessageId::Ulong(ulong) => ulong,
-            _ => panic!("Cannot convert MessageId to u64"),
+            MessageId::Ulong(ulong) => Ok(ulong),
+            _ => Err(EventHubsError::with_message(
+                "Cannot convert MessageId to u64",
+            )),
         }
     }
 }