rocketmq_protocol/protocol/header/
extra_info_util.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 std::collections::HashMap;
16use std::vec::Vec;
17
18use crate::common::wire_constants::KEY_SEPARATOR;
19use crate::common::wire_constants::RETRY_GROUP_TOPIC_PREFIX;
20use rocketmq_error::Error;
21use rocketmq_error::Result;
22use rocketmq_model::common::key_builder;
23use rocketmq_model::common::key_builder::KeyBuilder;
24use rocketmq_model::common::pop_retry_policy::PopRetryPolicy;
25
26pub struct ExtraInfoUtil;
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub struct AckExtraInfo<'a> {
30    pub ck_queue_offset: i64,
31    pub pop_time: i64,
32    pub invisible_time: i64,
33    pub revive_queue_id: i32,
34    pub retry: &'a str,
35    pub broker_name: &'a str,
36    pub queue_id: i32,
37    pub queue_offset: i64,
38    pub retry_generation: Option<u64>,
39}
40
41const NORMAL_TOPIC: &str = "0";
42const RETRY_TOPIC: &str = "1";
43const RETRY_TOPIC_V2: &str = "2";
44const QUEUE_OFFSET: &str = "qo";
45
46#[inline]
47fn illegal_argument(message: impl Into<String>) -> Error {
48    crate::error::invalid_argument(message)
49}
50
51impl ExtraInfoUtil {
52    /// Parses the fixed POP receipt-handle prefix without allocating its fields.
53    pub fn parse_ack_extra_info(extra_info: &str) -> Result<AckExtraInfo<'_>> {
54        let mut parts = extra_info.split(KEY_SEPARATOR);
55        let mut next = |name: &str| {
56            parts
57                .next()
58                .ok_or_else(|| illegal_argument(format!("parse POP receipt handle: missing {name}")))
59        };
60
61        let ck_queue_offset = next("checkpoint queue offset")?
62            .parse::<i64>()
63            .map_err(|_| illegal_argument("parse ck_queue_offset error"))?;
64        let pop_time = next("pop time")?
65            .parse::<i64>()
66            .map_err(|_| illegal_argument("parse pop_time error"))?;
67        let invisible_time = next("invisible time")?
68            .parse::<i64>()
69            .map_err(|_| illegal_argument("parse invisible_time error"))?;
70        let revive_queue_id = next("revive queue id")?
71            .parse::<i32>()
72            .map_err(|_| illegal_argument("parse revive_qid error"))?;
73        let retry = next("retry marker")?;
74        let broker_name = next("broker name")?;
75        let queue_id = next("queue id")?
76            .parse::<i32>()
77            .map_err(|_| illegal_argument("parse queue_id error"))?;
78        let queue_offset = next("queue offset")?
79            .parse::<i64>()
80            .map_err(|_| illegal_argument("parse queue_offset error"))?;
81        let retry_generation = parts
82            .next()
83            .map(|value| {
84                value
85                    .parse::<u64>()
86                    .map_err(|_| illegal_argument("parse retry generation error"))
87            })
88            .transpose()?;
89
90        if broker_name.trim().is_empty() {
91            return Err(illegal_argument("parse POP receipt handle: broker name is blank"));
92        }
93        if !matches!(retry, NORMAL_TOPIC | RETRY_TOPIC | RETRY_TOPIC_V2) {
94            return Err(illegal_argument("parse POP receipt handle: invalid retry marker"));
95        }
96
97        Ok(AckExtraInfo {
98            ck_queue_offset,
99            pop_time,
100            invisible_time,
101            revive_queue_id,
102            retry,
103            broker_name,
104            queue_id,
105            queue_offset,
106            retry_generation,
107        })
108    }
109
110    /// Split the extra info string using the key separator
111    pub fn split(extra_info: &str) -> Vec<String> {
112        extra_info.split(KEY_SEPARATOR).map(|s| s.to_string()).collect()
113    }
114
115    /// Get the checkpoint queue offset from the extra info
116    pub fn get_ck_queue_offset(extra_info_strs: &[String]) -> Result<i64> {
117        if extra_info_strs.is_empty() {
118            return Err(illegal_argument(format!(
119                "getCkQueueOffset fail, extraInfoStrs length {}",
120                extra_info_strs.len()
121            )));
122        }
123        extra_info_strs[0]
124            .parse::<i64>()
125            .map_err(|_| illegal_argument("parse ck_queue_offset error"))
126    }
127
128    /// Get the pop time from the extra info
129    pub fn get_pop_time(extra_info_strs: &[String]) -> Result<i64> {
130        if extra_info_strs.len() < 2 {
131            return Err(illegal_argument(format!(
132                "getPopTime fail, extraInfoStrs length {}",
133                extra_info_strs.len()
134            )));
135        }
136        extra_info_strs[1]
137            .parse::<i64>()
138            .map_err(|_| illegal_argument("parse pop_time error"))
139    }
140
141    /// Get the invisible time from the extra info
142    pub fn get_invisible_time(extra_info_strs: &[String]) -> Result<i64> {
143        if extra_info_strs.len() < 3 {
144            return Err(illegal_argument(format!(
145                "getInvisibleTime fail, extraInfoStrs length {}",
146                extra_info_strs.len()
147            )));
148        }
149        extra_info_strs[2]
150            .parse::<i64>()
151            .map_err(|_| illegal_argument("parse invisible_time error"))
152    }
153
154    /// Get the revive queue ID from the extra info
155    pub fn get_revive_qid(extra_info_strs: &[String]) -> Result<i32> {
156        if extra_info_strs.len() < 4 {
157            return Err(illegal_argument(format!(
158                "getReviveQid fail, extraInfoStrs length {}",
159                extra_info_strs.len()
160            )));
161        }
162        extra_info_strs[3]
163            .parse::<i32>()
164            .map_err(|_| illegal_argument("parse revive_qid error"))
165    }
166
167    /// Get the real topic name based on the retry flag
168    pub fn get_real_topic(extra_info_strs: &[String], topic: &str, cid: &str) -> Result<String> {
169        if extra_info_strs.len() < 5 {
170            return Err(illegal_argument(format!(
171                "getRealTopic fail, extraInfoStrs length {}",
172                extra_info_strs.len()
173            )));
174        }
175
176        if RETRY_TOPIC == extra_info_strs[4] {
177            Ok(KeyBuilder::build_pop_retry_topic_v1(topic, cid))
178        } else if RETRY_TOPIC_V2 == extra_info_strs[4] {
179            Ok(KeyBuilder::build_pop_retry_topic_v2(topic, cid))
180        } else {
181            Ok(topic.to_string())
182        }
183    }
184
185    /// Get the real topic name based on the retry flag
186    pub fn get_real_topic_with_retry(topic: &str, cid: &str, retry: &str) -> Result<String> {
187        if retry == NORMAL_TOPIC {
188            Ok(topic.to_string())
189        } else if retry == RETRY_TOPIC {
190            Ok(KeyBuilder::build_pop_retry_topic_v1(topic, cid))
191        } else if retry == RETRY_TOPIC_V2 {
192            Ok(KeyBuilder::build_pop_retry_topic_v2(topic, cid))
193        } else {
194            Err(illegal_argument("getRetry fail, format is wrong"))
195        }
196    }
197
198    /// Get the retry flag from the extra info
199    pub fn get_retry(extra_info_strs: &[String]) -> Result<String> {
200        if extra_info_strs.len() < 5 {
201            return Err(illegal_argument(format!(
202                "getRetry fail, extraInfoStrs length {}",
203                extra_info_strs.len()
204            )));
205        }
206        Ok(extra_info_strs[4].clone())
207    }
208
209    /// Get the broker name from the extra info
210    pub fn get_broker_name(extra_info_strs: &[String]) -> Result<String> {
211        if extra_info_strs.len() < 6 {
212            return Err(illegal_argument(format!(
213                "getBrokerName fail, extraInfoStrs length {}",
214                extra_info_strs.len()
215            )));
216        }
217        Ok(extra_info_strs[5].clone())
218    }
219
220    /// Get the queue ID from the extra info
221    pub fn get_queue_id(extra_info_strs: &[String]) -> rocketmq_error::Result<i32> {
222        if extra_info_strs.len() < 7 {
223            return Err(illegal_argument(format!(
224                "getQueueId fail, extraInfoStrs length {}",
225                extra_info_strs.len()
226            )));
227        }
228        extra_info_strs[6]
229            .parse::<i32>()
230            .map_err(|_| illegal_argument("parse queue_id error"))
231    }
232
233    /// Get the queue offset from the extra info
234    pub fn get_queue_offset(extra_info_strs: &[String]) -> rocketmq_error::Result<i64> {
235        if extra_info_strs.len() < 8 {
236            return Err(illegal_argument(format!(
237                "getQueueOffset fail, extraInfoStrs length {}",
238                extra_info_strs.len()
239            )));
240        }
241        extra_info_strs[7]
242            .parse::<i64>()
243            .map_err(|_| illegal_argument("parse queue_offset error"))
244    }
245
246    /// Build extra info string
247    pub fn build_extra_info(
248        ck_queue_offset: i64,
249        pop_time: i64,
250        invisible_time: i64,
251        revive_qid: i32,
252        topic: &str,
253        broker_name: &str,
254        queue_id: i32,
255    ) -> String {
256        let t = Self::get_retry_from_topic(topic);
257        format!(
258            "{}{sep}{}{sep}{}{sep}{}{sep}{}{sep}{}{sep}{}",
259            ck_queue_offset,
260            pop_time,
261            invisible_time,
262            revive_qid,
263            t,
264            broker_name,
265            queue_id,
266            sep = KEY_SEPARATOR
267        )
268    }
269
270    /// Build extra info string with queue offset
271    pub fn build_extra_info_with_offset(
272        ck_queue_offset: i64,
273        pop_time: i64,
274        invisible_time: i64,
275        revive_qid: i32,
276        topic: &str,
277        broker_name: &str,
278        queue_id: i32,
279        msg_queue_offset: i64,
280    ) -> String {
281        let t = Self::get_retry_from_topic(topic);
282        format!(
283            "{}{sep}{}{sep}{}{sep}{}{sep}{}{sep}{}{sep}{}{sep}{}",
284            ck_queue_offset,
285            pop_time,
286            invisible_time,
287            revive_qid,
288            t,
289            broker_name,
290            queue_id,
291            msg_queue_offset,
292            sep = KEY_SEPARATOR
293        )
294    }
295
296    /// Builds a receipt handle that carries the persisted POP retry policy generation.
297    ///
298    /// Existing eight-field handles remain valid. The optional ninth field is emitted only
299    /// for retry topics, while the existing retry marker continues to identify v1 or v2.
300    #[allow(clippy::too_many_arguments, reason = "preserves the established POP receipt fields")]
301    pub fn build_extra_info_with_offset_and_retry_policy(
302        ck_queue_offset: i64,
303        pop_time: i64,
304        invisible_time: i64,
305        revive_qid: i32,
306        topic: &str,
307        broker_name: &str,
308        queue_id: i32,
309        msg_queue_offset: i64,
310        retry_policy: &PopRetryPolicy,
311    ) -> String {
312        let base = Self::build_extra_info_with_offset(
313            ck_queue_offset,
314            pop_time,
315            invisible_time,
316            revive_qid,
317            topic,
318            broker_name,
319            queue_id,
320            msg_queue_offset,
321        );
322        if KeyBuilder::classify_pop_retry_topic(topic).is_some() {
323            format!("{base}{KEY_SEPARATOR}{}", retry_policy.generation)
324        } else {
325            base
326        }
327    }
328
329    /// Build start offset info
330    pub fn build_start_offset_info(string_builder: &mut String, topic: &str, queue_id: i32, start_offset: i64) {
331        if !string_builder.is_empty() {
332            string_builder.push(';');
333        }
334
335        string_builder.push_str(&Self::get_retry_from_topic(topic));
336        string_builder.push_str(KEY_SEPARATOR);
337        string_builder.push_str(&queue_id.to_string());
338        string_builder.push_str(KEY_SEPARATOR);
339        string_builder.push_str(&start_offset.to_string());
340    }
341
342    /// Build queue ID order count info
343    pub fn build_queue_id_order_count_info(string_builder: &mut String, topic: &str, queue_id: i32, order_count: i32) {
344        if !string_builder.is_empty() {
345            string_builder.push(';');
346        }
347
348        string_builder.push_str(&Self::get_retry_from_topic(topic));
349        string_builder.push_str(KEY_SEPARATOR);
350        string_builder.push_str(&queue_id.to_string());
351        string_builder.push_str(KEY_SEPARATOR);
352        string_builder.push_str(&order_count.to_string());
353    }
354
355    /// Build queue offset order count info
356    pub fn build_queue_offset_order_count_info(
357        string_builder: &mut String,
358        topic: &str,
359        queue_id: i64,
360        queue_offset: i64,
361        order_count: i32,
362    ) {
363        if !string_builder.is_empty() {
364            string_builder.push(';');
365        }
366
367        string_builder.push_str(&Self::get_retry_from_topic(topic));
368        string_builder.push_str(KEY_SEPARATOR);
369        string_builder.push_str(&Self::get_queue_offset_key_value_key(queue_id, queue_offset));
370        string_builder.push_str(KEY_SEPARATOR);
371        string_builder.push_str(&order_count.to_string());
372    }
373
374    /// Build message offset info
375    pub fn build_msg_offset_info(string_builder: &mut String, topic: &str, queue_id: i32, msg_offsets: &[u64]) {
376        if !string_builder.is_empty() {
377            string_builder.push(';');
378        }
379
380        string_builder.push_str(&Self::get_retry_from_topic(topic));
381        string_builder.push_str(KEY_SEPARATOR);
382        string_builder.push_str(&queue_id.to_string());
383        string_builder.push_str(KEY_SEPARATOR);
384
385        for (i, offset) in msg_offsets.iter().enumerate() {
386            string_builder.push_str(&offset.to_string());
387            if i < msg_offsets.len() - 1 {
388                string_builder.push(';');
389            }
390        }
391    }
392
393    /// Parse message offset info into a HashMap
394    pub fn parse_msg_offset_info(msg_offset_info: &str) -> Result<HashMap<String, Vec<i64>>> {
395        let mut msg_offset_map: HashMap<String, Vec<i64>> = HashMap::with_capacity(4);
396        if msg_offset_info.is_empty() {
397            return Ok(msg_offset_map);
398        }
399        let array = if !msg_offset_info.contains(";") {
400            vec![msg_offset_info]
401        } else {
402            msg_offset_info.split(";").collect::<Vec<&str>>()
403        };
404
405        for one in array {
406            let split: Vec<&str> = one.split(KEY_SEPARATOR).collect();
407            if split.len() != 3 {
408                return Err(illegal_argument("parse msgOffsetInfo error"));
409            }
410
411            let key = format!("{}@{}", split[0], split[1]);
412            let offsets = match msg_offset_map.entry(key) {
413                std::collections::hash_map::Entry::Occupied(_) => {
414                    return Err(illegal_argument("parse msgOffsetMap error, duplicate"));
415                }
416                std::collections::hash_map::Entry::Vacant(entry) => entry.insert(Vec::with_capacity(8)),
417            };
418            let msg_offsets: Vec<&str> = split[2].split(",").collect();
419
420            for msg_offset in msg_offsets {
421                let offset = msg_offset
422                    .parse::<i64>()
423                    .map_err(|_| illegal_argument("parse msgOffset error"))?;
424                offsets.push(offset);
425            }
426        }
427
428        Ok(msg_offset_map)
429    }
430
431    /// Parse start offset info into a HashMap
432    pub fn parse_start_offset_info(start_offset_info: &str) -> Result<HashMap<String, i64>> {
433        let mut start_offset_map: HashMap<String, i64> = HashMap::with_capacity(4);
434        if start_offset_info.is_empty() {
435            return Ok(start_offset_map);
436        }
437        let array = if !start_offset_info.contains(";") {
438            vec![start_offset_info]
439        } else {
440            start_offset_info.split(";").collect::<Vec<&str>>()
441        };
442
443        for one in array {
444            let split: Vec<&str> = one.split(KEY_SEPARATOR).collect();
445            if split.len() != 3 {
446                return Err(illegal_argument(format!(
447                    "parse startOffsetInfo error, {start_offset_info}"
448                )));
449            }
450
451            let key = format!("{}@{}", split[0], split[1]);
452            if start_offset_map.contains_key(&key) {
453                return Err(illegal_argument("parse startOffsetMap error, duplicate"));
454            }
455
456            start_offset_map.insert(
457                key,
458                split[2]
459                    .parse::<i64>()
460                    .map_err(|_| illegal_argument("parse startOffset error"))?,
461            );
462        }
463
464        Ok(start_offset_map)
465    }
466
467    /// Parse order count info into a HashMap
468    pub fn parse_order_count_info(order_count_info: &str) -> Result<HashMap<String, i32>> {
469        let mut order_count_map: HashMap<String, i32> = HashMap::with_capacity(4);
470        if order_count_info.is_empty() {
471            return Ok(order_count_map);
472        }
473        let array = if !order_count_info.contains(";") {
474            vec![order_count_info]
475        } else {
476            order_count_info.split(";").collect::<Vec<&str>>()
477        };
478
479        for one in array {
480            let split: Vec<&str> = one.split(KEY_SEPARATOR).collect();
481            if split.len() != 3 {
482                return Err(illegal_argument(format!(
483                    "parse orderCountInfo error {order_count_info}"
484                )));
485            }
486
487            let key = format!("{}@{}", split[0], split[1]);
488            if order_count_map.contains_key(&key) {
489                return Err(illegal_argument(format!(
490                    "parse orderCountInfo error, duplicate, {order_count_info}"
491                )));
492            }
493
494            order_count_map.insert(
495                key,
496                split[2]
497                    .parse::<i32>()
498                    .map_err(|_| illegal_argument("parse orderCountInfo error"))?,
499            );
500        }
501
502        Ok(order_count_map)
503    }
504
505    /// Get start offset info map key
506    pub fn get_start_offset_info_map_key(topic: &str, key: i64) -> String {
507        format!("{}@{}", Self::get_retry_from_topic(topic), key)
508    }
509
510    /// Get start offset info map key with pop check
511    pub fn get_start_offset_info_map_key_with_pop_ck(topic: &str, pop_ck: Option<&str>, key: i64) -> Result<String> {
512        Ok(format!("{}@{}", Self::get_retry_from_topic_pop_ck(topic, pop_ck)?, key))
513    }
514
515    /// Get queue offset key value key
516    pub fn get_queue_offset_key_value_key(queue_id: i64, queue_offset: i64) -> String {
517        format!("{}{}{}{}", QUEUE_OFFSET, queue_id, "%", queue_offset)
518    }
519
520    /// Get queue offset map key
521    pub fn get_queue_offset_map_key(topic: &str, queue_id: i64, queue_offset: i64) -> String {
522        format!(
523            "{}@{}",
524            Self::get_retry_from_topic(topic),
525            Self::get_queue_offset_key_value_key(queue_id, queue_offset)
526        )
527    }
528
529    /// Check if the extra info represents an order
530    pub fn is_order(extra_info: &[String]) -> bool {
531        Self::get_revive_qid(extra_info).unwrap_or_default() == key_builder::POP_ORDER_REVIVE_QUEUE
532    }
533
534    /// Get retry flag from topic name
535    fn get_retry_from_topic(topic: &str) -> String {
536        if KeyBuilder::is_pop_retry_topic_v2(topic) {
537            RETRY_TOPIC_V2.to_string()
538        } else if topic.starts_with(RETRY_GROUP_TOPIC_PREFIX) {
539            RETRY_TOPIC.to_string()
540        } else {
541            NORMAL_TOPIC.to_string()
542        }
543    }
544
545    /// Get retry flag with pop check
546    fn get_retry_from_topic_pop_ck(topic: &str, pop_ck: Option<&str>) -> Result<String> {
547        match pop_ck {
548            Some(ck) => Self::get_retry(&Self::split(ck)),
549            None => Ok(Self::get_retry_from_topic(topic)),
550        }
551    }
552}
553
554#[cfg(test)]
555mod tests {
556    use rocketmq_model::common::key_builder::KeyBuilder;
557
558    use super::*;
559
560    #[test]
561    fn split_with_valid_string() {
562        let result = ExtraInfoUtil::split("a b c");
563        assert_eq!(result, vec!["a", "b", "c"]);
564    }
565
566    #[test]
567    fn get_ck_queue_offset_with_valid_string() {
568        let result = ExtraInfoUtil::get_ck_queue_offset(&["123".to_string()]).unwrap();
569        assert_eq!(result, 123);
570    }
571
572    #[test]
573    fn get_ck_queue_offset_with_invalid_string() {
574        let result = ExtraInfoUtil::get_ck_queue_offset(&["abc".to_string()]).unwrap_err();
575        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
576    }
577
578    #[test]
579    fn get_pop_time_with_valid_string() {
580        let result = ExtraInfoUtil::get_pop_time(&["123".to_string(), "456".to_string()]).unwrap();
581        assert_eq!(result, 456);
582    }
583
584    #[test]
585    fn get_pop_time_with_insufficient_length() {
586        let result = ExtraInfoUtil::get_pop_time(&["123".to_string()]).unwrap_err();
587        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
588    }
589
590    #[test]
591    fn get_invisible_time_with_valid_string() {
592        let result =
593            ExtraInfoUtil::get_invisible_time(&["123".to_string(), "456".to_string(), "789".to_string()]).unwrap();
594        assert_eq!(result, 789);
595    }
596
597    #[test]
598    fn get_invisible_time_with_insufficient_length() {
599        let result = ExtraInfoUtil::get_invisible_time(&["123".to_string(), "456".to_string()]).unwrap_err();
600        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
601    }
602
603    #[test]
604    fn get_revive_qid_with_valid_string() {
605        let result = ExtraInfoUtil::get_revive_qid(&[
606            "123".to_string(),
607            "456".to_string(),
608            "789".to_string(),
609            "10".to_string(),
610        ])
611        .unwrap();
612        assert_eq!(result, 10);
613    }
614
615    #[test]
616    fn get_revive_qid_with_insufficient_length() {
617        let result =
618            ExtraInfoUtil::get_revive_qid(&["123".to_string(), "456".to_string(), "789".to_string()]).unwrap_err();
619        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
620    }
621
622    #[test]
623    fn get_real_topic_with_retry_v1() {
624        let result = ExtraInfoUtil::get_real_topic(
625            &[
626                "123".to_string(),
627                "456".to_string(),
628                "789".to_string(),
629                "10".to_string(),
630                "1".to_string(),
631            ],
632            "topic",
633            "cid",
634        )
635        .unwrap();
636        assert_eq!(result, KeyBuilder::build_pop_retry_topic_v1("topic", "cid"));
637    }
638
639    #[test]
640    fn get_real_topic_with_retry_v2() {
641        let result = ExtraInfoUtil::get_real_topic(
642            &[
643                "123".to_string(),
644                "456".to_string(),
645                "789".to_string(),
646                "10".to_string(),
647                "2".to_string(),
648            ],
649            "topic",
650            "cid",
651        )
652        .unwrap();
653        assert_eq!(result, KeyBuilder::build_pop_retry_topic_v2("topic", "cid"));
654    }
655
656    #[test]
657    fn get_real_topic_with_normal_topic() {
658        let result = ExtraInfoUtil::get_real_topic(
659            &[
660                "123".to_string(),
661                "456".to_string(),
662                "789".to_string(),
663                "10".to_string(),
664                "0".to_string(),
665            ],
666            "topic",
667            "cid",
668        )
669        .unwrap();
670        assert_eq!(result, "topic".to_string());
671    }
672
673    #[test]
674    fn get_real_topic_with_insufficient_length() {
675        let result = ExtraInfoUtil::get_real_topic(
676            &[
677                "123".to_string(),
678                "456".to_string(),
679                "789".to_string(),
680                "10".to_string(),
681            ],
682            "topic",
683            "cid",
684        )
685        .unwrap_err();
686        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
687    }
688
689    #[test]
690    fn get_real_topic_with_retry_invalid() {
691        let result = ExtraInfoUtil::get_real_topic_with_retry("topic", "cid", "3").unwrap_err();
692        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
693    }
694
695    #[test]
696    fn get_retry_slice_with_valid_string() {
697        let result = ExtraInfoUtil::get_retry(&[
698            "123".to_string(),
699            "456".to_string(),
700            "789".to_string(),
701            "10".to_string(),
702            "1".to_string(),
703        ])
704        .unwrap();
705        assert_eq!(result, "1".to_string());
706    }
707
708    #[test]
709    fn get_retry_slice_with_insufficient_length() {
710        let result = ExtraInfoUtil::get_retry(&[
711            "123".to_string(),
712            "456".to_string(),
713            "789".to_string(),
714            "10".to_string(),
715        ])
716        .unwrap_err();
717        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
718    }
719
720    #[test]
721    fn get_broker_name_with_valid_string() {
722        let result = ExtraInfoUtil::get_broker_name(&[
723            "123".to_string(),
724            "456".to_string(),
725            "789".to_string(),
726            "10".to_string(),
727            "1".to_string(),
728            "broker".to_string(),
729        ])
730        .unwrap();
731        assert_eq!(result, "broker".to_string());
732    }
733
734    #[test]
735    fn get_broker_name_with_insufficient_length() {
736        let result = ExtraInfoUtil::get_broker_name(&[
737            "123".to_string(),
738            "456".to_string(),
739            "789".to_string(),
740            "10".to_string(),
741            "1".to_string(),
742        ])
743        .unwrap_err();
744        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
745    }
746
747    #[test]
748    fn get_queue_id_with_valid_string() {
749        let result = ExtraInfoUtil::get_queue_id(&[
750            "123".to_string(),
751            "456".to_string(),
752            "789".to_string(),
753            "10".to_string(),
754            "1".to_string(),
755            "broker".to_string(),
756            "7".to_string(),
757        ])
758        .unwrap();
759        assert_eq!(result, 7);
760    }
761
762    #[test]
763    fn get_queue_id_with_insufficient_length() {
764        let result = ExtraInfoUtil::get_queue_id(&[
765            "123".to_string(),
766            "456".to_string(),
767            "789".to_string(),
768            "10".to_string(),
769            "1".to_string(),
770            "broker".to_string(),
771        ])
772        .unwrap_err();
773        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
774    }
775
776    #[test]
777    fn get_queue_offset_with_valid_string() {
778        let result = ExtraInfoUtil::get_queue_offset(&[
779            "123".to_string(),
780            "456".to_string(),
781            "789".to_string(),
782            "10".to_string(),
783            "1".to_string(),
784            "broker".to_string(),
785            "7".to_string(),
786            "100".to_string(),
787        ])
788        .unwrap();
789        assert_eq!(result, 100);
790    }
791
792    #[test]
793    fn get_queue_offset_with_insufficient_length() {
794        let result = ExtraInfoUtil::get_queue_offset(&[
795            "123".to_string(),
796            "456".to_string(),
797            "789".to_string(),
798            "10".to_string(),
799            "1".to_string(),
800            "broker".to_string(),
801            "7".to_string(),
802        ])
803        .unwrap_err();
804        assert_eq!(result.descriptor(), &rocketmq_error::CORE_ARGUMENT_INVALID);
805    }
806
807    #[test]
808    fn build_extra_info_creates_correct_string() {
809        let result = ExtraInfoUtil::build_extra_info(123, 456, 789, 10, "topic", "broker", 7);
810        assert_eq!(result, "123 456 789 10 0 broker 7");
811    }
812
813    #[test]
814    fn build_extra_info_with_msg_queue_offset_creates_correct_string() {
815        let result = ExtraInfoUtil::build_extra_info_with_offset(123, 456, 789, 10, "topic", "broker", 7, 100);
816        assert_eq!(result, "123 456 789 10 0 broker 7 100");
817    }
818
819    #[test]
820    fn retry_receipt_round_trips_actual_version_and_policy_generation() {
821        let policy = PopRetryPolicy::dual_read_v2_write(17);
822        let retry_topic = policy.write_topic("topic", "consumer-a");
823        let handle = ExtraInfoUtil::build_extra_info_with_offset_and_retry_policy(
824            123,
825            456,
826            789,
827            10,
828            &retry_topic,
829            "broker",
830            7,
831            100,
832            &policy,
833        );
834
835        let parsed = ExtraInfoUtil::parse_ack_extra_info(&handle).expect("generated receipt should parse");
836        assert_eq!(parsed.retry, RETRY_TOPIC_V2);
837        assert_eq!(parsed.retry_generation, Some(17));
838        assert_eq!(
839            ExtraInfoUtil::get_real_topic_with_retry("topic", "consumer-a", parsed.retry)
840                .expect("retry marker should resolve"),
841            retry_topic
842        );
843    }
844
845    #[test]
846    fn build_start_offset_info_creates_correct_string() {
847        let mut string_builder = String::new();
848        ExtraInfoUtil::build_start_offset_info(&mut string_builder, "topic", 7, 100);
849        assert_eq!(string_builder, "0 7 100");
850    }
851
852    #[test]
853    fn build_queue_id_order_count_info_creates_correct_string() {
854        let mut string_builder = String::new();
855        ExtraInfoUtil::build_queue_id_order_count_info(&mut string_builder, "topic", 7, 100);
856        assert_eq!(string_builder, "0 7 100");
857    }
858
859    #[test]
860    fn build_queue_offset_order_count_info_creates_correct_string() {
861        let mut string_builder = String::new();
862        ExtraInfoUtil::build_queue_offset_order_count_info(&mut string_builder, "topic", 7, 100, 100);
863        assert_eq!(string_builder, "0 qo7%100 100");
864    }
865
866    #[test]
867    fn build_msg_offset_info_creates_correct_string() {
868        let mut string_builder = String::new();
869        ExtraInfoUtil::build_msg_offset_info(&mut string_builder, "topic", 7, &[100, 200, 300]);
870        assert_eq!(string_builder, "0 7 100;200;300");
871    }
872
873    #[test]
874    fn parse_msg_offset_info_with_valid_string() {
875        let result = ExtraInfoUtil::parse_msg_offset_info("0 7 100,200,300").unwrap();
876        let mut expected = HashMap::new();
877        expected.insert("0@7".to_string(), vec![100, 200, 300]);
878        assert_eq!(result, expected);
879    }
880
881    #[test]
882    fn parse_start_offset_info_with_valid_string() {
883        let result = ExtraInfoUtil::parse_start_offset_info("0 7 100").unwrap();
884        let mut expected = HashMap::new();
885        expected.insert("0@7".to_string(), 100);
886        assert_eq!(result, expected);
887    }
888
889    #[test]
890    fn parse_order_count_info_with_valid_string() {
891        let result = ExtraInfoUtil::parse_order_count_info("0 7 100").unwrap();
892        let mut expected = HashMap::new();
893        expected.insert("0@7".to_string(), 100);
894        assert_eq!(result, expected);
895    }
896
897    #[test]
898    fn get_retry_for_topic_with_retry_topic() {
899        let result = ExtraInfoUtil::get_retry_from_topic("RETRY_topic");
900        assert_eq!(result, "0");
901    }
902
903    #[test]
904    fn get_retry_for_topic_with_normal_topic() {
905        let result = ExtraInfoUtil::get_retry_from_topic("topic");
906        assert_eq!(result, "0");
907    }
908
909    #[test]
910    fn get_start_offset_info_map_key_creates_correct_string() {
911        let result = ExtraInfoUtil::get_start_offset_info_map_key("topic", 7);
912        assert_eq!(result, "0@7");
913    }
914
915    #[test]
916    fn get_start_offset_info_map_key_with_pop_ck_creates_correct_string() {
917        let result = ExtraInfoUtil::get_start_offset_info_map_key_with_pop_ck("topic", None, 7).unwrap();
918        assert_eq!(result, "0@7");
919    }
920
921    #[test]
922    fn get_queue_offset_key_value_key_creates_correct_string() {
923        let result = ExtraInfoUtil::get_queue_offset_key_value_key(7, 100);
924        assert_eq!(result, "qo7%100");
925    }
926
927    #[test]
928    fn get_queue_offset_map_key_creates_correct_string() {
929        let result = ExtraInfoUtil::get_queue_offset_map_key("topic", 7, 100);
930        assert_eq!(result, "0@qo7%100");
931    }
932
933    #[test]
934    fn is_order_with_order_queue() {
935        let result = ExtraInfoUtil::is_order(&[
936            "123".to_string(),
937            "456".to_string(),
938            "789".to_string(),
939            "999".to_string(),
940            "1".to_string(),
941            "broker".to_string(),
942            "7".to_string(),
943            "999".to_string(),
944            "0".to_string(),
945        ]);
946        assert!(result);
947    }
948
949    #[test]
950    fn is_order_with_non_order_queue() {
951        let result = ExtraInfoUtil::is_order(&[
952            "123".to_string(),
953            "456".to_string(),
954            "789".to_string(),
955            "10".to_string(),
956            "1".to_string(),
957            "broker".to_string(),
958            "7".to_string(),
959            "100".to_string(),
960            "1".to_string(),
961        ]);
962        assert!(!result);
963    }
964}
rocketmq_protocol/protocol/header/extra_info_util.rs

rocketmq_protocol/protocol/header/
extra_info_util.rs
