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 rocketmq_common::common::FAQUrl;
18use rocketmq_error::RocketMQError;
19
20const DEFAULT_RESPONSE_CODE: i32 = -1;
21const DEFAULT_BROKER_RESPONSE_CODE: i32 = 0;
22
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct MQClientException {
25    response_code: i32,
26    error_message: Option<String>,
27    message: String,
28}
29
30impl MQClientException {
31    pub fn new(error_message: impl Into<String>) -> Self {
32        let error_message = error_message.into();
33        let message = FAQUrl::attach_default_url(Some(error_message.as_str()));
34        Self {
35            response_code: DEFAULT_RESPONSE_CODE,
36            error_message: Some(error_message),
37            message,
38        }
39    }
40
41    pub fn new_with_code(response_code: i32, error_message: impl Into<String>) -> Self {
42        let error_message = error_message.into();
43        let message =
44            FAQUrl::attach_default_url(Some(format!("CODE: {response_code}  DESC: {error_message}").as_str()));
45        Self {
46            response_code,
47            error_message: Some(error_message),
48            message,
49        }
50    }
51
52    pub fn response_code(&self) -> i32 {
53        self.response_code
54    }
55
56    pub fn get_response_code(&self) -> i32 {
57        self.response_code()
58    }
59
60    pub fn set_response_code(&mut self, response_code: i32) -> &mut Self {
61        self.response_code = response_code;
62        self
63    }
64
65    pub fn error_message(&self) -> Option<&str> {
66        self.error_message.as_deref()
67    }
68
69    pub fn get_error_message(&self) -> Option<&str> {
70        self.error_message()
71    }
72
73    pub fn set_error_message(&mut self, error_message: impl Into<String>) {
74        self.error_message = Some(error_message.into());
75    }
76
77    pub fn from_rocketmq_error(error: &RocketMQError) -> Self {
78        Self::new(error.to_string())
79    }
80}
81
82impl fmt::Display for MQClientException {
83    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
84        f.write_str(self.message.as_str())
85    }
86}
87
88impl std::error::Error for MQClientException {}
89
90impl From<MQClientException> for RocketMQError {
91    fn from(exception: MQClientException) -> Self {
92        RocketMQError::illegal_argument(exception.to_string())
93    }
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct MQBrokerException {
98    response_code: i32,
99    error_message: Option<String>,
100    broker_addr: Option<String>,
101    message: String,
102}
103
104impl Default for MQBrokerException {
105    fn default() -> Self {
106        Self {
107            response_code: DEFAULT_BROKER_RESPONSE_CODE,
108            error_message: None,
109            broker_addr: None,
110            message: String::new(),
111        }
112    }
113}
114
115impl MQBrokerException {
116    pub fn new(response_code: i32, error_message: impl Into<String>) -> Self {
117        Self::new_with_optional_broker(response_code, error_message, None)
118    }
119
120    pub fn new_with_broker(
121        response_code: i32,
122        error_message: impl Into<String>,
123        broker_addr: impl Into<String>,
124    ) -> Self {
125        Self::new_with_optional_broker(response_code, error_message, Some(broker_addr.into()))
126    }
127
128    pub fn new_with_optional_broker(
129        response_code: i32,
130        error_message: impl Into<String>,
131        broker_addr: Option<String>,
132    ) -> Self {
133        let error_message = error_message.into();
134        let broker_suffix = broker_addr
135            .as_ref()
136            .map(|broker_addr| format!(" BROKER: {broker_addr}"))
137            .unwrap_or_default();
138        let message = FAQUrl::attach_default_url(Some(
139            format!("CODE: {response_code}  DESC: {error_message}{broker_suffix}").as_str(),
140        ));
141        Self {
142            response_code,
143            error_message: Some(error_message),
144            broker_addr,
145            message,
146        }
147    }
148
149    pub fn response_code(&self) -> i32 {
150        self.response_code
151    }
152
153    pub fn get_response_code(&self) -> i32 {
154        self.response_code()
155    }
156
157    pub fn error_message(&self) -> Option<&str> {
158        self.error_message.as_deref()
159    }
160
161    pub fn get_error_message(&self) -> Option<&str> {
162        self.error_message()
163    }
164
165    pub fn broker_addr(&self) -> Option<&str> {
166        self.broker_addr.as_deref()
167    }
168
169    pub fn get_broker_addr(&self) -> Option<&str> {
170        self.broker_addr()
171    }
172
173    pub fn from_rocketmq_error(error: &RocketMQError) -> Option<Self> {
174        if let RocketMQError::BrokerOperationFailed {
175            code,
176            message,
177            broker_addr,
178            ..
179        } = error
180        {
181            Some(Self::new_with_optional_broker(
182                *code,
183                message.as_str(),
184                broker_addr.clone(),
185            ))
186        } else {
187            None
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 RocketMQError {
201    fn from(exception: MQBrokerException) -> Self {
202        let message = exception
203            .error_message
204            .clone()
205            .unwrap_or_else(|| exception.message.clone());
206        let error = RocketMQError::broker_operation_failed("BROKER_OPERATION", exception.response_code, message);
207        if let Some(broker_addr) = exception.broker_addr {
208            error.with_broker_addr(broker_addr)
209        } else {
210            error
211        }
212    }
213}
214
215#[derive(Debug, Clone, Default, PartialEq, Eq)]
216pub struct OffsetNotFoundException {
217    broker_exception: MQBrokerException,
218}
219
220impl OffsetNotFoundException {
221    pub fn new(response_code: i32, error_message: impl Into<String>) -> Self {
222        Self {
223            broker_exception: MQBrokerException::new(response_code, error_message),
224        }
225    }
226
227    pub fn new_with_broker(
228        response_code: i32,
229        error_message: impl Into<String>,
230        broker_addr: impl Into<String>,
231    ) -> Self {
232        Self {
233            broker_exception: MQBrokerException::new_with_broker(response_code, error_message, broker_addr),
234        }
235    }
236
237    pub fn response_code(&self) -> i32 {
238        self.broker_exception.response_code()
239    }
240
241    pub fn get_response_code(&self) -> i32 {
242        self.response_code()
243    }
244
245    pub fn error_message(&self) -> Option<&str> {
246        self.broker_exception.error_message()
247    }
248
249    pub fn get_error_message(&self) -> Option<&str> {
250        self.error_message()
251    }
252
253    pub fn broker_addr(&self) -> Option<&str> {
254        self.broker_exception.broker_addr()
255    }
256
257    pub fn get_broker_addr(&self) -> Option<&str> {
258        self.broker_addr()
259    }
260
261    pub fn as_broker_exception(&self) -> &MQBrokerException {
262        &self.broker_exception
263    }
264}
265
266impl fmt::Display for OffsetNotFoundException {
267    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
268        self.broker_exception.fmt(f)
269    }
270}
271
272impl std::error::Error for OffsetNotFoundException {}
273
274impl From<OffsetNotFoundException> for MQBrokerException {
275    fn from(exception: OffsetNotFoundException) -> Self {
276        exception.broker_exception
277    }
278}
279
280impl From<OffsetNotFoundException> for RocketMQError {
281    fn from(exception: OffsetNotFoundException) -> Self {
282        let broker_exception: MQBrokerException = exception.into();
283        broker_exception.into()
284    }
285}
286
287#[derive(Debug, Clone, PartialEq, Eq)]
288pub struct RequestTimeoutException {
289    response_code: i32,
290    error_message: Option<String>,
291    message: String,
292}
293
294impl RequestTimeoutException {
295    pub fn new(error_message: impl Into<String>) -> Self {
296        let error_message = error_message.into();
297        Self {
298            response_code: DEFAULT_RESPONSE_CODE,
299            message: error_message.clone(),
300            error_message: Some(error_message),
301        }
302    }
303
304    pub fn new_with_code(response_code: i32, error_message: impl Into<String>) -> Self {
305        let error_message = error_message.into();
306        let message = format!("CODE: {response_code}  DESC: {error_message}");
307        Self {
308            response_code,
309            error_message: Some(error_message),
310            message,
311        }
312    }
313
314    pub fn response_code(&self) -> i32 {
315        self.response_code
316    }
317
318    pub fn get_response_code(&self) -> i32 {
319        self.response_code()
320    }
321
322    pub fn set_response_code(&mut self, response_code: i32) -> &mut Self {
323        self.response_code = response_code;
324        self
325    }
326
327    pub fn error_message(&self) -> Option<&str> {
328        self.error_message.as_deref()
329    }
330
331    pub fn get_error_message(&self) -> Option<&str> {
332        self.error_message()
333    }
334
335    pub fn set_error_message(&mut self, error_message: impl Into<String>) {
336        self.error_message = Some(error_message.into());
337    }
338
339    pub fn from_rocketmq_error(error: &RocketMQError) -> Option<Self> {
340        match error {
341            RocketMQError::Timeout { .. } => Some(Self::new(error.to_string())),
342            RocketMQError::Network(network_error) if network_error.to_string().contains("timeout") => {
343                Some(Self::new(error.to_string()))
344            }
345            _ => None,
346        }
347    }
348}
349
350impl fmt::Display for RequestTimeoutException {
351    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
352        f.write_str(self.message.as_str())
353    }
354}
355
356impl std::error::Error for RequestTimeoutException {}
357
358impl From<RequestTimeoutException> for RocketMQError {
359    fn from(_exception: RequestTimeoutException) -> Self {
360        RocketMQError::Timeout {
361            operation: "REQUEST",
362            timeout_ms: 0,
363        }
364    }
365}
366
367#[cfg(test)]
368mod tests {
369    use super::*;
370
371    #[test]
372    fn mq_client_exception_matches_java_fields_and_faq_message() {
373        let mut exception = MQClientException::new_with_code(17, "client failed");
374
375        assert_eq!(exception.get_response_code(), 17);
376        assert_eq!(exception.get_error_message(), Some("client failed"));
377        assert!(exception.to_string().contains("CODE: 17  DESC: client failed"));
378        assert!(exception
379            .to_string()
380            .contains("For more information, please visit the url"));
381
382        exception.set_response_code(18).set_error_message("changed message");
383        assert_eq!(exception.get_response_code(), 18);
384        assert_eq!(exception.get_error_message(), Some("changed message"));
385        assert!(exception.to_string().contains("client failed"));
386    }
387
388    #[test]
389    fn mq_broker_exception_preserves_broker_addr_like_java() {
390        let exception = MQBrokerException::new_with_broker(25, "broker failed", "127.0.0.1:10911");
391
392        assert_eq!(exception.get_response_code(), 25);
393        assert_eq!(exception.get_error_message(), Some("broker failed"));
394        assert_eq!(exception.get_broker_addr(), Some("127.0.0.1:10911"));
395        assert!(exception.to_string().contains("BROKER: 127.0.0.1:10911"));
396
397        let error: RocketMQError = exception.into();
398        match error {
399            RocketMQError::BrokerOperationFailed {
400                code,
401                message,
402                broker_addr,
403                ..
404            } => {
405                assert_eq!(code, 25);
406                assert_eq!(message, "broker failed");
407                assert_eq!(broker_addr.as_deref(), Some("127.0.0.1:10911"));
408            }
409            other => panic!("expected broker operation error, got {other:?}"),
410        }
411    }
412
413    #[test]
414    fn mq_broker_exception_can_be_extracted_from_rocketmq_error() {
415        let error = RocketMQError::broker_operation_failed("SEND", 31, "send failed").with_broker_addr("broker-a");
416        let exception = MQBrokerException::from_rocketmq_error(&error).expect("broker exception");
417
418        assert_eq!(exception.get_response_code(), 31);
419        assert_eq!(exception.get_error_message(), Some("send failed"));
420        assert_eq!(exception.get_broker_addr(), Some("broker-a"));
421    }
422
423    #[test]
424    fn offset_not_found_exception_delegates_broker_fields() {
425        let exception = OffsetNotFoundException::new_with_broker(26, "offset not found", "127.0.0.1:10911");
426
427        assert_eq!(exception.get_response_code(), 26);
428        assert_eq!(exception.get_error_message(), Some("offset not found"));
429        assert_eq!(exception.get_broker_addr(), Some("127.0.0.1:10911"));
430        assert_eq!(
431            exception.as_broker_exception().get_broker_addr(),
432            Some("127.0.0.1:10911")
433        );
434    }
435
436    #[test]
437    fn request_timeout_exception_matches_java_message_without_faq() {
438        let mut exception = RequestTimeoutException::new_with_code(408, "request timeout");
439
440        assert_eq!(exception.get_response_code(), 408);
441        assert_eq!(exception.get_error_message(), Some("request timeout"));
442        assert_eq!(exception.to_string(), "CODE: 408  DESC: request timeout");
443
444        exception.set_response_code(409).set_error_message("changed timeout");
445        assert_eq!(exception.get_response_code(), 409);
446        assert_eq!(exception.get_error_message(), Some("changed timeout"));
447        assert_eq!(exception.to_string(), "CODE: 408  DESC: request timeout");
448    }
449
450    #[test]
451    fn request_timeout_exception_converts_to_timeout_error() {
452        let error: RocketMQError = RequestTimeoutException::new("request timeout").into();
453
454        match error {
455            RocketMQError::Timeout { operation, timeout_ms } => {
456                assert_eq!(operation, "REQUEST");
457                assert_eq!(timeout_ms, 0);
458            }
459            other => panic!("expected timeout error, got {other:?}"),
460        }
461    }
462}
rocketmq_client_rust/exception.rs

rocketmq_client_rust/
exception.rs
