rocketmq_model/
result.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 cheetah_string::CheetahString;
18use serde::Deserialize;
19use serde::Deserializer;
20use serde::Serialize;
21use serde::Serializer;
22
23use crate::message::MessageQueue;
24
25/// Outcome reported by a producer send operation.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
27pub enum SendStatus {
28    #[default]
29    SendOk,
30    FlushDiskTimeout,
31    FlushSlaveTimeout,
32    SlaveNotAvailable,
33}
34
35impl SendStatus {
36    const fn as_str(self) -> &'static str {
37        match self {
38            Self::SendOk => "SEND_OK",
39            Self::FlushDiskTimeout => "FLUSH_DISK_TIMEOUT",
40            Self::FlushSlaveTimeout => "FLUSH_SLAVE_TIMEOUT",
41            Self::SlaveNotAvailable => "SLAVE_NOT_AVAILABLE",
42        }
43    }
44}
45
46impl Serialize for SendStatus {
47    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
48    where
49        S: Serializer,
50    {
51        serializer.serialize_str(self.as_str())
52    }
53}
54
55impl<'de> Deserialize<'de> for SendStatus {
56    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
57    where
58        D: Deserializer<'de>,
59    {
60        let value = String::deserialize(deserializer)?;
61        match value.as_str() {
62            "SEND_OK" => Ok(Self::SendOk),
63            "FLUSH_DISK_TIMEOUT" => Ok(Self::FlushDiskTimeout),
64            "FLUSH_SLAVE_TIMEOUT" => Ok(Self::FlushSlaveTimeout),
65            "SLAVE_NOT_AVAILABLE" => Ok(Self::SlaveNotAvailable),
66            _ => Err(serde::de::Error::unknown_variant(
67                &value,
68                &[
69                    "SEND_OK",
70                    "FLUSH_DISK_TIMEOUT",
71                    "FLUSH_SLAVE_TIMEOUT",
72                    "SLAVE_NOT_AVAILABLE",
73                ],
74            )),
75        }
76    }
77}
78
79impl fmt::Display for SendStatus {
80    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
81        f.write_str(self.as_str())
82    }
83}
84
85/// Producer send result without client lifecycle state.
86#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
87pub struct SendResult {
88    pub send_status: SendStatus,
89    pub msg_id: Option<CheetahString>,
90    pub message_queue: Option<MessageQueue>,
91    pub queue_offset: u64,
92    pub transaction_id: Option<String>,
93    pub offset_msg_id: Option<String>,
94    pub recall_handle: Option<String>,
95    pub region_id: Option<String>,
96    pub trace_on: bool,
97    pub raw_resp_body: Option<Vec<u8>>,
98}
99
100impl Default for SendResult {
101    fn default() -> Self {
102        Self {
103            send_status: SendStatus::SendOk,
104            msg_id: None,
105            message_queue: None,
106            queue_offset: 0,
107            transaction_id: None,
108            offset_msg_id: None,
109            recall_handle: None,
110            region_id: None,
111            trace_on: true,
112            raw_resp_body: None,
113        }
114    }
115}
116
117impl SendResult {
118    pub fn new(
119        send_status: SendStatus,
120        msg_id: Option<CheetahString>,
121        offset_msg_id: Option<String>,
122        message_queue: Option<MessageQueue>,
123        queue_offset: u64,
124    ) -> Self {
125        Self {
126            send_status,
127            msg_id,
128            message_queue,
129            queue_offset,
130            offset_msg_id,
131            ..Self::default()
132        }
133    }
134
135    pub fn new_with_additional_fields(
136        send_status: SendStatus,
137        msg_id: Option<CheetahString>,
138        message_queue: Option<MessageQueue>,
139        queue_offset: u64,
140        transaction_id: Option<String>,
141        offset_msg_id: Option<String>,
142        region_id: Option<String>,
143    ) -> Self {
144        Self {
145            send_status,
146            msg_id,
147            message_queue,
148            queue_offset,
149            transaction_id,
150            offset_msg_id,
151            region_id,
152            ..Self::default()
153        }
154    }
155
156    pub fn encoder_send_result_to_json<T: Serialize>(value: &T) -> serde_json::Result<String> {
157        serde_json::to_string(value)
158    }
159
160    pub fn decoder_send_result_from_json(json: &str) -> serde_json::Result<Self> {
161        serde_json::from_str(json)
162    }
163
164    pub fn is_trace_on(&self) -> bool {
165        self.trace_on
166    }
167
168    pub fn set_trace_on(&mut self, trace_on: bool) {
169        self.trace_on = trace_on;
170    }
171
172    pub fn get_region_id(&self) -> Option<&str> {
173        self.region_id.as_deref()
174    }
175
176    pub fn set_region_id(&mut self, region_id: String) {
177        self.region_id = Some(region_id);
178    }
179
180    pub fn get_msg_id(&self) -> Option<&CheetahString> {
181        self.msg_id.as_ref()
182    }
183
184    pub fn set_msg_id(&mut self, msg_id: CheetahString) {
185        self.msg_id = Some(msg_id);
186    }
187
188    pub fn get_send_status(&self) -> SendStatus {
189        self.send_status
190    }
191
192    pub fn set_send_status(&mut self, send_status: SendStatus) {
193        self.send_status = send_status;
194    }
195
196    pub fn get_message_queue(&self) -> Option<&MessageQueue> {
197        self.message_queue.as_ref()
198    }
199
200    pub fn set_message_queue(&mut self, message_queue: MessageQueue) {
201        self.message_queue = Some(message_queue);
202    }
203
204    pub fn get_queue_offset(&self) -> u64 {
205        self.queue_offset
206    }
207
208    pub fn set_queue_offset(&mut self, queue_offset: u64) {
209        self.queue_offset = queue_offset;
210    }
211
212    pub fn get_transaction_id(&self) -> Option<&str> {
213        self.transaction_id.as_deref()
214    }
215
216    pub fn set_transaction_id(&mut self, transaction_id: String) {
217        self.transaction_id = Some(transaction_id);
218    }
219
220    pub fn get_offset_msg_id(&self) -> Option<&str> {
221        self.offset_msg_id.as_deref()
222    }
223
224    pub fn set_offset_msg_id(&mut self, offset_msg_id: String) {
225        self.offset_msg_id = Some(offset_msg_id);
226    }
227
228    pub fn set_recall_handle(&mut self, recall_handle: String) {
229        self.recall_handle = Some(recall_handle);
230    }
231
232    pub fn recall_handle(&self) -> Option<&str> {
233        self.recall_handle.as_deref()
234    }
235
236    pub fn get_recall_handle(&self) -> Option<&str> {
237        self.recall_handle()
238    }
239
240    pub fn set_raw_resp_body(&mut self, body: Vec<u8>) {
241        self.raw_resp_body = Some(body);
242    }
243
244    pub fn get_raw_resp_body(&self) -> Option<&[u8]> {
245        self.raw_resp_body.as_deref()
246    }
247}
248
249impl fmt::Display for SendResult {
250    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
251        fn java_option<T: fmt::Display>(value: Option<&T>) -> String {
252            value.map_or_else(|| "null".to_owned(), ToString::to_string)
253        }
254
255        write!(
256            f,
257            "SendResult [sendStatus={}, msgId={}, offsetMsgId={}, messageQueue={}, queueOffset={}, recallHandle={}]",
258            self.send_status,
259            java_option(self.msg_id.as_ref()),
260            java_option(self.offset_msg_id.as_ref()),
261            java_option(self.message_queue.as_ref()),
262            self.queue_offset,
263            java_option(self.recall_handle.as_ref()),
264        )
265    }
266}
267
268/// Query result parameterized by its owned message representation.
269#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
270pub struct QueryResult<T> {
271    index_last_update_timestamp: u64,
272    message_list: Vec<T>,
273}
274
275impl<T> QueryResult<T> {
276    pub fn new(index_last_update_timestamp: u64, message_list: Vec<T>) -> Self {
277        Self {
278            index_last_update_timestamp,
279            message_list,
280        }
281    }
282
283    pub fn index_last_update_timestamp(&self) -> u64 {
284        self.index_last_update_timestamp
285    }
286
287    pub fn message_list(&self) -> &Vec<T> {
288        &self.message_list
289    }
290}
291
292impl<T: fmt::Display> fmt::Display for QueryResult<T> {
293    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
294        let messages = self
295            .message_list
296            .iter()
297            .map(ToString::to_string)
298            .collect::<Vec<_>>()
299            .join(", ");
300        write!(
301            f,
302            "QueryResult [indexLastUpdateTimestamp={}, messageList=[{}]]",
303            self.index_last_update_timestamp, messages
304        )
305    }
306}
307
308/// Consumer pull status.
309#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
310pub enum PullStatus {
311    #[default]
312    Found,
313    NoNewMsg,
314    NoMatchedMsg,
315    OffsetIllegal,
316}
317
318impl From<i32> for PullStatus {
319    fn from(value: i32) -> Self {
320        match value {
321            1 => Self::NoNewMsg,
322            2 => Self::NoMatchedMsg,
323            3 => Self::OffsetIllegal,
324            _ => Self::Found,
325        }
326    }
327}
328
329impl From<PullStatus> for i32 {
330    fn from(value: PullStatus) -> Self {
331        match value {
332            PullStatus::Found => 0,
333            PullStatus::NoNewMsg => 1,
334            PullStatus::NoMatchedMsg => 2,
335            PullStatus::OffsetIllegal => 3,
336        }
337    }
338}
339
340impl fmt::Display for PullStatus {
341    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
342        f.write_str(match self {
343            Self::Found => "FOUND",
344            Self::NoNewMsg => "NO_NEW_MSG",
345            Self::NoMatchedMsg => "NO_MATCHED_MSG",
346            Self::OffsetIllegal => "OFFSET_ILLEGAL",
347        })
348    }
349}
350
351/// Immutable pull result that owns its message values.
352#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
353pub struct PullOutcome<T> {
354    pull_status: PullStatus,
355    next_begin_offset: u64,
356    min_offset: u64,
357    max_offset: u64,
358    messages: Option<Vec<T>>,
359}
360
361impl<T> PullOutcome<T> {
362    pub fn new<M>(
363        pull_status: PullStatus,
364        next_begin_offset: u64,
365        min_offset: u64,
366        max_offset: u64,
367        messages: M,
368    ) -> Self
369    where
370        M: Into<Option<Vec<T>>>,
371    {
372        Self {
373            pull_status,
374            next_begin_offset,
375            min_offset,
376            max_offset,
377            messages: messages.into(),
378        }
379    }
380
381    pub fn pull_status(&self) -> PullStatus {
382        self.pull_status
383    }
384
385    pub fn next_begin_offset(&self) -> u64 {
386        self.next_begin_offset
387    }
388
389    pub fn min_offset(&self) -> u64 {
390        self.min_offset
391    }
392
393    pub fn max_offset(&self) -> u64 {
394        self.max_offset
395    }
396
397    pub fn messages(&self) -> Option<&[T]> {
398        self.messages.as_deref()
399    }
400
401    pub fn into_messages(self) -> Option<Vec<T>> {
402        self.messages
403    }
404}
rocketmq_model/result.rs

rocketmq_model/
result.rs
