rocketmq_model/common/message/
message_queue_for_c.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::cmp::Ordering;
16use std::fmt;
17use std::hash::Hash;
18use std::hash::Hasher;
19
20use cheetah_string::CheetahString;
21use serde::Deserialize;
22use serde::Serialize;
23
24use crate::message::MessageQueue;
25
26/// MessageQueue structure for C++ client compatibility.
27/// Includes offset field that the standard MessageQueue doesn't have.
28#[derive(Clone, Debug, Serialize, Deserialize)]
29#[serde(rename_all = "camelCase")]
30pub struct MessageQueueForC {
31    pub topic: CheetahString,
32    pub broker_name: CheetahString,
33    pub queue_id: i32,
34    pub offset: i64,
35}
36
37impl MessageQueueForC {
38    pub fn new(
39        topic: impl Into<CheetahString>,
40        broker_name: impl Into<CheetahString>,
41        queue_id: i32,
42        offset: i64,
43    ) -> Self {
44        Self {
45            topic: topic.into(),
46            broker_name: broker_name.into(),
47            queue_id,
48            offset,
49        }
50    }
51
52    /// Create from MessageQueue with offset
53    pub fn from_message_queue(mq: &MessageQueue, offset: i64) -> Self {
54        Self {
55            topic: mq.topic_str().into(),
56            broker_name: mq.broker_name().into(),
57            queue_id: mq.queue_id(),
58            offset,
59        }
60    }
61}
62
63impl Default for MessageQueueForC {
64    fn default() -> Self {
65        Self {
66            topic: CheetahString::empty(),
67            broker_name: CheetahString::empty(),
68            queue_id: 0,
69            offset: 0,
70        }
71    }
72}
73
74impl PartialEq for MessageQueueForC {
75    fn eq(&self, other: &Self) -> bool {
76        self.topic == other.topic
77            && self.broker_name == other.broker_name
78            && self.queue_id == other.queue_id
79            && self.offset == other.offset
80    }
81}
82
83impl Eq for MessageQueueForC {}
84
85impl Hash for MessageQueueForC {
86    fn hash<H: Hasher>(&self, state: &mut H) {
87        self.topic.hash(state);
88        self.broker_name.hash(state);
89        self.queue_id.hash(state);
90    }
91}
92
93impl Ord for MessageQueueForC {
94    fn cmp(&self, other: &Self) -> Ordering {
95        match self.topic.cmp(&other.topic) {
96            Ordering::Equal => {}
97            ord => return ord,
98        }
99        match self.broker_name.cmp(&other.broker_name) {
100            Ordering::Equal => {}
101            ord => return ord,
102        }
103        match self.queue_id.cmp(&other.queue_id) {
104            Ordering::Equal => {}
105            ord => return ord,
106        }
107        self.offset.cmp(&other.offset)
108    }
109}
110
111impl PartialOrd for MessageQueueForC {
112    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
113        Some(self.cmp(other))
114    }
115}
116
117impl fmt::Display for MessageQueueForC {
118    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
119        write!(
120            f,
121            "MessageQueueForC [topic={}, brokerName={}, queueId={}, offset={}]",
122            self.topic, self.broker_name, self.queue_id, self.offset
123        )
124    }
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130
131    #[test]
132    fn test_message_queue_for_c_new() {
133        let mq = MessageQueueForC::new("test_topic", "broker-a", 1, 100);
134        assert_eq!(mq.topic.as_str(), "test_topic");
135        assert_eq!(mq.broker_name.as_str(), "broker-a");
136        assert_eq!(mq.queue_id, 1);
137        assert_eq!(mq.offset, 100);
138    }
139
140    #[test]
141    fn test_message_queue_for_c_ordering() {
142        let mq1 = MessageQueueForC::new("topic_a", "broker", 1, 100);
143        let mq2 = MessageQueueForC::new("topic_b", "broker", 1, 100);
144        assert!(mq1 < mq2);
145
146        let mq3 = MessageQueueForC::new("topic_a", "broker", 1, 200);
147        assert!(mq1 < mq3);
148    }
149
150    #[test]
151    fn test_message_queue_for_c_serialization() {
152        let mq = MessageQueueForC::new("test_topic", "broker-a", 1, 100);
153        let json = serde_json::to_string(&mq).unwrap();
154        assert!(json.contains("\"topic\":\"test_topic\""));
155        assert!(json.contains("\"offset\":100"));
156
157        let deserialized: MessageQueueForC = serde_json::from_str(&json).unwrap();
158        assert_eq!(mq, deserialized);
159    }
160}
rocketmq_model/common/message/message_queue_for_c.rs

rocketmq_model/common/message/
message_queue_for_c.rs
