rocketmq_protocol/protocol/header/
reply_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::namesrv::topic_operation_header::TopicRequestHeader;
21
22/// Represents the header of a reply message request.
23#[derive(Serialize, Deserialize, Debug, Default, RequestHeaderCodec)]
24#[header(
25    type_id = "rocketmq_protocol::protocol::header::reply_message_request_header::ReplyMessageRequestHeader",
26    java_class = "org.apache.rocketmq.remoting.protocol.header.ReplyMessageRequestHeader"
27)]
28#[serde(rename_all = "camelCase")]
29pub struct ReplyMessageRequestHeader {
30    /// Producer group associated with the message.
31    #[header(required)]
32    pub producer_group: CheetahString,
33
34    /// The topic of the message.
35    #[header(required)]
36    pub topic: CheetahString,
37
38    /// Default topic used when the specified topic is not found.
39    #[header(required)]
40    pub default_topic: CheetahString,
41
42    /// Number of queues in the default topic.
43    #[header(required)]
44    pub default_topic_queue_nums: i32,
45
46    /// Queue ID of the message.
47    #[header(required)]
48    pub queue_id: i32,
49
50    /// System flags associated with the message.
51    #[header(required)]
52    pub sys_flag: i32,
53
54    /// Timestamp of when the message was born.
55    #[header(required)]
56    pub born_timestamp: i64,
57
58    /// Flags associated with the message.
59    #[header(required)]
60    pub flag: i32,
61
62    /// Properties of the message (nullable).
63    pub properties: Option<CheetahString>,
64
65    /// Number of times the message has been reconsumed (nullable).
66    pub reconsume_times: Option<i32>,
67
68    /// Whether the message processing is in unit mode (nullable).
69    #[serde(default = "default_unit_mode")]
70    #[header(default_with = "default_unit_mode", default_semantic = "literal:false")]
71    pub unit_mode: Option<bool>,
72
73    /// Host where the message was born.
74    #[header(required)]
75    pub born_host: CheetahString,
76
77    /// Host where the message is stored.
78    #[header(required)]
79    pub store_host: CheetahString,
80
81    /// Timestamp of when the message was stored.
82    #[header(required)]
83    pub store_timestamp: i64,
84
85    #[serde(flatten)]
86    #[header(flatten, presence = "always")]
87    pub topic_request: Option<TopicRequestHeader>,
88}
89
90fn default_unit_mode() -> Option<bool> {
91    Some(false)
92}
93
94#[cfg(test)]
95mod reply_message_request_header_tests {
96    use std::collections::HashMap;
97
98    use super::*;
99    use crate::protocol::command_custom_header::CommandCustomHeader;
100    use crate::protocol::command_custom_header::FromMap;
101    #[test]
102    fn deserialize_from_map_with_all_fields_populates_struct_correctly() {
103        let mut map: HashMap<CheetahString, CheetahString> = HashMap::new();
104        map.insert("producerGroup".into(), "test_producer_group".into());
105        map.insert("topic".into(), "test_topic".into());
106        map.insert("defaultTopic".into(), "test_default_topic".into());
107        map.insert("defaultTopicQueueNums".into(), "10".into());
108        map.insert("queueId".into(), "1".into());
109        map.insert("sysFlag".into(), "0".into());
110        map.insert("flag".into(), "0".into());
111        map.insert("bornTimestamp".into(), "1622547800".into());
112        map.insert("bornHost".into(), "test_born_host".into());
113        map.insert("storeHost".into(), "test_store_host".into());
114        map.insert("storeTimestamp".into(), "1622547800".into());
115        map.insert("unitMode".into(), "true".into());
116
117        let header: ReplyMessageRequestHeader = <ReplyMessageRequestHeader as FromMap>::from(&map).unwrap();
118
119        assert_eq!(header.topic, "test_topic");
120        assert_eq!(header.producer_group, "test_producer_group");
121        assert_eq!(header.default_topic, "test_default_topic");
122        assert_eq!(header.default_topic_queue_nums, 10);
123        assert_eq!(header.queue_id, 1);
124        assert_eq!(header.sys_flag, 0);
125        assert_eq!(header.flag, 0);
126        assert_eq!(header.born_timestamp, 1622547800);
127        assert_eq!(header.born_host, "test_born_host");
128        assert_eq!(header.store_host, "test_store_host");
129        assert_eq!(header.store_timestamp, 1622547800);
130        assert_eq!(header.properties, None);
131        assert_eq!(header.reconsume_times, None);
132        assert_eq!(header.unit_mode, Some(true));
133    }
134
135    #[test]
136    fn deserialize_from_map_with_invalid_number_fields_returns_none() {
137        let mut map = HashMap::new();
138        map.insert("producerGroup".into(), "test_producer_group".into());
139        map.insert("topic".into(), "test_topic".into());
140        map.insert("defaultTopic".into(), "test_default_topic".into());
141        map.insert("defaultTopicQueueNums".into(), "invalid".into());
142        let header: Result<ReplyMessageRequestHeader, rocketmq_error::Error> =
143            <ReplyMessageRequestHeader as FromMap>::from(&map);
144        assert!(header.is_err());
145    }
146
147    #[test]
148    fn serialize_header_to_map() {
149        let header = ReplyMessageRequestHeader {
150            topic: "test_topic".into(),
151            producer_group: "test_producer_group".into(),
152            default_topic: "test_default_topic".into(),
153            default_topic_queue_nums: 10,
154            queue_id: 1,
155            flag: 2,
156            sys_flag: 0,
157            born_timestamp: 1622547800,
158            born_host: "test_born_host".into(),
159            store_host: "test_store_host".into(),
160            store_timestamp: 1622547800,
161            properties: Some("test_properties".into()),
162            reconsume_times: Some(1),
163            unit_mode: Some(true),
164            topic_request: None,
165        };
166        let map: HashMap<CheetahString, CheetahString> = header.to_map().unwrap();
167
168        assert_eq!(map.get("topic").unwrap(), "test_topic");
169        assert_eq!(map.get("producerGroup").unwrap(), "test_producer_group");
170        assert_eq!(map.get("defaultTopicQueueNums").unwrap(), "10");
171        assert_eq!(map.get("bornTimestamp").unwrap(), "1622547800");
172        assert!(!map.contains_key("topicRequest"));
173        assert_eq!(map.get("queueId").unwrap(), "1");
174        assert_eq!(map.get("sysFlag").unwrap(), "0");
175        assert_eq!(map.get("bornHost").unwrap(), "test_born_host");
176        assert_eq!(map.get("storeHost").unwrap(), "test_store_host");
177        assert_eq!(map.get("storeTimestamp").unwrap(), "1622547800");
178        assert_eq!(map.get("flag").unwrap(), "2");
179        assert_eq!(map.get("properties").unwrap(), "test_properties");
180        assert_eq!(map.get("reconsumeTimes").unwrap(), "1");
181        assert_eq!(map.get("unitMode").unwrap(), "true");
182    }
183
184    #[test]
185    fn serialize_header_to_map_with_topic_request_header_includes_nested_fields() {
186        let topic_request_header = TopicRequestHeader::default();
187        let header = ReplyMessageRequestHeader {
188            topic: "test_topic".into(),
189            producer_group: "test_producer_group".into(),
190            default_topic: "test_default_topic".into(),
191            default_topic_queue_nums: 10,
192            queue_id: 1,
193            flag: 2,
194            sys_flag: 0,
195            born_timestamp: 1622547800,
196            born_host: "test_born_host".into(),
197            store_host: "test_store_host".into(),
198            store_timestamp: 1622547800,
199            properties: Some("test_properties".into()),
200            reconsume_times: Some(1),
201            unit_mode: Some(true),
202            topic_request: Some(topic_request_header),
203        };
204        let map: HashMap<CheetahString, CheetahString> = header.to_map().unwrap();
205        assert!(!map.contains_key("nestedField"));
206    }
207}
rocketmq_protocol/protocol/header/reply_message_request_header.rs

rocketmq_protocol/protocol/header/
reply_message_request_header.rs
