Added Reconnect support for Event Hubs (#2826)

* Refactoring to enable reconnection support

* Refactored AMQP errors to distinguish between connection, link, and session disconnect; send_batch consumes the batch
---------

Co-authored-by: Heath Stewart <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

Cargo.lock

@@ -84,9 +84,9 @@ cargo_metadata = "0.18.1"
 clap = { version = "4.4.16", features = ["derive"] }
 criterion = { version = "0.5", features = ["async_tokio"] }
 dyn-clone = "1.0"
-fe2o3-amqp = { version = "0.14", features = ["native-tls", "tracing", "uuid"] }
+fe2o3-amqp = { version = "0.14", features = ["native-tls", "uuid"] }
 fe2o3-amqp-ext = { version = "0.14" }
-fe2o3-amqp-management = { version = "0.14", features = ["tracing"] }
+fe2o3-amqp-management = { version = "0.14" }
 fe2o3-amqp-cbs = { version = "0.14" }
 fe2o3-amqp-types = { version = "0.14" }
 flate2 = "1.1.0"

Cargo.toml

@@ -1,5 +1,15 @@
 # Release History
 
+## 0.7.0 (Unreleased)
+
+### Features Added
+
+### Breaking Changes
+
+- Distinguish remote disconnect and remote closed errors by origin
+
+### Bugs Fixed
+
 ## 0.6.0 (2025-08-01)
 
 ### Other Changes

sdk/core/azure_core_amqp/CHANGELOG.md

@@ -10,11 +10,21 @@ use crate::{AmqpOrderedMap, AmqpSymbol, AmqpValue};
 /// Type of AMQP error.
 pub enum AmqpErrorKind {
     AmqpDescribedError(AmqpDescribedError),
+
     /// Remote peer closed the link
-    ClosedByRemote(Box<dyn std::error::Error + Send + Sync>),
+    LinkClosedByRemote(Box<dyn std::error::Error + Send + Sync>),
+    /// Remote peer closed the session
+    SessionClosedByRemote(Box<dyn std::error::Error + Send + Sync>),
+    /// Remote peer closed the connection
+    ConnectionClosedByRemote(Box<dyn std::error::Error + Send + Sync>),
+
+    /// Remote peer detached the link
+    LinkDetachedByRemote(Box<dyn std::error::Error + Send + Sync>),
+    /// Remote peer detached the session
+    SessionDetachedByRemote(Box<dyn std::error::Error + Send + Sync>),
 
-    /// Remote peer detached
-    DetachedByRemote(Box<dyn std::error::Error + Send + Sync>),
+    /// Remote peer detached the connection
+    ConnectionDetachedByRemote(Box<dyn std::error::Error + Send + Sync>),
 
     /// The send request was rejected by the remote peer.
     NonTerminalDeliveryState,
@@ -83,6 +93,7 @@ create_extensible_enum!(
         SessionCannotBeLocked,
         "com.microsoft:session-cannot-be-locked"
     ),
+    (EntityUpdated, "com.microsoft:entity-updated"),
     (MessageNotFound, "com.microsoft:message-not-found"),
     (SessionNotFound, "com.microsoft:session-not-found"),
     (EntityAlreadyExists, "com.microsoft:entity-already-exists"),
@@ -182,8 +193,12 @@ impl std::error::Error for AmqpError {
         match &self.kind {
             AmqpErrorKind::TransportImplementationError(s)
             | AmqpErrorKind::DetachError(s)
-            | AmqpErrorKind::ClosedByRemote(s)
-            | AmqpErrorKind::DetachedByRemote(s)
+            | AmqpErrorKind::LinkClosedByRemote(s)
+            | AmqpErrorKind::LinkDetachedByRemote(s)
+            | AmqpErrorKind::SessionClosedByRemote(s)
+            | AmqpErrorKind::SessionDetachedByRemote(s)
+            | AmqpErrorKind::ConnectionClosedByRemote(s)
+            | AmqpErrorKind::ConnectionDetachedByRemote(s)
             | AmqpErrorKind::LinkStateError(s)
             | AmqpErrorKind::ConnectionDropped(s) => Some(s.as_ref()),
             AmqpErrorKind::ManagementStatusCode(_, _) => None,
@@ -211,11 +226,23 @@ impl std::fmt::Display for AmqpError {
                     write!(f, "Management API returned status code: {}", status_code,)
                 }
             }
-            AmqpErrorKind::DetachedByRemote(err) => {
-                write!(f, "Remote detached with error: {}", err)
+            AmqpErrorKind::ConnectionDetachedByRemote(err) => {
+                write!(f, "Remote connection detached with error: {}", err)
+            }
+            AmqpErrorKind::LinkDetachedByRemote(err) => {
+                write!(f, "Remote link detached with error: {}", err)
+            }
+            AmqpErrorKind::SessionDetachedByRemote(err) => {
+                write!(f, "Remote session detached with error: {}", err)
+            }
+            AmqpErrorKind::LinkClosedByRemote(err) => {
+                write!(f, "Remote link closed with error: {}", err)
+            }
+            AmqpErrorKind::SessionClosedByRemote(err) => {
+                write!(f, "Remote session closed with error: {}", err)
             }
-            AmqpErrorKind::ClosedByRemote(err) => {
-                write!(f, "Remote closed with error: {}", err)
+            AmqpErrorKind::ConnectionClosedByRemote(err) => {
+                write!(f, "Remote connection closed with error: {}", err)
             }
             AmqpErrorKind::DetachError(err) => {
                 write!(f, "AMQP Detach Error: {} ", err)

sdk/core/azure_core_amqp/src/error.rs

@@ -187,7 +187,7 @@ impl From<Fe2o3ConnectionOpenError> for azure_core::Error {
                 azure_core::Error::from(parse_error)
             }
             fe2o3_amqp::connection::OpenError::RemoteClosed => {
-                AmqpErrorKind::ClosedByRemote(Box::new(e.0)).into()
+                AmqpErrorKind::ConnectionClosedByRemote(Box::new(e.0)).into()
             }
             fe2o3_amqp::connection::OpenError::RemoteClosedWithError(error) => {
                 AmqpErrorKind::AmqpDescribedError(error.into()).into()
@@ -207,7 +207,7 @@ impl From<Fe2o3ConnectionError> for azure_core::Error {
                 azure_core::Error::from(Fe2o3TransportError(error))
             }
             fe2o3_amqp::connection::Error::RemoteClosed => {
-                AmqpErrorKind::ClosedByRemote(Box::new(e.0)).into()
+                AmqpErrorKind::ConnectionClosedByRemote(Box::new(e.0)).into()
             }
             fe2o3_amqp::connection::Error::RemoteClosedWithError(error) => {
                 AmqpErrorKind::AmqpDescribedError(error.into()).into()

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

@@ -115,13 +115,13 @@ impl From<fe2o3_amqp::link::DetachError> for AmqpError {
     fn from(e: fe2o3_amqp::link::DetachError) -> Self {
         match e {
             fe2o3_amqp::link::DetachError::DetachedByRemote => {
-                Self::from(AmqpErrorKind::DetachedByRemote(Box::new(e)))
+                Self::from(AmqpErrorKind::LinkDetachedByRemote(Box::new(e)))
             }
             fe2o3_amqp::link::DetachError::RemoteDetachedWithError(error) => {
                 Self::from(AmqpErrorKind::AmqpDescribedError(error.into()))
             }
             fe2o3_amqp::link::DetachError::ClosedByRemote => {
-                Self::from(AmqpErrorKind::ClosedByRemote(Box::new(e)))
+                Self::from(AmqpErrorKind::LinkClosedByRemote(Box::new(e)))
             }
             fe2o3_amqp::link::DetachError::RemoteClosedWithError(error) => {
                 Self::from(AmqpErrorKind::AmqpDescribedError(error.into()))
@@ -141,10 +141,10 @@ impl From<fe2o3_amqp::link::LinkStateError> for AmqpError {
                 AmqpErrorKind::AmqpDescribedError(e.into()).into()
             }
             fe2o3_amqp::link::LinkStateError::RemoteClosed => {
-                AmqpErrorKind::ClosedByRemote(Box::new(e)).into()
+                AmqpErrorKind::LinkClosedByRemote(Box::new(e)).into()
             }
             fe2o3_amqp::link::LinkStateError::RemoteDetached => {
-                AmqpErrorKind::DetachedByRemote(Box::new(e)).into()
+                AmqpErrorKind::LinkDetachedByRemote(Box::new(e)).into()
             }
             _ => AmqpErrorKind::LinkStateError(e.into()).into(),
         }

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

@@ -88,7 +88,9 @@ impl AmqpReceiverApis for Fe2o3AmqpReceiver {
         match res {
             Ok(_) => Ok(()),
             Err(e) => match e.kind() {
-                AmqpErrorKind::ClosedByRemote(_) => {
+                AmqpErrorKind::LinkClosedByRemote(_)
+                | AmqpErrorKind::SessionClosedByRemote(_)
+                | AmqpErrorKind::ConnectionClosedByRemote(_) => {
                     info!("Error detaching receiver: {:?} - ignored", e);
                     Ok(())
                 }

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

@@ -107,7 +107,9 @@ impl AmqpSenderApis for Fe2o3AmqpSender {
         match res {
             Ok(_) => Ok(()),
             Err(e) => match e.kind() {
-                AmqpErrorKind::ClosedByRemote(_) => {
+                AmqpErrorKind::LinkClosedByRemote(_)
+                | AmqpErrorKind::SessionClosedByRemote(_)
+                | AmqpErrorKind::ConnectionClosedByRemote(_) => {
                     info!("Error detaching sender: {:?}", e);
                     Ok(())
                 }

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

@@ -149,7 +149,7 @@ impl From<fe2o3_amqp::session::BeginError> for AmqpError {
                 AmqpErrorKind::ConnectionDropped(Box::new(e)).into()
             }
             fe2o3_amqp::session::BeginError::RemoteEnded => {
-                AmqpErrorKind::ClosedByRemote(Box::new(e)).into()
+                AmqpErrorKind::SessionClosedByRemote(Box::new(e)).into()
             }
             fe2o3_amqp::session::BeginError::RemoteEndedWithError(error) => {
                 AmqpErrorKind::AmqpDescribedError(error.into()).into()
@@ -173,7 +173,7 @@ impl From<fe2o3_amqp::session::Error> for AmqpError {
                 AmqpErrorKind::TransportImplementationError(Box::new(e)).into()
             }
             fe2o3_amqp::session::Error::RemoteEnded => {
-                AmqpErrorKind::ClosedByRemote(Box::new(e)).into()
+                AmqpErrorKind::SessionClosedByRemote(Box::new(e)).into()
             }
             fe2o3_amqp::session::Error::RemoteEndedWithError(error) => {
                 AmqpErrorKind::AmqpDescribedError(error.into()).into()

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

@@ -4,8 +4,13 @@
 
 ### Features Added
 
+- Reconnect support for EventHubs operations.
+
 ### Breaking Changes
 
+- `ProducerClient::send_batch` now consumes its `batch` argument.
+- `RetryOptions::max_retries` is a `u32` not a `usize`.
+
 ### Bugs Fixed
 
 ### Other Changes