rocketmq_remoting/protocol/header/
search_offset_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_common::common::boundary_type::BoundaryType;
17use rocketmq_macros::RequestHeaderCodecV2;
18use serde::Deserialize;
19use serde::Serialize;
20
21use crate::protocol::header::message_operation_header::TopicRequestHeaderTrait;
22use crate::rpc::topic_request_header::TopicRequestHeader;
23
24#[derive(Default, Debug, Serialize, Deserialize, RequestHeaderCodecV2)]
25#[serde(rename_all = "camelCase")]
26pub struct SearchOffsetRequestHeader {
27    #[required]
28    pub topic: CheetahString,
29
30    pub lite_topic: Option<CheetahString>,
31
32    #[required]
33    pub queue_id: i32,
34
35    #[required]
36    pub timestamp: i64,
37
38    pub boundary_type: BoundaryType,
39
40    #[serde(flatten)]
41    pub topic_request_header: Option<TopicRequestHeader>,
42}
43
44impl TopicRequestHeaderTrait for SearchOffsetRequestHeader {
45    fn set_lo(&mut self, lo: Option<bool>) {
46        if let Some(header) = self.topic_request_header.as_mut() {
47            header.lo = lo;
48        }
49    }
50
51    fn lo(&self) -> Option<bool> {
52        self.topic_request_header.as_ref().and_then(|h| h.lo)
53    }
54
55    fn set_topic(&mut self, topic: CheetahString) {
56        self.topic = topic;
57    }
58
59    fn topic(&self) -> &CheetahString {
60        &self.topic
61    }
62
63    fn broker_name(&self) -> Option<&CheetahString> {
64        self.topic_request_header
65            .as_ref()
66            .and_then(|h| h.rpc_request_header.as_ref())
67            .and_then(|h| h.broker_name.as_ref())
68    }
69
70    fn set_broker_name(&mut self, broker_name: CheetahString) {
71        if let Some(header) = self.topic_request_header.as_mut() {
72            if let Some(rpc_header) = header.rpc_request_header.as_mut() {
73                rpc_header.broker_name = Some(broker_name);
74            }
75        }
76    }
77
78    fn namespace(&self) -> Option<&str> {
79        self.topic_request_header
80            .as_ref()
81            .and_then(|h| h.rpc_request_header.as_ref())
82            .and_then(|r| r.namespace.as_deref())
83    }
84
85    fn set_namespace(&mut self, namespace: CheetahString) {
86        if let Some(header) = self.topic_request_header.as_mut() {
87            if let Some(rpc_header) = header.rpc_request_header.as_mut() {
88                rpc_header.namespace = Some(namespace);
89            }
90        }
91    }
92
93    fn namespaced(&self) -> Option<bool> {
94        self.topic_request_header
95            .as_ref()
96            .and_then(|h| h.rpc_request_header.as_ref())
97            .and_then(|r| r.namespaced)
98    }
99
100    fn set_namespaced(&mut self, namespaced: bool) {
101        if let Some(header) = self.topic_request_header.as_mut() {
102            if let Some(rpc_header) = header.rpc_request_header.as_mut() {
103                rpc_header.namespaced = Some(namespaced);
104            }
105        }
106    }
107
108    fn oneway(&self) -> Option<bool> {
109        self.topic_request_header
110            .as_ref()
111            .and_then(|h| h.rpc_request_header.as_ref())
112            .and_then(|r| r.oneway)
113    }
114
115    fn set_oneway(&mut self, oneway: bool) {
116        if let Some(header) = self.topic_request_header.as_mut() {
117            if let Some(rpc_header) = header.rpc_request_header.as_mut() {
118                rpc_header.oneway = Some(oneway);
119            }
120        }
121    }
122
123    fn queue_id(&self) -> i32 {
124        self.queue_id
125    }
126
127    fn set_queue_id(&mut self, queue_id: i32) {
128        self.queue_id = queue_id;
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135    use crate::protocol::command_custom_header::CommandCustomHeader;
136    use crate::protocol::command_custom_header::FromMap;
137
138    #[test]
139    fn search_offset_request_header_default() {
140        let header = SearchOffsetRequestHeader::default();
141        assert_eq!(header.topic, CheetahString::default());
142        assert!(header.lite_topic.is_none());
143        assert_eq!(header.queue_id, 0);
144        assert_eq!(header.timestamp, 0);
145        assert_eq!(header.boundary_type, BoundaryType::Lower);
146    }
147
148    #[test]
149    fn search_offset_request_header_creation() {
150        let header = SearchOffsetRequestHeader {
151            topic: CheetahString::from("test_topic"),
152            lite_topic: None,
153            queue_id: 1,
154            timestamp: 1702345678000,
155            boundary_type: BoundaryType::Upper,
156            topic_request_header: None,
157        };
158
159        assert_eq!(header.topic, CheetahString::from("test_topic"));
160        assert!(header.lite_topic.is_none());
161        assert_eq!(header.queue_id, 1);
162        assert_eq!(header.timestamp, 1702345678000);
163        assert_eq!(header.boundary_type, BoundaryType::Upper);
164    }
165
166    #[test]
167    fn search_offset_request_header_maps_lite_topic_like_java_fast_header() {
168        let header = SearchOffsetRequestHeader {
169            topic: CheetahString::from("parent_topic"),
170            lite_topic: Some(CheetahString::from("lite_topic")),
171            queue_id: 2,
172            timestamp: 1702345678999,
173            boundary_type: BoundaryType::Upper,
174            topic_request_header: None,
175        };
176
177        let map = header.to_map().unwrap();
178        assert_eq!(map.get("topic").map(|value| value.as_str()), Some("parent_topic"));
179        assert_eq!(map.get("liteTopic").map(|value| value.as_str()), Some("lite_topic"));
180
181        let decoded = <SearchOffsetRequestHeader as FromMap>::from(&map).unwrap();
182        assert_eq!(decoded.topic, "parent_topic");
183        assert_eq!(decoded.lite_topic.as_deref(), Some("lite_topic"));
184    }
185
186    #[test]
187    fn search_offset_request_header_serializes_to_json() {
188        let header = SearchOffsetRequestHeader {
189            topic: CheetahString::from("my_topic"),
190            lite_topic: Some(CheetahString::from("my_lite_topic")),
191            queue_id: 2,
192            timestamp: 1702345678999,
193            boundary_type: BoundaryType::Upper,
194            topic_request_header: None,
195        };
196
197        let json = serde_json::to_string(&header).unwrap();
198
199        // Verify camelCase field names and uppercase enum
200        assert!(json.contains(r#""topic":"my_topic""#));
201        assert!(json.contains(r#""liteTopic":"my_lite_topic""#));
202        assert!(json.contains(r#""queueId":2"#));
203        assert!(json.contains(r#""timestamp":1702345678999"#));
204        assert!(json.contains(r#""boundaryType":"UPPER""#));
205    }
206
207    #[test]
208    fn search_offset_request_header_deserializes_from_json() {
209        let json = r#"{
210            "topic": "test_topic",
211            "queueId": 3,
212            "timestamp": 1702345678123,
213            "boundaryType": "LOWER"
214        }"#;
215
216        let header: SearchOffsetRequestHeader = serde_json::from_str(json).unwrap();
217
218        assert_eq!(header.topic, CheetahString::from("test_topic"));
219        assert_eq!(header.queue_id, 3);
220        assert_eq!(header.timestamp, 1702345678123);
221        assert_eq!(header.boundary_type, BoundaryType::Lower);
222    }
223
224    #[test]
225    fn search_offset_request_header_deserializes_with_uppercase_boundary_type() {
226        let json = r#"{
227            "topic": "topic1",
228            "queueId": 0,
229            "timestamp": 1000000000,
230            "boundaryType": "UPPER"
231        }"#;
232
233        let header: SearchOffsetRequestHeader = serde_json::from_str(json).unwrap();
234
235        assert_eq!(header.boundary_type, BoundaryType::Upper);
236    }
237
238    #[test]
239    fn search_offset_request_header_deserializes_with_lowercase_boundary_type() {
240        let json = r#"{
241            "topic": "topic2",
242            "queueId": 5,
243            "timestamp": 2000000000,
244            "boundaryType": "lower"
245        }"#;
246
247        let header: SearchOffsetRequestHeader = serde_json::from_str(json).unwrap();
248
249        assert_eq!(header.boundary_type, BoundaryType::Lower);
250    }
251
252    #[test]
253    fn search_offset_request_header_boundary_type_defaults_to_lower_for_invalid_value() {
254        let json = r#"{
255            "topic": "topic3",
256            "queueId": 1,
257            "timestamp": 3000000000,
258            "boundaryType": "invalid"
259        }"#;
260
261        let header: SearchOffsetRequestHeader = serde_json::from_str(json).unwrap();
262
263        // Matches Java behavior: defaults to Lower
264        assert_eq!(header.boundary_type, BoundaryType::Lower);
265    }
266
267    #[test]
268    fn search_offset_request_header_roundtrip_serialization() {
269        let original = SearchOffsetRequestHeader {
270            topic: CheetahString::from("roundtrip_topic"),
271            lite_topic: Some(CheetahString::from("roundtrip_lite_topic")),
272            queue_id: 10,
273            timestamp: 1702400000000,
274            boundary_type: BoundaryType::Upper,
275            topic_request_header: None,
276        };
277
278        let json = serde_json::to_string(&original).unwrap();
279        let deserialized: SearchOffsetRequestHeader = serde_json::from_str(&json).unwrap();
280
281        assert_eq!(deserialized.topic, original.topic);
282        assert_eq!(deserialized.lite_topic, original.lite_topic);
283        assert_eq!(deserialized.queue_id, original.queue_id);
284        assert_eq!(deserialized.timestamp, original.timestamp);
285        assert_eq!(deserialized.boundary_type, original.boundary_type);
286    }
287
288    #[test]
289    fn search_offset_request_header_with_negative_queue_id() {
290        let header = SearchOffsetRequestHeader {
291            topic: CheetahString::from("test"),
292            lite_topic: None,
293            queue_id: -1,
294            timestamp: 1000,
295            boundary_type: BoundaryType::Lower,
296            topic_request_header: None,
297        };
298
299        assert_eq!(header.queue_id, -1);
300    }
301
302    #[test]
303    fn search_offset_request_header_with_empty_topic() {
304        let header = SearchOffsetRequestHeader {
305            topic: CheetahString::new(),
306            lite_topic: None,
307            queue_id: 0,
308            timestamp: 0,
309            boundary_type: BoundaryType::Lower,
310            topic_request_header: None,
311        };
312
313        assert!(header.topic.is_empty());
314    }
315
316    #[test]
317    fn search_offset_request_header_with_large_timestamp() {
318        let header = SearchOffsetRequestHeader {
319            topic: CheetahString::from("test"),
320            lite_topic: None,
321            queue_id: 0,
322            timestamp: i64::MAX,
323            boundary_type: BoundaryType::Upper,
324            topic_request_header: None,
325        };
326
327        assert_eq!(header.timestamp, i64::MAX);
328
329        let json = serde_json::to_string(&header).unwrap();
330        let deserialized: SearchOffsetRequestHeader = serde_json::from_str(&json).unwrap();
331        assert_eq!(deserialized.timestamp, i64::MAX);
332    }
333
334    // Tests for TopicRequestHeaderTrait implementation
335    #[test]
336    fn topic_request_header_trait_set_and_get_topic() {
337        let mut header = SearchOffsetRequestHeader::default();
338        let topic = CheetahString::from("test_topic");
339
340        header.set_topic(topic.clone());
341        assert_eq!(header.topic(), &topic);
342    }
343
344    #[test]
345    fn topic_request_header_trait_set_and_get_queue_id() {
346        let mut header = SearchOffsetRequestHeader::default();
347
348        header.set_queue_id(5);
349        assert_eq!(header.queue_id(), 5);
350
351        header.set_queue_id(-1);
352        assert_eq!(header.queue_id(), -1);
353    }
354
355    #[test]
356    fn topic_request_header_trait_lo_with_none_topic_request_header() {
357        let mut header = SearchOffsetRequestHeader::default();
358
359        // Should not panic when topic_request_header is None
360        header.set_lo(Some(true));
361        assert_eq!(header.lo(), None);
362    }
363
364    #[test]
365    fn topic_request_header_trait_lo_with_some_topic_request_header() {
366        let mut header = SearchOffsetRequestHeader {
367            topic: CheetahString::from("test"),
368            lite_topic: None,
369            queue_id: 0,
370            timestamp: 0,
371            boundary_type: BoundaryType::Lower,
372            topic_request_header: Some(TopicRequestHeader::default()),
373        };
374
375        header.set_lo(Some(true));
376        assert_eq!(header.lo(), Some(true));
377
378        header.set_lo(Some(false));
379        assert_eq!(header.lo(), Some(false));
380
381        header.set_lo(None);
382        assert_eq!(header.lo(), None);
383    }
384
385    #[test]
386    fn topic_request_header_trait_broker_name_with_none() {
387        let header = SearchOffsetRequestHeader::default();
388        assert_eq!(header.broker_name(), None);
389    }
390
391    #[test]
392    fn topic_request_header_trait_set_broker_name_with_none_topic_request_header() {
393        let mut header = SearchOffsetRequestHeader::default();
394
395        // Should not panic when topic_request_header is None
396        header.set_broker_name(CheetahString::from("broker1"));
397        assert_eq!(header.broker_name(), None);
398    }
399
400    #[test]
401    fn topic_request_header_trait_namespace_with_none() {
402        let header = SearchOffsetRequestHeader::default();
403        assert_eq!(header.namespace(), None);
404    }
405
406    #[test]
407    fn topic_request_header_trait_set_namespace_with_none_topic_request_header() {
408        let mut header = SearchOffsetRequestHeader::default();
409
410        // Should not panic when topic_request_header is None
411        header.set_namespace(CheetahString::from("namespace1"));
412        assert_eq!(header.namespace(), None);
413    }
414
415    #[test]
416    fn topic_request_header_trait_namespaced_with_none() {
417        let header = SearchOffsetRequestHeader::default();
418        assert_eq!(header.namespaced(), None);
419    }
420
421    #[test]
422    fn topic_request_header_trait_set_namespaced_with_none_topic_request_header() {
423        let mut header = SearchOffsetRequestHeader::default();
424
425        // Should not panic when topic_request_header is None
426        header.set_namespaced(true);
427        assert_eq!(header.namespaced(), None);
428    }
429
430    #[test]
431    fn topic_request_header_trait_oneway_with_none() {
432        let header = SearchOffsetRequestHeader::default();
433        assert_eq!(header.oneway(), None);
434    }
435
436    #[test]
437    fn topic_request_header_trait_set_oneway_with_none_topic_request_header() {
438        let mut header = SearchOffsetRequestHeader::default();
439
440        // Should not panic when topic_request_header is None
441        header.set_oneway(true);
442        assert_eq!(header.oneway(), None);
443    }
444
445    #[test]
446    fn topic_request_header_trait_all_methods_safe_with_none() {
447        let mut header = SearchOffsetRequestHeader::default();
448
449        // All these operations should be safe and not panic
450        header.set_lo(Some(true));
451        header.set_broker_name(CheetahString::from("broker"));
452        header.set_namespace(CheetahString::from("ns"));
453        header.set_namespaced(true);
454        header.set_oneway(false);
455
456        assert_eq!(header.lo(), None);
457        assert_eq!(header.broker_name(), None);
458        assert_eq!(header.namespace(), None);
459        assert_eq!(header.namespaced(), None);
460        assert_eq!(header.oneway(), None);
461    }
462}
rocketmq_remoting/protocol/header/search_offset_request_header.rs

rocketmq_remoting/protocol/header/
search_offset_request_header.rs
