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

rocketmq_protocol/protocol/header/
pull_message_request_header.rs
