rocketmq_remoting/protocol/header/
pop_lite_message_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_common::TimeUtils::current_millis;
19use rocketmq_macros::RequestHeaderCodecV2;
20use serde::Deserialize;
21use serde::Serialize;
22
23use crate::rpc::rpc_request_header::RpcRequestHeader;
24
25#[derive(Clone, Debug, Serialize, Deserialize, RequestHeaderCodecV2)]
26#[serde(rename_all = "camelCase")]
27pub struct PopLiteMessageRequestHeader {
28    #[required]
29    pub client_id: CheetahString,
30    #[required]
31    pub consumer_group: CheetahString,
32    #[required]
33    pub topic: CheetahString,
34    #[required]
35    pub max_msg_num: i32,
36    #[required]
37    pub invisible_time: i64,
38    #[required]
39    pub poll_time: i64,
40    #[required]
41    pub born_time: i64,
42    pub attempt_id: Option<CheetahString>,
43
44    #[serde(flatten)]
45    pub rpc: Option<RpcRequestHeader>,
46}
47
48impl PopLiteMessageRequestHeader {
49    pub fn is_timeout_too_much(&self) -> bool {
50        current_millis() as i64 - self.born_time - self.poll_time > 500
51    }
52}
53
54impl Default for PopLiteMessageRequestHeader {
55    fn default() -> Self {
56        Self {
57            client_id: CheetahString::new(),
58            consumer_group: CheetahString::new(),
59            topic: CheetahString::new(),
60            max_msg_num: 0,
61            invisible_time: 0,
62            poll_time: 0,
63            born_time: 0,
64            attempt_id: None,
65            rpc: None,
66        }
67    }
68}
69
70impl Display for PopLiteMessageRequestHeader {
71    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72        write!(
73            f,
74            "PopLiteMessageRequestHeader [client_id={}, consumer_group={}, topic={}, max_msg_num={}, \
75             invisible_time={}, poll_time={}, born_time={}, attempt_id={}]",
76            self.client_id,
77            self.consumer_group,
78            self.topic,
79            self.max_msg_num,
80            self.invisible_time,
81            self.poll_time,
82            self.born_time,
83            self.attempt_id.as_ref().unwrap_or(&CheetahString::new())
84        )
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use std::collections::HashMap;
91
92    use super::*;
93    use crate::protocol::command_custom_header::CommandCustomHeader;
94    use crate::protocol::command_custom_header::FromMap;
95
96    #[test]
97    fn pop_lite_message_request_header_serializes_to_map() {
98        let header = PopLiteMessageRequestHeader {
99            client_id: "client".into(),
100            consumer_group: "group".into(),
101            topic: "topic".into(),
102            max_msg_num: 32,
103            invisible_time: 1000,
104            poll_time: 3000,
105            born_time: current_millis() as i64,
106            attempt_id: Some("attempt".into()),
107            rpc: None,
108        };
109
110        let map = header.to_map().unwrap();
111        assert_eq!(map.get(&CheetahString::from_static_str("clientId")).unwrap(), "client");
112        assert_eq!(
113            map.get(&CheetahString::from_static_str("consumerGroup")).unwrap(),
114            "group"
115        );
116        assert_eq!(map.get(&CheetahString::from_static_str("topic")).unwrap(), "topic");
117        assert_eq!(map.get(&CheetahString::from_static_str("maxMsgNum")).unwrap(), "32");
118    }
119
120    #[test]
121    fn pop_lite_message_request_header_deserializes_from_map() {
122        let mut map = HashMap::new();
123        map.insert(
124            CheetahString::from_static_str("clientId"),
125            CheetahString::from_static_str("client"),
126        );
127        map.insert(
128            CheetahString::from_static_str("consumerGroup"),
129            CheetahString::from_static_str("group"),
130        );
131        map.insert(
132            CheetahString::from_static_str("topic"),
133            CheetahString::from_static_str("topic"),
134        );
135        map.insert(
136            CheetahString::from_static_str("maxMsgNum"),
137            CheetahString::from_static_str("16"),
138        );
139        map.insert(
140            CheetahString::from_static_str("invisibleTime"),
141            CheetahString::from_static_str("1000"),
142        );
143        map.insert(
144            CheetahString::from_static_str("pollTime"),
145            CheetahString::from_static_str("3000"),
146        );
147        map.insert(
148            CheetahString::from_static_str("bornTime"),
149            CheetahString::from_static_str("100"),
150        );
151
152        let header = <PopLiteMessageRequestHeader as FromMap>::from(&map).unwrap();
153        assert_eq!(header.client_id, "client");
154        assert_eq!(header.consumer_group, "group");
155        assert_eq!(header.topic, "topic");
156        assert_eq!(header.max_msg_num, 16);
157        assert_eq!(header.invisible_time, 1000);
158        assert_eq!(header.poll_time, 3000);
159        assert_eq!(header.born_time, 100);
160    }
161}
rocketmq_remoting/protocol/header/pop_lite_message_request_header.rs

rocketmq_remoting/protocol/header/
pop_lite_message_request_header.rs
