wowlab_centrifuge/
error.rs

1use crate::codes;
2
3const TEMPORARY_ERROR_CODE_THRESHOLD: u32 = 100;
4
5/// Failure returned by the Centrifugo client.
6#[derive(Debug, thiserror::Error)]
7#[error("{kind}")]
8pub struct Error {
9    #[source]
10    kind: ErrorKind,
11}
12
13#[derive(Debug, thiserror::Error)]
14enum ErrorKind {
15    #[error("WebSocket error: {0}")]
16    WebSocket(#[source] tokio_tungstenite::tungstenite::Error),
17    #[error("HTTP error: {0}")]
18    Http(#[source] reqwest::Error),
19    #[error("HTTP error: server returned {0}")]
20    HttpStatus(reqwest::StatusCode),
21    #[error("Protocol error: {0}")]
22    Protocol(String),
23    #[error("Protocol error: {0}")]
24    ProtocolDecode(#[source] prost::DecodeError),
25    #[error("Server error {code}: {message}")]
26    Server {
27        code: u32,
28        message: String,
29        temporary: bool,
30    },
31    #[error("Connection closed")]
32    ConnectionClosed,
33    #[error("Not connected")]
34    NotConnected,
35    #[error("Read timeout (no data for 45s)")]
36    Timeout,
37    #[error("Server ping timeout")]
38    NoPing,
39    #[error("Subscription not found: {0}")]
40    SubscriptionNotFound(String),
41    #[error("Already subscribed: {0}")]
42    AlreadySubscribed(String),
43}
44
45impl Error {
46    /// Creates an error for a malformed or unexpected protocol response.
47    pub fn protocol(message: impl Into<String>) -> Self {
48        Self::new(ErrorKind::Protocol(message.into()))
49    }
50
51    pub(crate) fn http(source: reqwest::Error) -> Self {
52        Self::new(ErrorKind::Http(source))
53    }
54
55    pub(crate) fn http_status(status: reqwest::StatusCode) -> Self {
56        Self::new(ErrorKind::HttpStatus(status))
57    }
58
59    pub(crate) fn protocol_decode(source: prost::DecodeError) -> Self {
60        Self::new(ErrorKind::ProtocolDecode(source))
61    }
62
63    pub(crate) fn server(code: u32, message: impl Into<String>, temporary: bool) -> Self {
64        Self::new(ErrorKind::Server {
65            code,
66            message: message.into(),
67            temporary,
68        })
69    }
70
71    pub(crate) fn connection_closed() -> Self {
72        Self::new(ErrorKind::ConnectionClosed)
73    }
74
75    pub(crate) fn not_connected() -> Self {
76        Self::new(ErrorKind::NotConnected)
77    }
78
79    pub(crate) fn timeout() -> Self {
80        Self::new(ErrorKind::Timeout)
81    }
82
83    pub(crate) fn no_ping() -> Self {
84        Self::new(ErrorKind::NoPing)
85    }
86
87    pub(crate) fn subscription_not_found(channel: impl Into<String>) -> Self {
88        Self::new(ErrorKind::SubscriptionNotFound(channel.into()))
89    }
90
91    pub(crate) fn already_subscribed(channel: impl Into<String>) -> Self {
92        Self::new(ErrorKind::AlreadySubscribed(channel.into()))
93    }
94
95    pub(crate) fn from_proto(error: crate::proto::Error) -> Self {
96        Self::server(error.code, error.message, error.temporary)
97    }
98
99    fn new(kind: ErrorKind) -> Self {
100        Self { kind }
101    }
102
103    /// Returns the Centrifugo server error code, when the server supplied one.
104    #[must_use]
105    pub fn server_code(&self) -> Option<u32> {
106        match &self.kind {
107            ErrorKind::Server { code, .. } => Some(*code),
108            _ => None,
109        }
110    }
111
112    /// Returns whether retrying the failed operation may succeed.
113    #[must_use]
114    pub fn is_temporary(&self) -> bool {
115        match &self.kind {
116            ErrorKind::Server {
117                code, temporary, ..
118            } => *temporary || *code < TEMPORARY_ERROR_CODE_THRESHOLD,
119            ErrorKind::WebSocket(_)
120            | ErrorKind::Timeout
121            | ErrorKind::NoPing
122            | ErrorKind::ConnectionClosed => true,
123            _ => false,
124        }
125    }
126
127    /// Returns whether reconnecting requires a fresh connection token.
128    #[must_use]
129    pub fn requires_token_refresh(&self) -> bool {
130        self.server_code() == Some(codes::error::TOKEN_EXPIRED)
131    }
132
133    /// Returns whether the disconnect is expected during normal connection teardown.
134    #[must_use]
135    pub fn is_benign_disconnect(&self) -> bool {
136        match &self.kind {
137            ErrorKind::WebSocket(error) => {
138                let message = error.to_string();
139
140                message.contains("close_notify")
141                    || message.contains("Connection reset")
142                    || message.contains("connection reset")
143            }
144            ErrorKind::NoPing | ErrorKind::ConnectionClosed => true,
145            _ => false,
146        }
147    }
148}
149
150impl From<tokio_tungstenite::tungstenite::Error> for Error {
151    fn from(error: tokio_tungstenite::tungstenite::Error) -> Self {
152        Self::new(ErrorKind::WebSocket(error))
153    }
154}
155
156impl From<reqwest::Error> for Error {
157    fn from(error: reqwest::Error) -> Self {
158        Self::http(error)
159    }
160}
161
162impl From<prost::DecodeError> for Error {
163    fn from(error: prost::DecodeError) -> Self {
164        Self::protocol_decode(error)
165    }
166}
167
168#[cfg(test)]
169mod tests {
170    use std::error::Error as _;
171
172    use googletest::prelude::*;
173    use prost::Message as _;
174
175    use super::Error;
176
177    #[gtest]
178    fn protocol_decode_preserves_source_chain() -> Result<()> {
179        let decode_error = crate::proto::Reply::decode_length_delimited([0xff].as_slice())
180            .err()
181            .or_fail()?;
182        let error = Error::protocol_decode(decode_error);
183
184        let kind = error.source().or_fail()?;
185        let decode = kind.source().or_fail()?;
186
187        verify_true!(decode.downcast_ref::<prost::DecodeError>().is_some())
188    }
189
190    #[gtest]
191    #[tokio::test]
192    async fn http_failure_preserves_source_chain() -> Result<()> {
193        let request_error = reqwest::Client::new()
194            .get("not-a-valid-absolute-url")
195            .send()
196            .await
197            .err()
198            .or_fail()?;
199        let error = Error::from(request_error);
200
201        let kind = error.source().or_fail()?;
202        let request = kind.source().or_fail()?;
203
204        verify_true!(request.downcast_ref::<reqwest::Error>().is_some())
205    }
206}
wowlab_centrifuge/error.rs

wowlab_centrifuge/
error.rs
