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

rocketmq_remoting/protocol/header/message_operation_header/
send_message_request_header.rs
