rocketmq_client_rust/
exception.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt;
16
17use crate::ClientError;
18use rocketmq_error::CORE_OPERATION_TIMED_OUT;
19use rocketmq_error::TRANSPORT_CONNECTION_TIMEOUT;
20use rocketmq_error::TRANSPORT_RESPONSE_TIMEOUT;
21use rocketmq_error::TRANSPORT_WRITE_TIMEOUT;
22use rocketmq_model::common::FAQUrl;
23
24const DEFAULT_RESPONSE_CODE: i32 = -1;
25const DEFAULT_BROKER_RESPONSE_CODE: i32 = 0;
26
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct MQClientException {
29    response_code: i32,
30    error_message: Option<String>,
31    message: String,
32}
33
34impl MQClientException {
35    pub fn new(error_message: impl Into<String>) -> Self {
36        let error_message = error_message.into();
37        let message = FAQUrl::attach_default_url(Some(error_message.as_str()));
38        Self {
39            response_code: DEFAULT_RESPONSE_CODE,
40            error_message: Some(error_message),
41            message,
42        }
43    }
44
45    pub fn new_with_code(response_code: i32, error_message: impl Into<String>) -> Self {
46        let error_message = error_message.into();
47        let message =
48            FAQUrl::attach_default_url(Some(format!("CODE: {response_code}  DESC: {error_message}").as_str()));
49        Self {
50            response_code,
51            error_message: Some(error_message),
52            message,
53        }
54    }
55
56    pub fn response_code(&self) -> i32 {
57        self.response_code
58    }
59
60    pub fn get_response_code(&self) -> i32 {
61        self.response_code()
62    }
63
64    pub fn set_response_code(&mut self, response_code: i32) -> &mut Self {
65        self.response_code = response_code;
66        self
67    }
68
69    pub fn error_message(&self) -> Option<&str> {
70        self.error_message.as_deref()
71    }
72
73    pub fn get_error_message(&self) -> Option<&str> {
74        self.error_message()
75    }
76
77    pub fn set_error_message(&mut self, error_message: impl Into<String>) {
78        self.error_message = Some(error_message.into());
79    }
80
81    pub fn from_client_error(error: &ClientError) -> Self {
82        Self::new(error.to_string())
83    }
84}
85
86impl fmt::Display for MQClientException {
87    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88        f.write_str(self.message.as_str())
89    }
90}
91
92impl std::error::Error for MQClientException {}
93
94impl From<MQClientException> for ClientError {
95    fn from(exception: MQClientException) -> Self {
96        ClientError::illegal_argument_source(exception)
97    }
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct MQBrokerException {
102    response_code: i32,
103    error_message: Option<String>,
104    broker_addr: Option<String>,
105    message: String,
106}
107
108impl Default for MQBrokerException {
109    fn default() -> Self {
110        Self {
111            response_code: DEFAULT_BROKER_RESPONSE_CODE,
112            error_message: None,
113            broker_addr: None,
114            message: String::new(),
115        }
116    }
117}
118
119impl MQBrokerException {
120    pub fn new(response_code: i32, error_message: impl Into<String>) -> Self {
121        Self::new_with_optional_broker(response_code, error_message, None)
122    }
123
124    pub fn new_with_broker(
125        response_code: i32,
126        error_message: impl Into<String>,
127        broker_addr: impl Into<String>,
128    ) -> Self {
129        Self::new_with_optional_broker(response_code, error_message, Some(broker_addr.into()))
130    }
131
132    pub fn new_with_optional_broker(
133        response_code: i32,
134        error_message: impl Into<String>,
135        broker_addr: Option<String>,
136    ) -> Self {
137        let error_message = error_message.into();
138        let broker_suffix = broker_addr
139            .as_ref()
140            .map(|broker_addr| format!(" BROKER: {broker_addr}"))
141            .unwrap_or_default();
142        let message = FAQUrl::attach_default_url(Some(
143            format!("CODE: {response_code}  DESC: {error_message}{broker_suffix}").as_str(),
144        ));
145        Self {
146            response_code,
147            error_message: Some(error_message),
148            broker_addr,
149            message,
150        }
151    }
152
153    pub fn response_code(&self) -> i32 {
154        self.response_code
155    }
156
157    pub fn get_response_code(&self) -> i32 {
158        self.response_code()
159    }
160
161    pub fn error_message(&self) -> Option<&str> {
162        self.error_message.as_deref()
163    }
164
165    pub fn get_error_message(&self) -> Option<&str> {
166        self.error_message()
167    }
168
169    pub fn broker_addr(&self) -> Option<&str> {
170        self.broker_addr.as_deref()
171    }
172
173    pub fn get_broker_addr(&self) -> Option<&str> {
174        self.broker_addr()
175    }
176
177    pub fn from_client_error(error: &ClientError) -> Option<Self> {
178        if !error.is(&rocketmq_error::BROKER_OPERATION_FAILED) {
179            return None;
180        }
181        if let Some(source) = error.source_ref::<MQBrokerException>() {
182            return Some(source.clone());
183        }
184        Some(Self::new_with_optional_broker(
185            error.broker_response_code()?,
186            error.descriptor().public_message(),
187            error.broker_addr().map(ToOwned::to_owned),
188        ))
189    }
190}
191
192impl fmt::Display for MQBrokerException {
193    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
194        f.write_str(self.message.as_str())
195    }
196}
197
198impl std::error::Error for MQBrokerException {}
199
200impl From<MQBrokerException> for ClientError {
201    fn from(exception: MQBrokerException) -> Self {
202        let code = exception.response_code;
203        let broker_addr = exception.broker_addr.clone();
204        ClientError::broker_operation_source("BROKER_OPERATION", code, broker_addr, exception)
205    }
206}
207
208#[derive(Debug, Clone, Default, PartialEq, Eq)]
209pub struct OffsetNotFoundException {
210    broker_exception: MQBrokerException,
211}
212
213impl OffsetNotFoundException {
214    pub fn new(response_code: i32, error_message: impl Into<String>) -> Self {
215        Self {
216            broker_exception: MQBrokerException::new(response_code, error_message),
217        }
218    }
219
220    pub fn new_with_broker(
221        response_code: i32,
222        error_message: impl Into<String>,
223        broker_addr: impl Into<String>,
224    ) -> Self {
225        Self {
226            broker_exception: MQBrokerException::new_with_broker(response_code, error_message, broker_addr),
227        }
228    }
229
230    pub fn response_code(&self) -> i32 {
231        self.broker_exception.response_code()
232    }
233
234    pub fn get_response_code(&self) -> i32 {
235        self.response_code()
236    }
237
238    pub fn error_message(&self) -> Option<&str> {
239        self.broker_exception.error_message()
240    }
241
242    pub fn get_error_message(&self) -> Option<&str> {
243        self.error_message()
244    }
245
246    pub fn broker_addr(&self) -> Option<&str> {
247        self.broker_exception.broker_addr()
248    }
249
250    pub fn get_broker_addr(&self) -> Option<&str> {
251        self.broker_addr()
252    }
253
254    pub fn as_broker_exception(&self) -> &MQBrokerException {
255        &self.broker_exception
256    }
257}
258
259impl fmt::Display for OffsetNotFoundException {
260    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
261        self.broker_exception.fmt(f)
262    }
263}
264
265impl std::error::Error for OffsetNotFoundException {}
266
267impl From<OffsetNotFoundException> for MQBrokerException {
268    fn from(exception: OffsetNotFoundException) -> Self {
269        exception.broker_exception
270    }
271}
272
273impl From<OffsetNotFoundException> for ClientError {
274    fn from(exception: OffsetNotFoundException) -> Self {
275        let broker_exception: MQBrokerException = exception.into();
276        broker_exception.into()
277    }
278}
279
280#[derive(Debug, Clone, PartialEq, Eq)]
281pub struct RequestTimeoutException {
282    response_code: i32,
283    error_message: Option<String>,
284    message: String,
285}
286
287impl RequestTimeoutException {
288    pub fn new(error_message: impl Into<String>) -> Self {
289        let error_message = error_message.into();
290        Self {
291            response_code: DEFAULT_RESPONSE_CODE,
292            message: error_message.clone(),
293            error_message: Some(error_message),
294        }
295    }
296
297    pub fn new_with_code(response_code: i32, error_message: impl Into<String>) -> Self {
298        let error_message = error_message.into();
299        let message = format!("CODE: {response_code}  DESC: {error_message}");
300        Self {
301            response_code,
302            error_message: Some(error_message),
303            message,
304        }
305    }
306
307    pub fn response_code(&self) -> i32 {
308        self.response_code
309    }
310
311    pub fn get_response_code(&self) -> i32 {
312        self.response_code()
313    }
314
315    pub fn set_response_code(&mut self, response_code: i32) -> &mut Self {
316        self.response_code = response_code;
317        self
318    }
319
320    pub fn error_message(&self) -> Option<&str> {
321        self.error_message.as_deref()
322    }
323
324    pub fn get_error_message(&self) -> Option<&str> {
325        self.error_message()
326    }
327
328    pub fn set_error_message(&mut self, error_message: impl Into<String>) {
329        self.error_message = Some(error_message.into());
330    }
331
332    pub fn from_client_error(error: &ClientError) -> Option<Self> {
333        match error.descriptor() {
334            descriptor if descriptor.code() == CORE_OPERATION_TIMED_OUT.code() => Some(Self::new(error.to_string())),
335            descriptor
336                if [
337                    &TRANSPORT_CONNECTION_TIMEOUT,
338                    &TRANSPORT_WRITE_TIMEOUT,
339                    &TRANSPORT_RESPONSE_TIMEOUT,
340                ]
341                .into_iter()
342                .any(|candidate| descriptor.code() == candidate.code()) =>
343            {
344                Some(Self::new(error.to_string()))
345            }
346            _ => None,
347        }
348    }
349}
350
351impl fmt::Display for RequestTimeoutException {
352    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
353        f.write_str(self.message.as_str())
354    }
355}
356
357impl std::error::Error for RequestTimeoutException {}
358
359impl From<RequestTimeoutException> for ClientError {
360    fn from(_exception: RequestTimeoutException) -> Self {
361        ClientError::timeout("REQUEST", 0)
362    }
363}
364
365#[cfg(test)]
366mod tests {
367    use std::sync::Arc;
368
369    use super::*;
370    use rocketmq_error::Error;
371    use rocketmq_error::TRANSPORT_CONNECTION_FAILED;
372
373    #[test]
374    fn mq_client_exception_matches_java_fields_and_faq_message() {
375        let mut exception = MQClientException::new_with_code(17, "client failed");
376
377        assert_eq!(exception.get_response_code(), 17);
378        assert_eq!(exception.get_error_message(), Some("client failed"));
379        assert!(exception.to_string().contains("CODE: 17  DESC: client failed"));
380        assert!(exception
381            .to_string()
382            .contains("For more information, please visit the url"));
383
384        exception.set_response_code(18).set_error_message("changed message");
385        assert_eq!(exception.get_response_code(), 18);
386        assert_eq!(exception.get_error_message(), Some("changed message"));
387        assert!(exception.to_string().contains("client failed"));
388    }
389
390    #[test]
391    fn mq_broker_exception_preserves_broker_addr_like_java() {
392        let exception = MQBrokerException::new_with_broker(25, "broker failed", "127.0.0.1:10911");
393
394        assert_eq!(exception.get_response_code(), 25);
395        assert_eq!(exception.get_error_message(), Some("broker failed"));
396        assert_eq!(exception.get_broker_addr(), Some("127.0.0.1:10911"));
397        assert!(exception.to_string().contains("BROKER: 127.0.0.1:10911"));
398
399        let error: ClientError = exception.into();
400        assert!(error.is(&rocketmq_error::BROKER_OPERATION_FAILED));
401        assert_eq!(error.broker_response_code(), Some(25));
402        assert_eq!(error.broker_addr(), Some("127.0.0.1:10911"));
403        let canonical = std::error::Error::source(&error).expect("canonical source");
404        assert!(canonical.downcast_ref::<rocketmq_error::Error>().is_some());
405        assert!(error.source_ref::<MQBrokerException>().is_some());
406    }
407
408    #[test]
409    fn mq_broker_exception_can_be_extracted_from_client_error() {
410        let error: ClientError = MQBrokerException::new_with_broker(31, "send failed", "broker-a").into();
411        let exception = MQBrokerException::from_client_error(&error).expect("broker exception");
412
413        assert_eq!(exception.get_response_code(), 31);
414        assert_eq!(exception.get_error_message(), Some("send failed"));
415        assert_eq!(exception.get_broker_addr(), Some("broker-a"));
416    }
417
418    #[test]
419    fn offset_not_found_exception_delegates_broker_fields() {
420        let exception = OffsetNotFoundException::new_with_broker(26, "offset not found", "127.0.0.1:10911");
421
422        assert_eq!(exception.get_response_code(), 26);
423        assert_eq!(exception.get_error_message(), Some("offset not found"));
424        assert_eq!(exception.get_broker_addr(), Some("127.0.0.1:10911"));
425        assert_eq!(
426            exception.as_broker_exception().get_broker_addr(),
427            Some("127.0.0.1:10911")
428        );
429    }
430
431    #[test]
432    fn request_timeout_exception_matches_java_message_without_faq() {
433        let mut exception = RequestTimeoutException::new_with_code(408, "request timeout");
434
435        assert_eq!(exception.get_response_code(), 408);
436        assert_eq!(exception.get_error_message(), Some("request timeout"));
437        assert_eq!(exception.to_string(), "CODE: 408  DESC: request timeout");
438
439        exception.set_response_code(409).set_error_message("changed timeout");
440        assert_eq!(exception.get_response_code(), 409);
441        assert_eq!(exception.get_error_message(), Some("changed timeout"));
442        assert_eq!(exception.to_string(), "CODE: 408  DESC: request timeout");
443    }
444
445    #[test]
446    fn request_timeout_exception_converts_to_timeout_error() {
447        let error: ClientError = RequestTimeoutException::new("request timeout").into();
448
449        assert!(error.is(&rocketmq_error::CORE_OPERATION_TIMED_OUT));
450    }
451
452    #[test]
453    fn request_timeout_classification_uses_the_canonical_shared_descriptor() {
454        for descriptor in [
455            &TRANSPORT_CONNECTION_TIMEOUT,
456            &TRANSPORT_WRITE_TIMEOUT,
457            &TRANSPORT_RESPONSE_TIMEOUT,
458        ] {
459            let canonical = Arc::new(Error::new(descriptor));
460            let shared = ClientError::from_shared(Arc::clone(&canonical));
461
462            assert!(RequestTimeoutException::from_client_error(&shared).is_some());
463            assert!(Arc::ptr_eq(&canonical, shared.shared_error()));
464        }
465    }
466
467    #[test]
468    fn timeout_word_in_non_timeout_source_does_not_change_classification() {
469        let canonical = Arc::new(Error::caused_by(
470            &TRANSPORT_CONNECTION_FAILED,
471            std::io::Error::other("rendered source contains timeout"),
472        ));
473        let shared = ClientError::from_shared(Arc::clone(&canonical));
474
475        assert!(RequestTimeoutException::from_client_error(&shared).is_none());
476        assert!(Arc::ptr_eq(&canonical, shared.shared_error()));
477    }
478}
rocketmq_client_rust/exception.rs

rocketmq_client_rust/
exception.rs
