rocketmq_remoting/protocol/header/
change_invisible_time_request_header.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::Display;
16
17use cheetah_string::CheetahString;
18use rocketmq_macros::RequestHeaderCodecV2;
19use serde::Deserialize;
20use serde::Serialize;
21
22use crate::rpc::topic_request_header::TopicRequestHeader;
23
24#[derive(Serialize, Deserialize, Debug, Default, RequestHeaderCodecV2)]
25#[serde(rename_all = "camelCase")]
26pub struct ChangeInvisibleTimeRequestHeader {
27    #[required]
28    pub consumer_group: CheetahString,
29
30    #[required]
31    pub topic: CheetahString,
32
33    #[required]
34    pub queue_id: i32,
35
36    //startOffset popTime invisibleTime queueId
37    #[required]
38    pub extra_info: CheetahString,
39
40    #[required]
41    pub offset: i64,
42
43    #[required]
44    pub invisible_time: i64,
45
46    #[serde(skip_serializing_if = "Option::is_none")]
47    pub lite_topic: Option<CheetahString>,
48
49    pub suspend: bool,
50
51    #[serde(flatten)]
52    pub topic_request_header: Option<TopicRequestHeader>,
53}
54
55impl Display for ChangeInvisibleTimeRequestHeader {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        write!(
58            f,
59            "ChangeInvisibleTimeRequestHeader {{ consumer_group: {}, topic: {}, queue_id: {}, extra_info: {}, offset: \
60             {}, invisible_time: {}, lite_topic: {:?}, suspend: {} }}",
61            self.consumer_group,
62            self.topic,
63            self.queue_id,
64            self.extra_info,
65            self.offset,
66            self.invisible_time,
67            self.lite_topic,
68            self.suspend
69        )
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use cheetah_string::CheetahString;
76    use serde_json;
77
78    use super::*;
79    use crate::protocol::command_custom_header::CommandCustomHeader;
80    use crate::protocol::command_custom_header::FromMap;
81
82    #[test]
83    fn change_invisible_time_request_header_display_format() {
84        let header = ChangeInvisibleTimeRequestHeader {
85            consumer_group: CheetahString::from("group1"),
86            topic: CheetahString::from("topic1"),
87            queue_id: 1,
88            extra_info: CheetahString::from("info"),
89            offset: 12345,
90            invisible_time: 67890,
91            lite_topic: None,
92            suspend: false,
93            topic_request_header: None,
94        };
95        assert_eq!(
96            format!("{}", header),
97            "ChangeInvisibleTimeRequestHeader { consumer_group: group1, topic: topic1, queue_id: 1, extra_info: info, \
98             offset: 12345, invisible_time: 67890, lite_topic: None, suspend: false }"
99        );
100    }
101
102    #[test]
103    fn change_invisible_time_request_header_display_format_with_topic_request_header() {
104        let header = ChangeInvisibleTimeRequestHeader {
105            consumer_group: CheetahString::from("group1"),
106            topic: CheetahString::from("topic1"),
107            queue_id: 1,
108            extra_info: CheetahString::from("info"),
109            offset: 12345,
110            invisible_time: 67890,
111            lite_topic: Some(CheetahString::from("lite_topic")),
112            suspend: true,
113            topic_request_header: Some(TopicRequestHeader {
114                rpc_request_header: None,
115                lo: None,
116            }),
117        };
118        assert_eq!(
119            format!("{}", header),
120            "ChangeInvisibleTimeRequestHeader { consumer_group: group1, topic: topic1, queue_id: 1, extra_info: info, \
121             offset: 12345, invisible_time: 67890, lite_topic: Some(\"lite_topic\"), suspend: true }"
122        );
123    }
124
125    #[test]
126    fn change_invisible_time_request_header_serialize() {
127        let header = ChangeInvisibleTimeRequestHeader {
128            consumer_group: CheetahString::from("group1"),
129            topic: CheetahString::from("topic1"),
130            queue_id: 1,
131            extra_info: CheetahString::from("info"),
132            offset: 12345,
133            invisible_time: 67890,
134            lite_topic: Some(CheetahString::from("lite_topic")),
135            suspend: true,
136            topic_request_header: None,
137        };
138        let serialized = serde_json::to_string(&header).unwrap();
139        assert_eq!(
140            serialized,
141            r#"{"consumerGroup":"group1","topic":"topic1","queueId":1,"extraInfo":"info","offset":12345,"invisibleTime":67890,"liteTopic":"lite_topic","suspend":true}"#
142        );
143    }
144
145    #[test]
146    fn change_invisible_time_request_header_deserialize() {
147        let json = r#"{"consumerGroup":"group1","topic":"topic1","queueId":1,"extraInfo":"info","offset":12345,"invisibleTime":67890,"liteTopic":"lite_topic","suspend":true}"#;
148        let header: ChangeInvisibleTimeRequestHeader = serde_json::from_str(json).unwrap();
149        assert_eq!(header.consumer_group, CheetahString::from("group1"));
150        assert_eq!(header.topic, CheetahString::from("topic1"));
151        assert_eq!(header.queue_id, 1);
152        assert_eq!(header.extra_info, CheetahString::from("info"));
153        assert_eq!(header.offset, 12345);
154        assert_eq!(header.invisible_time, 67890);
155        assert_eq!(header.lite_topic.as_deref(), Some("lite_topic"));
156        assert!(header.suspend);
157    }
158
159    #[test]
160    fn change_invisible_time_request_header_maps_lite_topic_and_suspend_like_java_header() {
161        let header = ChangeInvisibleTimeRequestHeader {
162            consumer_group: CheetahString::from("group1"),
163            topic: CheetahString::from("topic1"),
164            queue_id: 1,
165            extra_info: CheetahString::from("info"),
166            offset: 12345,
167            invisible_time: 67890,
168            lite_topic: Some(CheetahString::from("lite_topic")),
169            suspend: true,
170            topic_request_header: None,
171        };
172
173        let map = header.to_map().unwrap();
174        assert_eq!(map.get("liteTopic").map(|value| value.as_str()), Some("lite_topic"));
175        assert_eq!(map.get("suspend").map(|value| value.as_str()), Some("true"));
176
177        let decoded = <ChangeInvisibleTimeRequestHeader as FromMap>::from(&map).unwrap();
178        assert_eq!(decoded.lite_topic.as_deref(), Some("lite_topic"));
179        assert!(decoded.suspend);
180    }
181}
rocketmq_remoting/protocol/header/change_invisible_time_request_header.rs

rocketmq_remoting/protocol/header/
change_invisible_time_request_header.rs
