rocketmq_protocol/protocol/header/
polling_info_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 cheetah_string::CheetahString;
16use rocketmq_macros::RequestHeaderCodec;
17use serde::Deserialize;
18use serde::Serialize;
19
20use crate::protocol::header::message_operation_header::TopicRequestHeaderTrait;
21use crate::protocol::header::namesrv::topic_operation_header::TopicRequestHeader;
22
23#[derive(Debug, Clone, Serialize, Deserialize, RequestHeaderCodec)]
24#[header(
25    type_id = "rocketmq_protocol::protocol::header::polling_info_request_header::PollingInfoRequestHeader",
26    java_class = "org.apache.rocketmq.remoting.protocol.header.PollingInfoRequestHeader"
27)]
28#[serde(rename_all = "camelCase")]
29pub struct PollingInfoRequestHeader {
30    #[header(required)]
31    pub consumer_group: CheetahString,
32
33    #[header(required)]
34    pub topic: CheetahString,
35
36    #[header(required)]
37    pub queue_id: i32,
38
39    #[serde(flatten)]
40    #[header(flatten, presence = "always")]
41    pub topic_request_header: Option<TopicRequestHeader>,
42}
43
44impl TopicRequestHeaderTrait for PollingInfoRequestHeader {
45    fn set_lo(&mut self, lo: Option<bool>) {
46        if let Some(header) = self.topic_request_header.as_mut() {
47            header.lo = lo;
48        }
49    }
50
51    fn lo(&self) -> Option<bool> {
52        self.topic_request_header.as_ref().and_then(|h| h.lo)
53    }
54
55    fn set_topic(&mut self, topic: CheetahString) {
56        self.topic = topic;
57    }
58
59    fn topic(&self) -> &CheetahString {
60        &self.topic
61    }
62
63    fn broker_name(&self) -> Option<&CheetahString> {
64        self.topic_request_header
65            .as_ref()
66            .and_then(|h| h.rpc.as_ref())
67            .and_then(|h| h.broker_name.as_ref())
68    }
69
70    fn set_broker_name(&mut self, broker_name: CheetahString) {
71        if let Some(header) = self.topic_request_header.as_mut() {
72            if let Some(rpc_header) = header.rpc.as_mut() {
73                rpc_header.broker_name = Some(broker_name);
74            }
75        }
76    }
77
78    fn namespace(&self) -> Option<&str> {
79        self.topic_request_header
80            .as_ref()
81            .and_then(|h| h.rpc.as_ref())
82            .and_then(|r| r.namespace.as_deref())
83    }
84
85    fn set_namespace(&mut self, namespace: CheetahString) {
86        if let Some(header) = self.topic_request_header.as_mut() {
87            if let Some(rpc_header) = header.rpc.as_mut() {
88                rpc_header.namespace = Some(namespace);
89            }
90        }
91    }
92
93    fn namespaced(&self) -> Option<bool> {
94        self.topic_request_header
95            .as_ref()
96            .and_then(|h| h.rpc.as_ref())
97            .and_then(|r| r.namespaced)
98    }
99
100    fn set_namespaced(&mut self, namespaced: bool) {
101        if let Some(header) = self.topic_request_header.as_mut() {
102            if let Some(rpc_header) = header.rpc.as_mut() {
103                rpc_header.namespaced = Some(namespaced);
104            }
105        }
106    }
107
108    fn oneway(&self) -> Option<bool> {
109        self.topic_request_header
110            .as_ref()
111            .and_then(|h| h.rpc.as_ref())
112            .and_then(|r| r.oneway)
113    }
114
115    fn set_oneway(&mut self, oneway: bool) {
116        if let Some(header) = self.topic_request_header.as_mut() {
117            if let Some(rpc_header) = header.rpc.as_mut() {
118                rpc_header.oneway = Some(oneway);
119            }
120        }
121    }
122
123    fn queue_id(&self) -> i32 {
124        self.queue_id
125    }
126
127    fn set_queue_id(&mut self, queue_id: i32) {
128        self.queue_id = queue_id;
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use std::collections::HashMap;
135
136    use super::*;
137    use crate::protocol::command_custom_header::CommandCustomHeader;
138    use crate::protocol::command_custom_header::FromMap;
139
140    #[test]
141    fn polling_info_request_header_serializes_correctly() {
142        let header = PollingInfoRequestHeader {
143            consumer_group: CheetahString::from_static_str("test_consumer_group"),
144            topic: CheetahString::from_static_str("test_topic"),
145            queue_id: 1,
146            topic_request_header: None,
147        };
148        let map = header.to_map().unwrap();
149        assert_eq!(
150            map.get(&CheetahString::from_static_str("consumerGroup")).unwrap(),
151            "test_consumer_group"
152        );
153        assert_eq!(map.get(&CheetahString::from_static_str("topic")).unwrap(), "test_topic");
154        assert_eq!(map.get(&CheetahString::from_static_str("queueId")).unwrap(), "1");
155    }
156
157    #[test]
158    fn polling_info_request_header_deserializes_correctly() {
159        let mut map = HashMap::new();
160        map.insert(
161            CheetahString::from_static_str("consumerGroup"),
162            CheetahString::from_static_str("test_consumer_group"),
163        );
164        map.insert(
165            CheetahString::from_static_str("topic"),
166            CheetahString::from_static_str("test_topic"),
167        );
168        map.insert(
169            CheetahString::from_static_str("queueId"),
170            CheetahString::from_static_str("1"),
171        );
172
173        let header = <PollingInfoRequestHeader as FromMap>::from(&map).unwrap();
174        assert_eq!(header.consumer_group, "test_consumer_group");
175        assert_eq!(header.topic, "test_topic");
176        assert_eq!(header.queue_id, 1);
177        assert!(header.topic_request_header.is_some());
178    }
179
180    #[test]
181    fn polling_info_request_header_handles_invalid_data() {
182        let mut map = HashMap::new();
183        map.insert(
184            CheetahString::from_static_str("consumerGroup"),
185            CheetahString::from_static_str("test_consumer_group"),
186        );
187        map.insert(
188            CheetahString::from_static_str("topic"),
189            CheetahString::from_static_str("test_topic"),
190        );
191        map.insert(
192            CheetahString::from_static_str("queueId"),
193            CheetahString::from_static_str("invalid"),
194        );
195
196        let result = <PollingInfoRequestHeader as FromMap>::from(&map);
197        assert!(result.is_err());
198    }
199
200    #[test]
201    fn topic_request_header_trait_ignores_nested_setters_without_envelope() {
202        let mut header = PollingInfoRequestHeader {
203            consumer_group: CheetahString::from_static_str("test_group"),
204            topic: CheetahString::from_static_str("test_topic"),
205            queue_id: 0,
206            topic_request_header: None,
207        };
208
209        header.set_lo(Some(true));
210        header.set_broker_name(CheetahString::from_static_str("new_broker"));
211        header.set_namespace(CheetahString::from_static_str("new_namespace"));
212        header.set_namespaced(true);
213        header.set_oneway(true);
214
215        assert_eq!(header.lo(), None);
216        assert_eq!(header.broker_name(), None);
217        assert_eq!(header.namespace(), None);
218        assert_eq!(header.namespaced(), None);
219        assert_eq!(header.oneway(), None);
220    }
221}
rocketmq_protocol/protocol/header/polling_info_request_header.rs

rocketmq_protocol/protocol/header/
polling_info_request_header.rs
