rocketmq_protocol/protocol/header/message_operation_header/
send_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::code::request_code::RequestCode;
21use crate::protocol::header::message_operation_header::send_message_request_header_v2::SendMessageRequestHeaderV2;
22use crate::protocol::header::message_operation_header::TopicRequestHeaderTrait;
23use crate::protocol::remoting_command::RemotingCommand;
24use crate::rpc::rpc_request_header::RpcRequestHeader;
25use crate::rpc::topic_request_header::TopicRequestHeader;
26
27#[derive(Debug, Clone, Serialize, Deserialize, Default, RequestHeaderCodec)]
28#[header(
29    type_id = "rocketmq_protocol::protocol::header::message_operation_header::send_message_request_header::SendMessageRequestHeader",
30    java_class = "org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader",
31    lookup = "scan",
32    fast
33)]
34#[serde(rename_all = "camelCase")]
35pub struct SendMessageRequestHeader {
36    #[header(required)]
37    pub producer_group: CheetahString,
38
39    #[header(required)]
40    pub topic: CheetahString,
41
42    #[header(required)]
43    pub default_topic: CheetahString,
44
45    #[header(required)]
46    pub default_topic_queue_nums: i32,
47
48    #[header(required)]
49    pub queue_id: i32,
50
51    #[header(required)]
52    pub sys_flag: i32,
53
54    #[header(required)]
55    pub born_timestamp: i64,
56
57    #[header(required)]
58    pub flag: i32,
59
60    pub properties: Option<CheetahString>,
61    pub reconsume_times: Option<i32>,
62    pub unit_mode: Option<bool>,
63    pub batch: Option<bool>,
64    pub max_reconsume_times: Option<i32>,
65    #[serde(flatten)]
66    #[header(flatten, presence = "always")]
67    pub topic_request_header: Option<TopicRequestHeader>,
68}
69
70impl SendMessageRequestHeader {
71    #[inline(always)]
72    pub fn is_batch(&self) -> bool {
73        self.batch.unwrap_or_default()
74    }
75
76    #[inline]
77    fn topic_request_header_mut(&mut self) -> &mut TopicRequestHeader {
78        self.topic_request_header
79            .get_or_insert_with(TopicRequestHeader::default)
80    }
81
82    #[inline]
83    fn rpc_request_header_mut(&mut self) -> &mut RpcRequestHeader {
84        self.topic_request_header_mut()
85            .rpc_request_header
86            .get_or_insert_with(RpcRequestHeader::default)
87    }
88}
89
90impl TopicRequestHeaderTrait for SendMessageRequestHeader {
91    fn set_lo(&mut self, lo: Option<bool>) {
92        self.topic_request_header_mut().lo = lo;
93    }
94
95    fn lo(&self) -> Option<bool> {
96        match self.topic_request_header {
97            None => None,
98            Some(ref value) => value.lo,
99        }
100    }
101
102    fn set_topic(&mut self, topic: CheetahString) {
103        self.topic = topic;
104    }
105
106    fn topic(&self) -> &CheetahString {
107        &self.topic
108    }
109
110    fn broker_name(&self) -> Option<&CheetahString> {
111        self.topic_request_header
112            .as_ref()?
113            .rpc_request_header
114            .as_ref()?
115            .broker_name
116            .as_ref()
117    }
118
119    fn set_broker_name(&mut self, broker_name: CheetahString) {
120        self.rpc_request_header_mut().broker_name = Some(broker_name);
121    }
122
123    fn namespace(&self) -> Option<&str> {
124        self.topic_request_header
125            .as_ref()?
126            .rpc_request_header
127            .as_ref()?
128            .namespace
129            .as_deref()
130    }
131
132    fn set_namespace(&mut self, namespace: CheetahString) {
133        self.rpc_request_header_mut().namespace = Some(namespace);
134    }
135
136    fn namespaced(&self) -> Option<bool> {
137        self.topic_request_header
138            .as_ref()?
139            .rpc_request_header
140            .as_ref()?
141            .namespaced
142            .as_ref()
143            .cloned()
144    }
145
146    fn set_namespaced(&mut self, namespaced: bool) {
147        self.rpc_request_header_mut().namespaced = Some(namespaced);
148    }
149
150    fn oneway(&self) -> Option<bool> {
151        self.topic_request_header
152            .as_ref()?
153            .rpc_request_header
154            .as_ref()?
155            .oneway
156            .as_ref()
157            .cloned()
158    }
159
160    fn set_oneway(&mut self, oneway: bool) {
161        self.rpc_request_header_mut().oneway = Some(oneway);
162    }
163
164    fn queue_id(&self) -> i32 {
165        self.queue_id
166    }
167
168    fn set_queue_id(&mut self, queue_id: i32) {
169        self.queue_id = queue_id;
170    }
171}
172
173/// Parses the request header from a `RemotingCommand` based on the `RequestCode`.
174///
175/// This function attempts to decode the command custom header from the provided `RemotingCommand`.
176/// If the `RequestCode` is `SendMessageV2` or `SendBatchMessage`, it tries to decode the header
177/// as `SendMessageRequestHeaderV2`. If successful, it converts the `V2` header to a `V1` header.
178/// Otherwise, it decodes the header directly as `SendMessageRequestHeader`.
179///
180/// # Arguments
181///
182/// * `request` - A reference to the `RemotingCommand` containing the command and its headers.
183/// * `request_code` - The `RequestCode` that indicates which version of the request header to
184///   expect.
185///
186/// # Returns
187///
188/// * `Ok(SendMessageRequestHeader)` if the header is successfully parsed and converted (if
189///   necessary).
190/// * `Err(crate::Error)` if there is an error in decoding the header.
191#[inline]
192pub fn parse_request_header(
193    request: &RemotingCommand,
194    request_code: RequestCode,
195) -> rocketmq_error::Result<SendMessageRequestHeader> {
196    let mut request_header_v2 = None;
197    if RequestCode::SendMessageV2 == request_code || RequestCode::SendBatchMessage == request_code {
198        // Attempt to decode the command custom header as SendMessageRequestHeaderV2
199        request_header_v2 = request
200            .decode_command_custom_header_fast::<SendMessageRequestHeaderV2>()
201            .ok();
202    }
203    // Match on the result of the V2 header decoding
204    match request_header_v2 {
205        Some(header) => Ok(SendMessageRequestHeaderV2::create_send_message_request_header_v1(
206            &header,
207        )),
208        None => request.decode_command_custom_header_fast::<SendMessageRequestHeader>(),
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use std::collections::HashMap;
215
216    use cheetah_string::CheetahString;
217
218    use super::*;
219    use crate::code::request_code::RequestCode;
220    use crate::protocol::command_custom_header::CommandCustomHeader;
221    use crate::protocol::command_custom_header::FromMap;
222    use crate::protocol::remoting_command::RemotingCommand;
223
224    fn minimal_header() -> SendMessageRequestHeader {
225        SendMessageRequestHeader {
226            producer_group: CheetahString::from_static_str("test_producer_group"),
227            topic: CheetahString::from_static_str("test_topic"),
228            default_topic: CheetahString::from_static_str("test_default_topic"),
229            default_topic_queue_nums: 8,
230            queue_id: 1,
231            sys_flag: 0,
232            born_timestamp: 1622547800000,
233            flag: 0,
234            properties: None,
235            reconsume_times: None,
236            unit_mode: None,
237            batch: None,
238            max_reconsume_times: None,
239            topic_request_header: None,
240        }
241    }
242
243    fn assert_same_send_message_header(left: &SendMessageRequestHeader, right: &SendMessageRequestHeader) {
244        assert_eq!(left.producer_group, right.producer_group);
245        assert_eq!(left.topic, right.topic);
246        assert_eq!(left.default_topic, right.default_topic);
247        assert_eq!(left.default_topic_queue_nums, right.default_topic_queue_nums);
248        assert_eq!(left.queue_id, right.queue_id);
249        assert_eq!(left.sys_flag, right.sys_flag);
250        assert_eq!(left.born_timestamp, right.born_timestamp);
251        assert_eq!(left.flag, right.flag);
252        assert_eq!(left.properties, right.properties);
253        assert_eq!(left.reconsume_times, right.reconsume_times);
254        assert_eq!(left.unit_mode, right.unit_mode);
255        assert_eq!(left.batch, right.batch);
256        assert_eq!(left.max_reconsume_times, right.max_reconsume_times);
257    }
258
259    #[test]
260    fn send_message_request_header_fast_decode_matches_normal_decode() {
261        let mut header = minimal_header();
262        header.properties = Some(CheetahString::from_static_str("KEYS=abc"));
263        header.reconsume_times = Some(2);
264        header.unit_mode = Some(true);
265        header.batch = Some(false);
266        header.max_reconsume_times = Some(16);
267        let request =
268            RemotingCommand::create_remoting_command(RequestCode::SendMessage).set_ext_fields(header.to_map().unwrap());
269
270        let normal = request
271            .decode_command_custom_header::<SendMessageRequestHeader>()
272            .expect("normal decode should succeed");
273        let fast = request
274            .decode_command_custom_header_fast::<SendMessageRequestHeader>()
275            .expect("fast decode should succeed");
276        let parsed = parse_request_header(&request, RequestCode::SendMessage).expect("parse should use fast decode");
277
278        assert_same_send_message_header(&normal, &fast);
279        assert_same_send_message_header(&normal, &parsed);
280    }
281
282    #[test]
283    fn send_message_v2_fast_decode_matches_normal_v1_conversion() {
284        let mut header = minimal_header();
285        header.properties = Some(CheetahString::from_static_str("TAGS=TagA"));
286        header.reconsume_times = Some(1);
287        header.unit_mode = Some(false);
288        header.batch = Some(true);
289        header.max_reconsume_times = Some(8);
290        header.set_broker_name(CheetahString::from_static_str("broker-a"));
291        let v2 = SendMessageRequestHeaderV2::create_send_message_request_header_v2(&header);
292        let request =
293            RemotingCommand::create_remoting_command(RequestCode::SendMessageV2).set_ext_fields(v2.to_map().unwrap());
294
295        let normal_v2 = request
296            .decode_command_custom_header::<SendMessageRequestHeaderV2>()
297            .expect("normal V2 decode should succeed");
298        let fast_v2 = request
299            .decode_command_custom_header_fast::<SendMessageRequestHeaderV2>()
300            .expect("fast V2 decode should succeed");
301        let normal_v1 = SendMessageRequestHeaderV2::create_send_message_request_header_v1(&normal_v2);
302        let fast_v1 = SendMessageRequestHeaderV2::create_send_message_request_header_v1(&fast_v2);
303        let parsed =
304            parse_request_header(&request, RequestCode::SendMessageV2).expect("parse should use V2 fast decode");
305
306        assert_same_send_message_header(&normal_v1, &fast_v1);
307        assert_same_send_message_header(&normal_v1, &parsed);
308        assert_eq!(parsed.broker_name().map(CheetahString::as_str), Some("broker-a"));
309    }
310
311    #[test]
312    fn parse_request_header_handles_invalid_request_code() {
313        let request = RemotingCommand::create_remoting_command(RequestCode::SendBatchMessage);
314        let request_code = RequestCode::SendBatchMessage;
315        let result = parse_request_header(&request, request_code);
316        assert!(result.is_err());
317    }
318
319    #[test]
320    fn parse_request_header_handles_missing_header() {
321        let request = RemotingCommand::create_remoting_command(RequestCode::SendMessageV2);
322        let request_code = RequestCode::SendMessageV2;
323        let result = parse_request_header(&request, request_code);
324        assert!(result.is_err());
325    }
326
327    #[test]
328    fn send_message_request_header_serializes_correctly() {
329        let header = SendMessageRequestHeader {
330            producer_group: CheetahString::from_static_str("test_producer_group"),
331            topic: CheetahString::from_static_str("test_topic"),
332            default_topic: CheetahString::from_static_str("test_default_topic"),
333            default_topic_queue_nums: 8,
334            queue_id: 1,
335            sys_flag: 0,
336            born_timestamp: 1622547800000,
337            flag: 0,
338            properties: Some(CheetahString::from_static_str("test_properties")),
339            reconsume_times: Some(3),
340            unit_mode: Some(true),
341            batch: Some(false),
342            max_reconsume_times: Some(5),
343            topic_request_header: None,
344        };
345        let map = header.to_map().unwrap();
346        assert_eq!(
347            map.get(&CheetahString::from_static_str("producerGroup")).unwrap(),
348            "test_producer_group"
349        );
350        assert_eq!(map.get(&CheetahString::from_static_str("topic")).unwrap(), "test_topic");
351        assert_eq!(
352            map.get(&CheetahString::from_static_str("defaultTopic")).unwrap(),
353            "test_default_topic"
354        );
355        assert_eq!(
356            map.get(&CheetahString::from_static_str("defaultTopicQueueNums"))
357                .unwrap(),
358            "8"
359        );
360        assert_eq!(map.get(&CheetahString::from_static_str("queueId")).unwrap(), "1");
361        assert_eq!(map.get(&CheetahString::from_static_str("sysFlag")).unwrap(), "0");
362        assert_eq!(
363            map.get(&CheetahString::from_static_str("bornTimestamp")).unwrap(),
364            "1622547800000"
365        );
366        assert_eq!(map.get(&CheetahString::from_static_str("flag")).unwrap(), "0");
367        assert_eq!(
368            map.get(&CheetahString::from_static_str("properties")).unwrap(),
369            "test_properties"
370        );
371        assert_eq!(map.get(&CheetahString::from_static_str("reconsumeTimes")).unwrap(), "3");
372        assert_eq!(map.get(&CheetahString::from_static_str("unitMode")).unwrap(), "true");
373        assert_eq!(map.get(&CheetahString::from_static_str("batch")).unwrap(), "false");
374        assert_eq!(
375            map.get(&CheetahString::from_static_str("maxReconsumeTimes")).unwrap(),
376            "5"
377        );
378    }
379
380    #[test]
381    fn topic_request_header_setters_initialize_missing_nested_headers() {
382        let mut header = minimal_header();
383
384        header.set_lo(Some(true));
385        header.set_broker_name(CheetahString::from_static_str("broker-a"));
386        header.set_namespace(CheetahString::from_static_str("ns-a"));
387        header.set_namespaced(true);
388        header.set_oneway(true);
389
390        assert_eq!(header.lo(), Some(true));
391        assert_eq!(header.broker_name().map(CheetahString::as_str), Some("broker-a"));
392        assert_eq!(header.namespace(), Some("ns-a"));
393        assert_eq!(header.namespaced(), Some(true));
394        assert_eq!(header.oneway(), Some(true));
395
396        let map = header.to_map().expect("header should encode");
397        assert_eq!(map.get("lo").map(CheetahString::as_str), Some("true"));
398        assert_eq!(map.get("bname").map(CheetahString::as_str), Some("broker-a"));
399        assert_eq!(map.get("ns").map(CheetahString::as_str), Some("ns-a"));
400        assert_eq!(map.get("nsd").map(CheetahString::as_str), Some("true"));
401        assert_eq!(map.get("oway").map(CheetahString::as_str), Some("true"));
402    }
403
404    #[test]
405    fn send_message_request_header_deserializes_correctly() {
406        let mut map = HashMap::new();
407        map.insert(
408            CheetahString::from_static_str("producerGroup"),
409            CheetahString::from_static_str("test_producer_group"),
410        );
411        map.insert(
412            CheetahString::from_static_str("topic"),
413            CheetahString::from_static_str("test_topic"),
414        );
415        map.insert(
416            CheetahString::from_static_str("defaultTopic"),
417            CheetahString::from_static_str("test_default_topic"),
418        );
419        map.insert(
420            CheetahString::from_static_str("defaultTopicQueueNums"),
421            CheetahString::from_static_str("8"),
422        );
423        map.insert(
424            CheetahString::from_static_str("queueId"),
425            CheetahString::from_static_str("1"),
426        );
427        map.insert(
428            CheetahString::from_static_str("sysFlag"),
429            CheetahString::from_static_str("0"),
430        );
431        map.insert(
432            CheetahString::from_static_str("bornTimestamp"),
433            CheetahString::from_static_str("1622547800000"),
434        );
435        map.insert(
436            CheetahString::from_static_str("flag"),
437            CheetahString::from_static_str("0"),
438        );
439        map.insert(
440            CheetahString::from_static_str("properties"),
441            CheetahString::from_static_str("test_properties"),
442        );
443        map.insert(
444            CheetahString::from_static_str("reconsumeTimes"),
445            CheetahString::from_static_str("3"),
446        );
447        map.insert(
448            CheetahString::from_static_str("unitMode"),
449            CheetahString::from_static_str("true"),
450        );
451        map.insert(
452            CheetahString::from_static_str("batch"),
453            CheetahString::from_static_str("false"),
454        );
455        map.insert(
456            CheetahString::from_static_str("maxReconsumeTimes"),
457            CheetahString::from_static_str("5"),
458        );
459
460        let header = <SendMessageRequestHeader as FromMap>::from(&map).unwrap();
461        assert_eq!(header.producer_group, "test_producer_group");
462        assert_eq!(header.topic, "test_topic");
463        assert_eq!(header.default_topic, "test_default_topic");
464        assert_eq!(header.default_topic_queue_nums, 8);
465        assert_eq!(header.queue_id, 1);
466        assert_eq!(header.sys_flag, 0);
467        assert_eq!(header.born_timestamp, 1622547800000);
468        assert_eq!(header.flag, 0);
469        assert_eq!(header.properties.unwrap(), "test_properties");
470        assert_eq!(header.reconsume_times.unwrap(), 3);
471        assert!(header.unit_mode.unwrap());
472        assert!(!header.batch.unwrap());
473        assert_eq!(header.max_reconsume_times.unwrap(), 5);
474    }
475
476    #[test]
477    fn send_message_request_header_handles_missing_optional_fields() {
478        let mut map = HashMap::new();
479        map.insert(
480            CheetahString::from_static_str("queueId"),
481            CheetahString::from_static_str("1"),
482        );
483        map.insert(
484            CheetahString::from_static_str("producerGroup"),
485            CheetahString::from_static_str("test_producer_group"),
486        );
487        map.insert(
488            CheetahString::from_static_str("topic"),
489            CheetahString::from_static_str("test_topic"),
490        );
491        map.insert(
492            CheetahString::from_static_str("defaultTopic"),
493            CheetahString::from_static_str("test_default_topic"),
494        );
495        map.insert(
496            CheetahString::from_static_str("defaultTopicQueueNums"),
497            CheetahString::from_static_str("8"),
498        );
499        map.insert(
500            CheetahString::from_static_str("sysFlag"),
501            CheetahString::from_static_str("0"),
502        );
503        map.insert(
504            CheetahString::from_static_str("bornTimestamp"),
505            CheetahString::from_static_str("1622547800000"),
506        );
507        map.insert(
508            CheetahString::from_static_str("flag"),
509            CheetahString::from_static_str("0"),
510        );
511
512        let header = <SendMessageRequestHeader as FromMap>::from(&map).unwrap();
513        assert_eq!(header.producer_group, "test_producer_group");
514        assert_eq!(header.topic, "test_topic");
515        assert_eq!(header.default_topic, "test_default_topic");
516        assert_eq!(header.default_topic_queue_nums, 8);
517        //assert!(header.queue_id.is_some());
518        assert_eq!(header.sys_flag, 0);
519        assert_eq!(header.born_timestamp, 1622547800000);
520        assert_eq!(header.flag, 0);
521        assert!(header.properties.is_none());
522        assert!(header.reconsume_times.is_none());
523        assert!(header.unit_mode.is_none());
524        assert!(header.batch.is_none());
525        assert!(header.max_reconsume_times.is_none());
526    }
527
528    #[test]
529    fn send_message_request_header_handles_invalid_data() {
530        let mut map = HashMap::new();
531        map.insert(
532            CheetahString::from_static_str("producerGroup"),
533            CheetahString::from_static_str("test_producer_group"),
534        );
535        map.insert(
536            CheetahString::from_static_str("topic"),
537            CheetahString::from_static_str("test_topic"),
538        );
539        map.insert(
540            CheetahString::from_static_str("defaultTopic"),
541            CheetahString::from_static_str("test_default_topic"),
542        );
543        map.insert(
544            CheetahString::from_static_str("defaultTopicQueueNums"),
545            CheetahString::from_static_str("invalid"),
546        );
547        map.insert(
548            CheetahString::from_static_str("sysFlag"),
549            CheetahString::from_static_str("invalid"),
550        );
551        map.insert(
552            CheetahString::from_static_str("bornTimestamp"),
553            CheetahString::from_static_str("invalid"),
554        );
555        map.insert(
556            CheetahString::from_static_str("flag"),
557            CheetahString::from_static_str("invalid"),
558        );
559
560        let result = <SendMessageRequestHeader as FromMap>::from(&map);
561        assert!(result.is_err());
562    }
563}
rocketmq_protocol/protocol/header/message_operation_header/send_message_request_header.rs

rocketmq_protocol/protocol/header/message_operation_header/
send_message_request_header.rs
