rocketmq_protocol/trace/
trace_codec.rs

1// Copyright 2026 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;
16
17use super::trace_constants::TraceConstants;
18use super::TraceRecord;
19use super::TraceType;
20
21pub fn encode_records(records: &[TraceRecord]) -> CheetahString {
22    let mut output = String::new();
23    for record in records {
24        output.push_str(&record.trace_type.to_string());
25        for field in &record.fields {
26            output.push(TraceConstants::CONTENT_SPLITOR);
27            output.push_str(field);
28        }
29        output.push(TraceConstants::FIELD_SPLITOR);
30    }
31    CheetahString::from_string(output)
32}
33
34pub fn decode_records(input: &str) -> Vec<TraceRecord> {
35    input
36        .split(TraceConstants::FIELD_SPLITOR)
37        .filter(|record| !record.is_empty())
38        .filter_map(|record| {
39            let mut fields = record.split(TraceConstants::CONTENT_SPLITOR);
40            let trace_type = TraceType::parse(fields.next()?)?;
41            Some(TraceRecord::new(
42                trace_type,
43                fields.map(CheetahString::from_slice).collect(),
44            ))
45        })
46        .collect()
47}
48
49#[cfg(test)]
50mod tests {
51    use super::*;
52
53    #[test]
54    fn frozen_pub_trace_round_trip_preserves_delimiters_and_order() {
55        let frozen = "Pub\u{1}1710000000000\u{1}region\u{1}group\u{1}topic\u{1}msg-id\u{2}";
56        let records = decode_records(frozen);
57        assert_eq!(records.len(), 1);
58        assert_eq!(records[0].trace_type, TraceType::Pub);
59        assert_eq!(encode_records(&records), frozen);
60    }
61
62    #[test]
63    fn malformed_and_unknown_records_are_skipped() {
64        let records = decode_records("Unknown\u{1}x\u{2}\u{2}SubAfter\u{1}request\u{2}");
65        assert_eq!(records.len(), 1);
66        assert_eq!(records[0].trace_type, TraceType::SubAfter);
67    }
68}
rocketmq_protocol/trace/trace_codec.rs

rocketmq_protocol/trace/
trace_codec.rs
