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

rocketmq_protocol/protocol/header/
pop_lite_message_request_header.rs
