rocketmq_protocol/protocol/
topic.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 rocketmq_model::message::MessageQueue;
16use serde::Deserialize;
17use serde::Serialize;
18
19#[derive(Serialize, Deserialize, Debug)]
20pub struct OffsetMovedEvent {
21    pub consumer_group: String,
22    pub message_queue: MessageQueue,
23    pub offset_request: i64,
24    pub offset_new: i64,
25}
26
27impl OffsetMovedEvent {
28    pub fn get_consumer_group(&self) -> &str {
29        &self.consumer_group
30    }
31
32    pub fn set_consumer_group(&mut self, consumer_group: String) {
33        self.consumer_group = consumer_group;
34    }
35
36    pub fn get_message_queue(&self) -> &MessageQueue {
37        &self.message_queue
38    }
39
40    pub fn set_message_queue(&mut self, message_queue: MessageQueue) {
41        self.message_queue = message_queue;
42    }
43
44    pub fn get_offset_request(&self) -> i64 {
45        self.offset_request
46    }
47
48    pub fn set_offset_request(&mut self, offset_request: i64) {
49        self.offset_request = offset_request;
50    }
51
52    pub fn get_offset_new(&self) -> i64 {
53        self.offset_new
54    }
55
56    pub fn set_offset_new(&mut self, offset_new: i64) {
57        self.offset_new = offset_new;
58    }
59}
60
61impl std::fmt::Display for OffsetMovedEvent {
62    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        write!(
64            f,
65            "OffsetMovedEvent [consumer_group={}, message_queue={:?}, offset_request={}, offset_new={}]",
66            self.consumer_group, self.message_queue, self.offset_request, self.offset_new
67        )
68    }
69}
70
71#[cfg(test)]
72mod tests {
73    use super::*;
74
75    #[test]
76    fn offset_moved_event_field_initialization() {
77        let body: OffsetMovedEvent = OffsetMovedEvent {
78            consumer_group: "test_group".to_string(),
79            message_queue: MessageQueue::new(),
80            offset_request: 100,
81            offset_new: 200,
82        };
83        assert_eq!(body.get_consumer_group(), "test_group");
84        assert_eq!(body.get_offset_request(), 100);
85        assert_eq!(body.get_offset_new(), 200);
86        assert_eq!(body.get_message_queue(), &MessageQueue::new());
87    }
88
89    #[test]
90    fn offset_moved_event_setters() {
91        let mut body = OffsetMovedEvent {
92            consumer_group: "test_group".to_string(),
93            message_queue: MessageQueue::new(),
94            offset_request: 100,
95            offset_new: 200,
96        };
97        body.set_consumer_group("new_group".to_string());
98        body.set_message_queue(MessageQueue::new());
99        body.set_offset_request(150);
100        body.set_offset_new(250);
101        assert_eq!(body.get_consumer_group(), "new_group");
102        assert_eq!(body.get_message_queue(), &MessageQueue::new());
103        assert_eq!(body.get_offset_request(), 150);
104        assert_eq!(body.get_offset_new(), 250);
105    }
106
107    #[test]
108    fn offset_moved_event_getters() {
109        let body = OffsetMovedEvent {
110            consumer_group: "another_group".to_string(),
111            message_queue: MessageQueue::new(),
112            offset_request: 300,
113            offset_new: 400,
114        };
115        assert_eq!(body.get_consumer_group(), "another_group");
116        assert_eq!(body.get_offset_request(), 300);
117        assert_eq!(body.get_offset_new(), 400);
118    }
119
120    #[test]
121    fn offset_moved_event_display() {
122        let body = OffsetMovedEvent {
123            consumer_group: "test_group".to_string(),
124            message_queue: MessageQueue::new(),
125            offset_request: 100,
126            offset_new: 200,
127        };
128        let display = format!("{}", body);
129        assert!(display.contains("OffsetMovedEvent"));
130        assert!(display.contains("consumer_group=test_group"));
131        assert!(display.contains("offset_request=100"));
132        assert!(display.contains("offset_new=200"));
133    }
134    #[test]
135    fn offset_moved_event_serialization() {
136        let body = OffsetMovedEvent {
137            consumer_group: "test_group".to_string(),
138            message_queue: MessageQueue::new(),
139            offset_request: 100,
140            offset_new: 200,
141        };
142        let json = serde_json::to_string(&body).unwrap();
143        assert!(json.contains("\"consumer_group\":\"test_group\""));
144        assert!(json.contains("\"offset_request\":100"));
145        assert!(json.contains("\"offset_new\":200"));
146    }
147
148    #[test]
149    fn offset_moved_event_deserialization() {
150        let json = r#"{"consumer_group":"test_group","message_queue":{"topic":"","brokerName":"","queueId":0},"offset_request":100,"offset_new":200}"#;
151        let body: OffsetMovedEvent = serde_json::from_str(json).unwrap();
152        assert_eq!(body.get_consumer_group(), "test_group");
153        assert_eq!(body.get_offset_request(), 100);
154        assert_eq!(body.get_offset_new(), 200);
155        assert_eq!(body.get_message_queue(), &MessageQueue::new());
156    }
157}
rocketmq_protocol/protocol/topic.rs

rocketmq_protocol/protocol/
topic.rs
