azure_messaging_eventhubs 0.16.0

// Copyright (c) Microsoft Corporation. All Rights Reserved.
// Licensed under the MIT License.

use azure_core_amqp::{
    error::{AmqpErrorCondition, AmqpErrorKind},
    AmqpDescribedError, AmqpError,
};
use std::borrow::Cow;

/// A specialized `Result` type for Event Hubs operations.
pub type Result<T> = std::result::Result<T, EventHubsError>;

/// Represents the different kinds of errors that can occur in the Eventhubs module.
#[derive(Debug)]
#[non_exhaustive]
pub enum ErrorKind {
    /// A simple error.
    SimpleMessage(Cow<'static, str>),

    /// The management response is invalid.
    InvalidManagementResponse,

    /// The message was rejected.
    SendRejected(Option<AmqpDescribedError>),

    /// Represents an Azure Core error
    AzureCore(azure_core::Error),

    /// The maximum batch size the caller asked for cannot be used. It is
    /// either zero, which is too small to hold the batch envelope, or it is
    /// larger than the maximum the sender link allows.
    ///
    /// Mirrors the `ArgumentOutOfRangeException` that .NET raises for the same
    /// input. Match on the variant to tell it apart from a transport failure:
    /// `matches!(err.kind, ErrorKind::InvalidBatchSize { .. })`.
    InvalidBatchSize {
        /// The maximum size in bytes the caller asked for.
        requested: u64,
        /// The largest maximum size in bytes the sender link allows.
        max_allowed: u64,
    },

    /// Represents the source of the AMQP error.
    /// This is used to wrap an AMQP error in an Even Hubs error.
    ///
    AmqpError(AmqpError),

    /// The service settled the transfer, but it did not durably accept it.
    ///
    /// The broker returned an AMQP `Modified` or `Released` outcome. Neither
    /// outcome means the service stored the events. The buffered producer
    /// reports this as a delivery failure.
    ///
    /// A `Released` or `Modified` outcome does not prove that the service
    /// discarded the events either. If the caller sends the same events again,
    /// the service can store them two times.
    SendNotAccepted(Cow<'static, str>),

    /// Receiver was disconnected by the broker because another receiver
    /// attached with the same or higher epoch (owner level). The inner
    /// `AmqpDescribedError` is for logging; match on the variant:
    /// `matches!(err.kind, ErrorKind::ConsumerDisconnected(_))`.
    /// Mirrors `EventHubsException.FailureReason.ConsumerDisconnected` (.NET).
    ConsumerDisconnected(Option<AmqpDescribedError>),

    /// The event carries no offset and no sequence number, so it names no
    /// position in the partition. A checkpoint built from such an event holds
    /// no position, and it erases the position the checkpoint store already
    /// holds.
    ///
    /// Mirrors the `InvalidOperationException` that .NET raises for the same
    /// input ("A checkpoint cannot be created or updated using an empty
    /// event."). Match on the variant to tell it apart from a store failure:
    /// `matches!(err.kind, ErrorKind::MissingCheckpointMetadata { .. })`.
    MissingCheckpointMetadata {
        /// The identifier of the partition the checkpoint is for.
        partition_id: String,
    },
}

/// Represents an error that can occur in the Event Hubs module.
pub struct EventHubsError {
    /// The kind of error that occurred.
    pub kind: ErrorKind,
}

impl EventHubsError {
    pub(crate) fn with_message<C>(message: C) -> EventHubsError
    where
        C: Into<Cow<'static, str>>,
    {
        Self::from(ErrorKind::SimpleMessage(message.into()))
    }
}

impl std::error::Error for EventHubsError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match &self.kind {
            ErrorKind::AmqpError(source) => Some(source),
            ErrorKind::AzureCore(e) => Some(e),
            _ => None,
        }
    }
}

impl std::fmt::Display for EventHubsError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.kind {
            ErrorKind::SimpleMessage(msg) => write!(f, "{}", msg),
            ErrorKind::AzureCore(e) => write!(f, "Azure Core Error: {}", e),
            ErrorKind::InvalidBatchSize {
                requested,
                max_allowed,
            } => write!(
                f,
                "Invalid maximum batch size: {} bytes. \
                 It must be from 1 to {} bytes, which is the maximum the sender link allows.",
                requested, max_allowed
            ),
            ErrorKind::SendRejected(e) => write!(f, "Send rejected: {:?}", e),
            ErrorKind::InvalidManagementResponse => f.write_str("Invalid management response"),
            ErrorKind::AmqpError(source) => write!(f, "AMQP Error: {:?}", source),
            ErrorKind::SendNotAccepted(msg) => {
                write!(f, "Send was not durably accepted: {}", msg)
            }
            ErrorKind::ConsumerDisconnected(e) => {
                write!(
                    f,
                    "Consumer disconnected by broker (partition stolen): {:?}",
                    e
                )
            }
            ErrorKind::MissingCheckpointMetadata { partition_id } => write!(
                f,
                "Cannot record a checkpoint for partition {}. \
                 The event carries no offset and no sequence number, \
                 so there is nothing to record.",
                partition_id
            ),
        }
    }
}

impl std::fmt::Debug for EventHubsError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Event Hubs Error: {}", self)
    }
}

impl From<ErrorKind> for EventHubsError {
    fn from(kind: ErrorKind) -> Self {
        Self { kind }
    }
}

/// Maximum number of links to follow in an error source chain. It stops a
/// pathological self-referential chain from looping forever.
const MAX_ERROR_CHAIN_DEPTH: usize = 16;

/// Returns the `amqp:link:stolen` described error if `error` is one, or wraps
/// one in its [`std::error::Error::source`] chain.
///
/// The condition can arrive at the top level (an in-flight receive, or a
/// re-attach that the broker rejected) or wrapped through `azure_core::Error`
/// by the `ensure_*` wrappers. Both must be recognized, so the whole chain is
/// walked instead of only the top-level kind.
pub(crate) fn find_link_stolen(error: &AmqpError) -> Option<&AmqpDescribedError> {
    use std::error::Error as _;
    fn described(e: &AmqpError) -> Option<&AmqpDescribedError> {
        match e.kind() {
            AmqpErrorKind::AmqpDescribedError(d)
                if matches!(d.condition, AmqpErrorCondition::LinkStolen) =>
            {
                Some(d)
            }
            _ => None,
        }
    }
    if let Some(d) = described(error) {
        return Some(d);
    }
    let mut cause: Option<&(dyn std::error::Error + 'static)> = error.source();
    for _ in 0..MAX_ERROR_CHAIN_DEPTH {
        let c = cause?;
        if let Some(amqp) = c.downcast_ref::<AmqpError>() {
            if let Some(d) = described(amqp) {
                return Some(d);
            }
        }
        cause = c.source();
    }
    None
}

impl From<AmqpError> for EventHubsError {
    fn from(e: AmqpError) -> Self {
        Self {
            kind: ErrorKind::AmqpError(e),
        }
    }
}

impl From<azure_core::Error> for EventHubsError {
    fn from(e: azure_core::Error) -> Self {
        Self {
            kind: ErrorKind::AzureCore(e),
        }
    }
}

impl From<EventHubsError> for azure_core::Error {
    fn from(value: EventHubsError) -> Self {
        match value.kind {
            ErrorKind::AzureCore(e) => e,
            _ => azure_core::Error::with_error(
                azure_core::error::ErrorKind::Other,
                value,
                "EventHubs Error",
            ),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use azure_core_amqp::{AmqpErrorCondition, AmqpOrderedMap, AmqpSymbol};
    use std::error::Error;

    #[test]
    fn test_eventhubs_error_with_message_borrowed() {
        let error = EventHubsError::with_message("Test error");
        assert!(matches!(error.kind, ErrorKind::SimpleMessage(_)));
        assert_eq!(format!("{}", error), "Test error");
    }

    #[test]
    fn test_eventhubs_error_with_message_owned() {
        let error = EventHubsError::with_message("Owned error".to_string());
        assert!(matches!(error.kind, ErrorKind::SimpleMessage(_)));
        assert_eq!(format!("{}", error), "Owned error");
    }

    #[test]
    fn test_eventhubs_error_from_error_kind_simple_message() {
        let kind = ErrorKind::SimpleMessage(Cow::Borrowed("Simple message"));
        let error: EventHubsError = kind.into();
        assert!(matches!(error.kind, ErrorKind::SimpleMessage(_)));
        assert_eq!(format!("{}", error), "Simple message");
    }

    #[test]
    fn test_eventhubs_error_from_error_kind_invalid_management_response() {
        let kind = ErrorKind::InvalidManagementResponse;
        let error: EventHubsError = kind.into();
        assert!(matches!(error.kind, ErrorKind::InvalidManagementResponse));
        assert_eq!(format!("{}", error), "Invalid management response");
    }

    #[test]
    fn test_eventhubs_error_from_error_kind_send_rejected_with_details() {
        let mut info = AmqpOrderedMap::new();
        info.insert(
            AmqpSymbol("error-detail".to_string()),
            azure_core_amqp::AmqpValue::String("Quota exceeded".to_string()),
        );
        let described_error = AmqpDescribedError::new(
            AmqpErrorCondition::ResourceLimitExceeded,
            Some("Send quota exceeded".to_string()),
            info,
        );
        let kind = ErrorKind::SendRejected(Some(described_error));
        let error: EventHubsError = kind.into();
        assert!(matches!(error.kind, ErrorKind::SendRejected(_)));
        let display = format!("{}", error);
        assert!(display.contains("Send rejected"));
    }

    #[test]
    fn test_eventhubs_error_from_error_kind_send_rejected_without_details() {
        let kind = ErrorKind::SendRejected(None);
        let error: EventHubsError = kind.into();
        assert!(matches!(error.kind, ErrorKind::SendRejected(None)));
        let display = format!("{}", error);
        assert!(display.contains("Send rejected"));
    }

    #[test]
    fn test_eventhubs_error_from_amqp_error() {
        let amqp_error = AmqpError::from(azure_core_amqp::AmqpErrorKind::SimpleMessage(
            Cow::Borrowed("AMQP error"),
        ));
        let error: EventHubsError = amqp_error.into();
        assert!(matches!(error.kind, ErrorKind::AmqpError(_)));
        let display = format!("{}", error);
        assert!(display.contains("AMQP Error"));
    }

    #[test]
    fn test_eventhubs_error_from_azure_core_error() {
        let azure_error =
            azure_core::Error::with_message(azure_core::error::ErrorKind::Other, "Azure error");
        let error: EventHubsError = azure_error.into();
        assert!(matches!(error.kind, ErrorKind::AzureCore(_)));
        let display = format!("{}", error);
        assert!(display.contains("Azure Core Error"));
    }
    #[test]
    fn test_eventhubs_error_to_azure_core_error_from_simple_message() {
        let eventhubs_error = EventHubsError::with_message("Simple error");
        let converted: azure_core::Error = eventhubs_error.into();
        let display = format!("{}", converted);
        assert!(display.contains("EventHubs Error"));
    }

    #[test]
    fn test_eventhubs_error_to_azure_core_error_from_azure_core() {
        let azure_error = azure_core::Error::with_message(
            azure_core::error::ErrorKind::Other,
            "Original Azure error",
        );
        let eventhubs_error: EventHubsError = azure_error.into();
        let converted: azure_core::Error = eventhubs_error.into();
        assert!(format!("{}", converted).contains("Original Azure error"));
    }

    #[test]
    fn test_eventhubs_error_source_amqp() {
        let amqp_error = AmqpError::from(azure_core_amqp::AmqpErrorKind::SimpleMessage(
            Cow::Borrowed("AMQP source"),
        ));
        let error: EventHubsError = amqp_error.into();
        assert!(error.source().is_some());
    }

    #[test]
    fn test_eventhubs_error_source_azure_core() {
        let azure_error =
            azure_core::Error::with_message(azure_core::error::ErrorKind::Other, "Azure source");
        let error: EventHubsError = azure_error.into();
        assert!(error.source().is_some());
    }

    #[test]
    fn test_eventhubs_error_source_simple_message() {
        let error = EventHubsError::with_message("No source");
        assert!(error.source().is_none());
    }

    #[test]
    fn test_eventhubs_error_source_invalid_management_response() {
        let error = EventHubsError::from(ErrorKind::InvalidManagementResponse);
        assert!(error.source().is_none());
    }

    #[test]
    fn test_eventhubs_error_source_send_rejected() {
        let error = EventHubsError::from(ErrorKind::SendRejected(None));
        assert!(error.source().is_none());
    }

    #[test]
    fn test_eventhubs_error_display_simple_message() {
        let error = EventHubsError::with_message("Display test");
        assert_eq!(format!("{}", error), "Display test");
    }

    #[test]
    fn test_eventhubs_error_display_azure_core() {
        let azure_error =
            azure_core::Error::with_message(azure_core::error::ErrorKind::Other, "Azure display");
        let error: EventHubsError = azure_error.into();
        let display = format!("{}", error);
        assert!(display.contains("Azure Core Error"));
        assert!(display.contains("Azure display"));
    }

    #[test]
    fn test_eventhubs_error_display_send_rejected() {
        let error = EventHubsError::from(ErrorKind::SendRejected(None));
        let display = format!("{}", error);
        assert!(display.contains("Send rejected"));
    }

    #[test]
    fn test_eventhubs_error_display_invalid_management_response() {
        let error = EventHubsError::from(ErrorKind::InvalidManagementResponse);
        assert_eq!(format!("{}", error), "Invalid management response");
    }

    #[test]
    fn test_eventhubs_error_display_amqp_error() {
        let amqp_error = AmqpError::from(azure_core_amqp::AmqpErrorKind::SimpleMessage(
            Cow::Borrowed("AMQP display"),
        ));
        let error: EventHubsError = amqp_error.into();
        let display = format!("{}", error);
        assert!(display.contains("AMQP Error"));
    }

    #[test]
    fn test_eventhubs_error_debug() {
        let error = EventHubsError::with_message("Debug test");
        let debug_output = format!("{:?}", error);
        assert!(debug_output.contains("Event Hubs Error"));
        assert!(debug_output.contains("Debug test"));
    }

    #[test]
    fn test_eventhubs_error_debug_complex() {
        let azure_error =
            azure_core::Error::with_message(azure_core::error::ErrorKind::Other, "Complex");
        let error: EventHubsError = azure_error.into();
        let debug_output = format!("{:?}", error);
        assert!(debug_output.contains("Event Hubs Error"));
    }

    #[test]
    fn test_result_type_alias() {
        fn returns_result() -> Result<String> {
            Ok("Success".to_string())
        }

        let result = returns_result();
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "Success");
    }

    #[test]
    fn test_result_type_alias_error() {
        fn returns_error() -> Result<String> {
            Err(EventHubsError::with_message("Error"))
        }

        let result = returns_error();
        assert!(result.is_err());
        assert_eq!(format!("{}", result.unwrap_err()), "Error");
    }

    #[test]
    fn test_eventhubs_error_chain_conversions() {
        // Create an Azure Core error
        let azure_error =
            azure_core::Error::with_message(azure_core::error::ErrorKind::Other, "Original error");

        // Convert to EventHubsError
        let eventhubs_error: EventHubsError = azure_error.into();

        // Verify it's stored as AzureCore variant
        assert!(matches!(eventhubs_error.kind, ErrorKind::AzureCore(_)));

        // Convert back to Azure Core error
        let converted_azure: azure_core::Error = eventhubs_error.into();

        assert!(format!("{}", converted_azure).contains("Original error"));
    }

    #[test]
    fn test_eventhubs_error_amqp_described_error_integration() {
        let mut info = AmqpOrderedMap::new();
        info.insert(
            AmqpSymbol("tracking-id".to_string()),
            azure_core_amqp::AmqpValue::String("12345".to_string()),
        );

        let described_error = AmqpDescribedError::new(
            AmqpErrorCondition::UnauthorizedAccess,
            Some("Unauthorized access to partition".to_string()),
            info,
        );

        let error = EventHubsError::from(ErrorKind::SendRejected(Some(described_error)));

        assert!(matches!(error.kind, ErrorKind::SendRejected(Some(_))));
        let display = format!("{}", error);
        assert!(display.contains("Send rejected"));
    }

    #[test]
    fn test_eventhubs_error_all_variants_can_be_displayed() {
        let errors = vec![
            EventHubsError::with_message("Simple"),
            EventHubsError::from(ErrorKind::InvalidManagementResponse),
            EventHubsError::from(ErrorKind::SendRejected(None)),
            EventHubsError::from(ErrorKind::AzureCore(azure_core::Error::with_message(
                azure_core::error::ErrorKind::Other,
                "Azure",
            ))),
            EventHubsError::from(ErrorKind::AmqpError(AmqpError::from(
                azure_core_amqp::AmqpErrorKind::SimpleMessage(Cow::Borrowed("AMQP")),
            ))),
        ];

        for error in errors {
            let _ = format!("{}", error);
            let _ = format!("{:?}", error);
        }
    }
}