rocketmq_protocol/protocol/header/
pop_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::protocol::header::namesrv::topic_operation_header::TopicRequestHeader;
23
24#[derive(Clone, Debug, Serialize, Deserialize, RequestHeaderCodec)]
25#[header(
26    type_id = "rocketmq_protocol::protocol::header::pop_message_request_header::PopMessageRequestHeader",
27    java_class = "org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader"
28)]
29#[serde(rename_all = "camelCase")]
30pub struct PopMessageRequestHeader {
31    #[header(required)]
32    pub consumer_group: CheetahString,
33    #[header(required)]
34    pub topic: CheetahString,
35    #[header(required)]
36    pub queue_id: i32,
37    #[header(required, range = "i32")]
38    pub max_msg_nums: u32,
39    #[header(required, range = "i64")]
40    pub invisible_time: u64,
41    #[header(required, range = "i64")]
42    pub poll_time: u64,
43    #[header(required, range = "i64")]
44    pub born_time: u64,
45    #[header(required)]
46    pub init_mode: i32,
47    pub exp_type: Option<CheetahString>,
48    pub exp: Option<CheetahString>,
49    #[serde(default = "default_order")]
50    #[header(default_with = "default_order", default_semantic = "literal:false")]
51    pub order: Option<bool>,
52    pub attempt_id: Option<CheetahString>,
53
54    #[serde(flatten)]
55    #[header(flatten, presence = "always")]
56    pub topic_request_header: Option<TopicRequestHeader>,
57}
58
59fn default_order() -> Option<bool> {
60    Some(false)
61}
62
63impl PopMessageRequestHeader {
64    pub fn is_timeout_too_much_at(&self, now_millis: i64) -> bool {
65        now_millis - self.born_time as i64 - self.poll_time as i64 > 500
66    }
67}
68
69impl Default for PopMessageRequestHeader {
70    fn default() -> Self {
71        PopMessageRequestHeader {
72            consumer_group: CheetahString::new(),
73            topic: CheetahString::new(),
74            queue_id: 0,
75            max_msg_nums: 0,
76            invisible_time: 0,
77            poll_time: 0,
78            born_time: 0,
79            init_mode: 0,
80            exp_type: None,
81            exp: None,
82            order: Some(false),
83            attempt_id: None,
84            topic_request_header: None,
85        }
86    }
87}
88
89impl Display for PopMessageRequestHeader {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        write!(
92            f,
93            "PopMessageRequestHeader [consumer_group={}, topic={}, queue_id={}, max_msg_nums={}, invisible_time={}, \
94             poll_time={}, born_time={}, init_mode={}, exp_type={}, exp={}, order={}, attempt_id={}]",
95            self.consumer_group,
96            self.topic,
97            self.queue_id,
98            self.max_msg_nums,
99            self.invisible_time,
100            self.poll_time,
101            self.born_time,
102            self.init_mode,
103            self.exp_type.as_ref().unwrap_or(&CheetahString::new()),
104            self.exp.as_ref().unwrap_or(&CheetahString::new()),
105            self.order.as_ref().unwrap_or(&false),
106            self.attempt_id.as_ref().unwrap_or(&CheetahString::new())
107        )
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use cheetah_string::CheetahString;
114
115    use super::*;
116
117    #[test]
118    fn default_pop_message_request_header() {
119        let header = PopMessageRequestHeader::default();
120        assert_eq!(header.consumer_group, CheetahString::new());
121        assert_eq!(header.topic, CheetahString::new());
122        assert_eq!(header.queue_id, 0);
123        assert_eq!(header.max_msg_nums, 0);
124        assert_eq!(header.invisible_time, 0);
125        assert_eq!(header.poll_time, 0);
126        assert_eq!(header.born_time, 0);
127        assert_eq!(header.init_mode, 0);
128        assert!(header.exp_type.is_none());
129        assert!(header.exp.is_none());
130        assert_eq!(header.order, Some(false));
131        assert!(header.attempt_id.is_none());
132        assert!(header.topic_request_header.is_none());
133    }
134
135    #[test]
136    fn display_pop_message_request_header() {
137        let header = PopMessageRequestHeader {
138            consumer_group: CheetahString::from("group1"),
139            topic: CheetahString::from("topic1"),
140            queue_id: 1,
141            max_msg_nums: 10,
142            invisible_time: 1000,
143            poll_time: 2000,
144            born_time: 3000,
145            init_mode: 1,
146            exp_type: Some(CheetahString::from("type1")),
147            exp: Some(CheetahString::from("exp1")),
148            order: Some(true),
149            attempt_id: Some(CheetahString::from("attempt1")),
150            topic_request_header: None,
151        };
152        assert_eq!(
153            format!("{}", header),
154            "PopMessageRequestHeader [consumer_group=group1, topic=topic1, queue_id=1, max_msg_nums=10, \
155             invisible_time=1000, poll_time=2000, born_time=3000, init_mode=1, exp_type=type1, exp=exp1, order=true, \
156             attempt_id=attempt1]"
157        );
158    }
159
160    #[test]
161    fn display_pop_message_request_header_with_defaults() {
162        let header = PopMessageRequestHeader::default();
163        assert_eq!(
164            format!("{}", header),
165            "PopMessageRequestHeader [consumer_group=, topic=, queue_id=0, max_msg_nums=0, invisible_time=0, \
166             poll_time=0, born_time=0, init_mode=0, exp_type=, exp=, order=false, attempt_id=]"
167        );
168    }
169}
rocketmq_protocol/protocol/header/pop_message_request_header.rs

rocketmq_protocol/protocol/header/
pop_message_request_header.rs
