rocketmq_protocol/protocol/header/
pull_message_response_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 rocketmq_macros::RequestHeaderCodec;
16use serde::Deserialize;
17use serde::Serialize;
18
19#[derive(Serialize, Deserialize, Debug, Default, Clone, RequestHeaderCodec)]
20#[header(
21    type_id = "rocketmq_protocol::protocol::header::pull_message_response_header::PullMessageResponseHeader",
22    java_class = "org.apache.rocketmq.remoting.protocol.header.PullMessageResponseHeader",
23    fast
24)]
25#[serde(rename_all = "camelCase")]
26pub struct PullMessageResponseHeader {
27    #[header(required, range = "i64")]
28    pub suggest_which_broker_id: u64,
29
30    #[header(required)]
31    pub next_begin_offset: i64,
32
33    #[header(required)]
34    pub min_offset: i64,
35
36    #[header(required)]
37    pub max_offset: i64,
38
39    pub offset_delta: Option<i64>,
40    pub topic_sys_flag: Option<i32>,
41    pub group_sys_flag: Option<i32>,
42    pub forbidden_type: Option<i32>,
43}
44
45#[cfg(test)]
46mod tests {
47    use std::collections::HashMap;
48
49    use cheetah_string::CheetahString;
50
51    use super::*;
52    use crate::protocol::command_custom_header::CommandCustomHeader;
53    use crate::protocol::command_custom_header::FromMap;
54
55    #[test]
56    fn pull_message_response_header_serializes_correctly() {
57        let header = PullMessageResponseHeader {
58            suggest_which_broker_id: 123,
59            next_begin_offset: 456,
60            min_offset: 789,
61            max_offset: 101112,
62            offset_delta: Some(131415),
63            topic_sys_flag: Some(161718),
64            group_sys_flag: Some(192021),
65            forbidden_type: Some(222324),
66        };
67        let map = header.to_map().unwrap();
68        assert_eq!(
69            map.get(&CheetahString::from_static_str("suggestWhichBrokerId"))
70                .unwrap(),
71            "123"
72        );
73        assert_eq!(
74            map.get(&CheetahString::from_static_str("nextBeginOffset")).unwrap(),
75            "456"
76        );
77        assert_eq!(map.get(&CheetahString::from_static_str("minOffset")).unwrap(), "789");
78        assert_eq!(map.get(&CheetahString::from_static_str("maxOffset")).unwrap(), "101112");
79        assert_eq!(
80            map.get(&CheetahString::from_static_str("offsetDelta")).unwrap(),
81            "131415"
82        );
83        assert_eq!(
84            map.get(&CheetahString::from_static_str("topicSysFlag")).unwrap(),
85            "161718"
86        );
87        assert_eq!(
88            map.get(&CheetahString::from_static_str("groupSysFlag")).unwrap(),
89            "192021"
90        );
91        assert_eq!(
92            map.get(&CheetahString::from_static_str("forbiddenType")).unwrap(),
93            "222324"
94        );
95    }
96
97    #[test]
98    fn pull_message_response_header_deserializes_correctly() {
99        let mut map = HashMap::new();
100        map.insert(
101            CheetahString::from_static_str("suggestWhichBrokerId"),
102            CheetahString::from_static_str("123"),
103        );
104        map.insert(
105            CheetahString::from_static_str("nextBeginOffset"),
106            CheetahString::from_static_str("456"),
107        );
108        map.insert(
109            CheetahString::from_static_str("minOffset"),
110            CheetahString::from_static_str("789"),
111        );
112        map.insert(
113            CheetahString::from_static_str("maxOffset"),
114            CheetahString::from_static_str("101112"),
115        );
116        map.insert(
117            CheetahString::from_static_str("offsetDelta"),
118            CheetahString::from_static_str("131415"),
119        );
120        map.insert(
121            CheetahString::from_static_str("topicSysFlag"),
122            CheetahString::from_static_str("161718"),
123        );
124        map.insert(
125            CheetahString::from_static_str("groupSysFlag"),
126            CheetahString::from_static_str("192021"),
127        );
128        map.insert(
129            CheetahString::from_static_str("forbiddenType"),
130            CheetahString::from_static_str("222324"),
131        );
132
133        let header: PullMessageResponseHeader = <PullMessageResponseHeader as FromMap>::from(&map).unwrap();
134        assert_eq!(header.suggest_which_broker_id, 123);
135        assert_eq!(header.next_begin_offset, 456);
136        assert_eq!(header.min_offset, 789);
137        assert_eq!(header.max_offset, 101112);
138        assert_eq!(header.offset_delta.unwrap(), 131415);
139        assert_eq!(header.topic_sys_flag.unwrap(), 161718);
140        assert_eq!(header.group_sys_flag.unwrap(), 192021);
141        assert_eq!(header.forbidden_type.unwrap(), 222324);
142    }
143
144    #[test]
145    fn pull_message_response_header_handles_missing_optional_fields() {
146        let mut map = HashMap::new();
147        map.insert(
148            CheetahString::from_static_str("suggestWhichBrokerId"),
149            CheetahString::from_static_str("123"),
150        );
151        map.insert(
152            CheetahString::from_static_str("nextBeginOffset"),
153            CheetahString::from_static_str("456"),
154        );
155        map.insert(
156            CheetahString::from_static_str("minOffset"),
157            CheetahString::from_static_str("789"),
158        );
159        map.insert(
160            CheetahString::from_static_str("maxOffset"),
161            CheetahString::from_static_str("101112"),
162        );
163
164        let header = <PullMessageResponseHeader as FromMap>::from(&map).unwrap();
165        assert_eq!(header.suggest_which_broker_id, 123);
166        assert_eq!(header.next_begin_offset, 456);
167        assert_eq!(header.min_offset, 789);
168        assert_eq!(header.max_offset, 101112);
169        assert!(header.offset_delta.is_none());
170        assert!(header.topic_sys_flag.is_none());
171        assert!(header.group_sys_flag.is_none());
172        assert!(header.forbidden_type.is_none());
173    }
174
175    #[test]
176    fn pull_message_response_header_handles_invalid_data() {
177        let mut map = HashMap::new();
178        map.insert(
179            CheetahString::from_static_str("suggestWhichBrokerId"),
180            CheetahString::from_static_str("invalid"),
181        );
182        map.insert(
183            CheetahString::from_static_str("nextBeginOffset"),
184            CheetahString::from_static_str("invalid"),
185        );
186        map.insert(
187            CheetahString::from_static_str("minOffset"),
188            CheetahString::from_static_str("invalid"),
189        );
190        map.insert(
191            CheetahString::from_static_str("maxOffset"),
192            CheetahString::from_static_str("invalid"),
193        );
194
195        let result = <PullMessageResponseHeader as FromMap>::from(&map);
196        assert!(result.is_err());
197    }
198}
rocketmq_protocol/protocol/header/pull_message_response_header.rs

rocketmq_protocol/protocol/header/
pull_message_response_header.rs
