ruststream_lapin/
error.rs

1//! The crate error type shared by broker, publishers, subscribers, and the requester.
2
3use std::error::Error as StdError;
4use std::time::Duration;
5
6use thiserror::Error;
7
8/// Errors returned by [`LapinBroker`](crate::LapinBroker) and the types it hands out.
9///
10/// Underlying [`lapin`](https://docs.rs/lapin) errors are boxed as sources so the client library
11/// does not leak into this crate's public API surface.
12#[derive(Debug, Error)]
13#[non_exhaustive]
14pub enum AmqpError {
15    /// Establishing or closing the connection failed.
16    #[error("amqp connection error: {0}")]
17    Connect(#[source] Box<dyn StdError + Send + Sync>),
18
19    /// Publishing a message failed, or the broker refused to confirm it.
20    #[error("amqp publish error: {0}")]
21    Publish(#[source] Box<dyn StdError + Send + Sync>),
22
23    /// Opening a subscription (channel, `QoS`, or consume) failed.
24    #[error("amqp subscribe error: {0}")]
25    Subscribe(#[source] Box<dyn StdError + Send + Sync>),
26
27    /// Receiving a delivery from an open consumer failed.
28    #[error("amqp consume error: {0}")]
29    Consume(#[source] Box<dyn StdError + Send + Sync>),
30
31    /// Declaring the expected topology (exchange, queue, or binding) failed.
32    #[error("amqp topology declaration error: {0}")]
33    Declare(#[source] Box<dyn StdError + Send + Sync>),
34
35    /// Sending a request or receiving its reply failed.
36    #[error("amqp request error: {0}")]
37    Request(#[source] Box<dyn StdError + Send + Sync>),
38
39    /// No reply arrived within the caller's deadline.
40    ///
41    /// The pending request is dropped; a reply arriving later is discarded.
42    #[error("amqp request timed out after {0:?} without a reply")]
43    RequestTimeout(Duration),
44
45    /// An operation ran against a connection that has already shut down.
46    ///
47    /// The ladder makes misuse through the owner of the connected broker a compile error, so
48    /// this reports the case it cannot cover: a handle aliasing the connection (a publisher
49    /// paired before the shutdown, a requester clone) used afterwards, which must fail rather
50    /// than silently succeed against a dead connection.
51    #[error("amqp connection is closed; the operation targeting {target:?} cannot proceed")]
52    Closed {
53        /// The routing key, queue, or exchange the operation targeted.
54        target: String,
55    },
56
57    /// A transaction call ran out of order on a transactional publisher.
58    ///
59    /// A commit or an abort with no open transaction, or a second begin while one is open. The
60    /// message names the offending call.
61    #[error("invalid transaction state: {0}")]
62    Transaction(String),
63
64    /// The requested combination of options cannot be executed.
65    ///
66    /// The message names the offending option and the remediation.
67    #[error("invalid options: {0}")]
68    InvalidOptions(String),
69}
70
71impl AmqpError {
72    pub(crate) fn connect(err: lapin::Error) -> Self {
73        Self::Connect(Box::new(err))
74    }
75
76    pub(crate) fn publish(err: lapin::Error) -> Self {
77        Self::Publish(Box::new(err))
78    }
79
80    pub(crate) fn subscribe(err: lapin::Error) -> Self {
81        Self::Subscribe(Box::new(err))
82    }
83
84    pub(crate) fn consume(err: lapin::Error) -> Self {
85        Self::Consume(Box::new(err))
86    }
87
88    pub(crate) fn declare(err: lapin::Error) -> Self {
89        Self::Declare(Box::new(err))
90    }
91
92    pub(crate) fn request(err: lapin::Error) -> Self {
93        Self::Request(Box::new(err))
94    }
95
96    pub(crate) fn closed(target: &str) -> Self {
97        Self::Closed {
98            target: target.to_owned(),
99        }
100    }
101}
ruststream_lapin/error.rs

ruststream_lapin/
error.rs
