rocketmq_remoting/protocol/header/
pull_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 cheetah_string::CheetahString;
16use rocketmq_macros::RequestHeaderCodecV2;
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, Deserialize, Serialize, Default, RequestHeaderCodecV2)]
24#[serde(rename_all = "camelCase")]
25pub struct PullMessageRequestHeader {
26    #[required]
27    pub consumer_group: CheetahString,
28
29    #[required]
30    pub topic: CheetahString,
31
32    pub lite_topic: Option<CheetahString>,
33
34    #[required]
35    pub queue_id: i32,
36
37    #[required]
38    pub queue_offset: i64,
39
40    #[required]
41    pub max_msg_nums: i32,
42
43    #[required]
44    pub sys_flag: i32,
45
46    #[required]
47    pub commit_offset: i64,
48
49    #[required]
50    pub suspend_timeout_millis: u64,
51
52    #[required]
53    pub sub_version: i64,
54
55    pub subscription: Option<CheetahString>,
56    pub expression_type: Option<CheetahString>,
57    pub max_msg_bytes: Option<i32>,
58    pub request_source: Option<i32>,
59    #[serde(rename = "proxyFrowardClientId", alias = "proxyForwardClientId")]
60    pub proxy_forward_client_id: Option<CheetahString>,
61    #[serde(flatten)]
62    pub topic_request: Option<TopicRequestHeader>,
63}
64
65impl TopicRequestHeaderTrait for PullMessageRequestHeader {
66    fn set_lo(&mut self, lo: Option<bool>) {
67        if let Some(header) = self.topic_request.as_mut() {
68            header.lo = lo;
69        }
70    }
71
72    fn lo(&self) -> Option<bool> {
73        self.topic_request.as_ref().and_then(|h| h.lo)
74    }
75
76    fn set_topic(&mut self, topic: CheetahString) {
77        self.topic = topic;
78    }
79
80    fn topic(&self) -> &CheetahString {
81        &self.topic
82    }
83
84    fn broker_name(&self) -> Option<&CheetahString> {
85        self.topic_request
86            .as_ref()
87            .and_then(|h| h.rpc.as_ref())
88            .and_then(|h| h.broker_name.as_ref())
89    }
90
91    fn set_broker_name(&mut self, broker_name: CheetahString) {
92        if let Some(header) = self.topic_request.as_mut() {
93            if let Some(rpc_header) = header.rpc.as_mut() {
94                rpc_header.broker_name = Some(broker_name);
95            }
96        }
97    }
98
99    fn namespace(&self) -> Option<&str> {
100        self.topic_request
101            .as_ref()
102            .and_then(|h| h.rpc.as_ref())
103            .and_then(|r| r.namespace.as_deref())
104    }
105
106    fn set_namespace(&mut self, namespace: CheetahString) {
107        if let Some(header) = self.topic_request.as_mut() {
108            if let Some(rpc_header) = header.rpc.as_mut() {
109                rpc_header.namespace = Some(namespace);
110            }
111        }
112    }
113
114    fn namespaced(&self) -> Option<bool> {
115        self.topic_request
116            .as_ref()
117            .and_then(|h| h.rpc.as_ref())
118            .and_then(|r| r.namespaced)
119    }
120
121    fn set_namespaced(&mut self, namespaced: bool) {
122        if let Some(header) = self.topic_request.as_mut() {
123            if let Some(rpc_header) = header.rpc.as_mut() {
124                rpc_header.namespaced = Some(namespaced);
125            }
126        }
127    }
128
129    fn oneway(&self) -> Option<bool> {
130        self.topic_request
131            .as_ref()
132            .and_then(|h| h.rpc.as_ref())
133            .and_then(|r| r.oneway)
134    }
135
136    fn set_oneway(&mut self, oneway: bool) {
137        if let Some(header) = self.topic_request.as_mut() {
138            if let Some(rpc_header) = header.rpc.as_mut() {
139                rpc_header.oneway = Some(oneway);
140            }
141        }
142    }
143
144    fn queue_id(&self) -> i32 {
145        self.queue_id
146    }
147
148    fn set_queue_id(&mut self, queue_id: i32) {
149        self.queue_id = queue_id;
150    }
151}
152
153#[cfg(test)]
154mod tests {
155    use std::collections::HashMap;
156
157    use cheetah_string::CheetahString;
158
159    use super::*;
160    use crate::protocol::command_custom_header::CommandCustomHeader;
161    use crate::protocol::command_custom_header::FromMap;
162
163    #[test]
164    fn pull_message_request_header_serializes_correctly() {
165        let header = PullMessageRequestHeader {
166            consumer_group: CheetahString::from_static_str("test_consumer_group"),
167            topic: CheetahString::from_static_str("test_topic"),
168            lite_topic: Some(CheetahString::from_static_str("test_lite_topic")),
169            queue_id: 1,
170            queue_offset: 100,
171            max_msg_nums: 10,
172            sys_flag: 0,
173            commit_offset: 50,
174            suspend_timeout_millis: 3000,
175            subscription: Some(CheetahString::from_static_str("test_subscription")),
176            sub_version: 1,
177            expression_type: Some(CheetahString::from_static_str("test_expression")),
178            max_msg_bytes: Some(1024),
179            request_source: Some(1),
180            proxy_forward_client_id: Some(CheetahString::from_static_str("test_client_id")),
181            topic_request: None,
182        };
183        let map: HashMap<CheetahString, CheetahString> = header.to_map().unwrap();
184        assert_eq!(
185            map.get(&CheetahString::from_static_str("consumerGroup")).unwrap(),
186            "test_consumer_group"
187        );
188        assert_eq!(map.get(&CheetahString::from_static_str("topic")).unwrap(), "test_topic");
189        assert_eq!(
190            map.get(&CheetahString::from_static_str("liteTopic")).unwrap(),
191            "test_lite_topic"
192        );
193        assert_eq!(map.get(&CheetahString::from_static_str("queueId")).unwrap(), "1");
194        assert_eq!(map.get(&CheetahString::from_static_str("queueOffset")).unwrap(), "100");
195        assert_eq!(map.get(&CheetahString::from_static_str("maxMsgNums")).unwrap(), "10");
196        assert_eq!(map.get(&CheetahString::from_static_str("sysFlag")).unwrap(), "0");
197        assert_eq!(map.get(&CheetahString::from_static_str("commitOffset")).unwrap(), "50");
198        assert_eq!(
199            map.get(&CheetahString::from_static_str("suspendTimeoutMillis"))
200                .unwrap(),
201            "3000"
202        );
203        assert_eq!(
204            map.get(&CheetahString::from_static_str("subscription")).unwrap(),
205            "test_subscription"
206        );
207        assert_eq!(map.get(&CheetahString::from_static_str("subVersion")).unwrap(), "1");
208        assert_eq!(
209            map.get(&CheetahString::from_static_str("expressionType")).unwrap(),
210            "test_expression"
211        );
212        assert_eq!(map.get(&CheetahString::from_static_str("maxMsgBytes")).unwrap(), "1024");
213        assert_eq!(map.get(&CheetahString::from_static_str("requestSource")).unwrap(), "1");
214        assert_eq!(
215            map.get(&CheetahString::from_static_str("proxyFrowardClientId"))
216                .unwrap(),
217            "test_client_id"
218        );
219    }
220
221    #[test]
222    fn pull_message_request_header_deserializes_correctly() {
223        let mut map = HashMap::new();
224        map.insert(
225            CheetahString::from_static_str("consumerGroup"),
226            CheetahString::from_static_str("test_consumer_group"),
227        );
228        map.insert(
229            CheetahString::from_static_str("topic"),
230            CheetahString::from_static_str("test_topic"),
231        );
232        map.insert(
233            CheetahString::from_static_str("liteTopic"),
234            CheetahString::from_static_str("test_lite_topic"),
235        );
236        map.insert(
237            CheetahString::from_static_str("queueId"),
238            CheetahString::from_static_str("1"),
239        );
240        map.insert(
241            CheetahString::from_static_str("queueOffset"),
242            CheetahString::from_static_str("100"),
243        );
244        map.insert(
245            CheetahString::from_static_str("maxMsgNums"),
246            CheetahString::from_static_str("10"),
247        );
248        map.insert(
249            CheetahString::from_static_str("sysFlag"),
250            CheetahString::from_static_str("0"),
251        );
252        map.insert(
253            CheetahString::from_static_str("commitOffset"),
254            CheetahString::from_static_str("50"),
255        );
256        map.insert(
257            CheetahString::from_static_str("suspendTimeoutMillis"),
258            CheetahString::from_static_str("3000"),
259        );
260        map.insert(
261            CheetahString::from_static_str("subscription"),
262            CheetahString::from_static_str("test_subscription"),
263        );
264        map.insert(
265            CheetahString::from_static_str("subVersion"),
266            CheetahString::from_static_str("1"),
267        );
268        map.insert(
269            CheetahString::from_static_str("expressionType"),
270            CheetahString::from_static_str("test_expression"),
271        );
272        map.insert(
273            CheetahString::from_static_str("maxMsgBytes"),
274            CheetahString::from_static_str("1024"),
275        );
276        map.insert(
277            CheetahString::from_static_str("requestSource"),
278            CheetahString::from_static_str("1"),
279        );
280        map.insert(
281            CheetahString::from_static_str("proxyFrowardClientId"),
282            CheetahString::from_static_str("test_client_id"),
283        );
284
285        let header = <PullMessageRequestHeader as FromMap>::from(&map).unwrap();
286        assert_eq!(header.consumer_group, "test_consumer_group");
287        assert_eq!(header.topic, "test_topic");
288        assert_eq!(header.lite_topic.unwrap(), "test_lite_topic");
289        assert_eq!(header.queue_id, 1);
290        assert_eq!(header.queue_offset, 100);
291        assert_eq!(header.max_msg_nums, 10);
292        assert_eq!(header.sys_flag, 0);
293        assert_eq!(header.commit_offset, 50);
294        assert_eq!(header.suspend_timeout_millis, 3000);
295        assert_eq!(header.subscription.unwrap(), "test_subscription");
296        assert_eq!(header.sub_version, 1);
297        assert_eq!(header.expression_type.unwrap(), "test_expression");
298        assert_eq!(header.max_msg_bytes.unwrap(), 1024);
299        assert_eq!(header.request_source.unwrap(), 1);
300        assert_eq!(header.proxy_forward_client_id.unwrap(), "test_client_id");
301    }
302
303    #[test]
304    fn pull_message_request_header_accepts_legacy_corrected_proxy_forward_key() {
305        let mut map = HashMap::new();
306        map.insert(
307            CheetahString::from_static_str("consumerGroup"),
308            CheetahString::from_static_str("test_consumer_group"),
309        );
310        map.insert(
311            CheetahString::from_static_str("topic"),
312            CheetahString::from_static_str("test_topic"),
313        );
314        map.insert(
315            CheetahString::from_static_str("liteTopic"),
316            CheetahString::from_static_str("test_lite_topic"),
317        );
318        map.insert(
319            CheetahString::from_static_str("queueId"),
320            CheetahString::from_static_str("1"),
321        );
322        map.insert(
323            CheetahString::from_static_str("queueOffset"),
324            CheetahString::from_static_str("100"),
325        );
326        map.insert(
327            CheetahString::from_static_str("maxMsgNums"),
328            CheetahString::from_static_str("10"),
329        );
330        map.insert(
331            CheetahString::from_static_str("sysFlag"),
332            CheetahString::from_static_str("0"),
333        );
334        map.insert(
335            CheetahString::from_static_str("commitOffset"),
336            CheetahString::from_static_str("50"),
337        );
338        map.insert(
339            CheetahString::from_static_str("suspendTimeoutMillis"),
340            CheetahString::from_static_str("3000"),
341        );
342        map.insert(
343            CheetahString::from_static_str("subVersion"),
344            CheetahString::from_static_str("1"),
345        );
346        map.insert(
347            CheetahString::from_static_str("proxyForwardClientId"),
348            CheetahString::from_static_str("legacy_client_id"),
349        );
350
351        let header = <PullMessageRequestHeader as FromMap>::from(&map).unwrap();
352
353        assert_eq!(header.proxy_forward_client_id.unwrap(), "legacy_client_id");
354    }
355
356    #[test]
357    fn pull_message_request_header_handles_missing_optional_fields() {
358        let mut map = HashMap::new();
359        map.insert(
360            CheetahString::from_static_str("consumerGroup"),
361            CheetahString::from_static_str("test_consumer_group"),
362        );
363        map.insert(
364            CheetahString::from_static_str("topic"),
365            CheetahString::from_static_str("test_topic"),
366        );
367        map.insert(
368            CheetahString::from_static_str("queueId"),
369            CheetahString::from_static_str("1"),
370        );
371        map.insert(
372            CheetahString::from_static_str("queueOffset"),
373            CheetahString::from_static_str("100"),
374        );
375        map.insert(
376            CheetahString::from_static_str("maxMsgNums"),
377            CheetahString::from_static_str("10"),
378        );
379        map.insert(
380            CheetahString::from_static_str("sysFlag"),
381            CheetahString::from_static_str("0"),
382        );
383        map.insert(
384            CheetahString::from_static_str("commitOffset"),
385            CheetahString::from_static_str("50"),
386        );
387        map.insert(
388            CheetahString::from_static_str("suspendTimeoutMillis"),
389            CheetahString::from_static_str("3000"),
390        );
391        map.insert(
392            CheetahString::from_static_str("subVersion"),
393            CheetahString::from_static_str("1"),
394        );
395
396        let header = <PullMessageRequestHeader as FromMap>::from(&map).unwrap();
397        assert_eq!(header.consumer_group, "test_consumer_group");
398        assert_eq!(header.topic, "test_topic");
399        assert!(header.lite_topic.is_none());
400        assert_eq!(header.queue_id, 1);
401        assert_eq!(header.queue_offset, 100);
402        assert_eq!(header.max_msg_nums, 10);
403        assert_eq!(header.sys_flag, 0);
404        assert_eq!(header.commit_offset, 50);
405        assert_eq!(header.suspend_timeout_millis, 3000);
406        assert_eq!(header.sub_version, 1);
407        assert!(header.subscription.is_none());
408        assert!(header.expression_type.is_none());
409        assert!(header.max_msg_bytes.is_none());
410        assert!(header.request_source.is_none());
411        assert!(header.proxy_forward_client_id.is_none());
412    }
413
414    #[test]
415    fn pull_message_request_header_handles_invalid_data() {
416        let mut map = HashMap::new();
417        map.insert(
418            CheetahString::from_static_str("consumerGroup"),
419            CheetahString::from_static_str("test_consumer_group"),
420        );
421        map.insert(
422            CheetahString::from_static_str("topic"),
423            CheetahString::from_static_str("test_topic"),
424        );
425        map.insert(
426            CheetahString::from_static_str("queueId"),
427            CheetahString::from_static_str("invalid"),
428        );
429        map.insert(
430            CheetahString::from_static_str("queueOffset"),
431            CheetahString::from_static_str("invalid"),
432        );
433        map.insert(
434            CheetahString::from_static_str("maxMsgNums"),
435            CheetahString::from_static_str("invalid"),
436        );
437        map.insert(
438            CheetahString::from_static_str("sysFlag"),
439            CheetahString::from_static_str("invalid"),
440        );
441        map.insert(
442            CheetahString::from_static_str("commitOffset"),
443            CheetahString::from_static_str("invalid"),
444        );
445        map.insert(
446            CheetahString::from_static_str("suspendTimeoutMillis"),
447            CheetahString::from_static_str("invalid"),
448        );
449        map.insert(
450            CheetahString::from_static_str("subVersion"),
451            CheetahString::from_static_str("invalid"),
452        );
453
454        let result = <PullMessageRequestHeader as FromMap>::from(&map);
455        assert!(result.is_err());
456    }
457}
rocketmq_remoting/protocol/header/pull_message_request_header.rs

rocketmq_remoting/protocol/header/
pull_message_request_header.rs
