rocketmq_protocol/common/compression/
compressor_factory.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 crate::common::compression::compressor::Compressor;
16use crate::common::compression::lz4_compressor::Lz4Compressor;
17use crate::common::compression::zlib_compressor::ZlibCompressor;
18use crate::common::compression::zstd_compressor::ZstdCompressor;
19use rocketmq_model::common::compression::compression_type::CompressionType;
20
21static LZ4_COMPRESSOR: Lz4Compressor = Lz4Compressor;
22static ZLIB_COMPRESSOR: ZlibCompressor = ZlibCompressor;
23static ZSTD_COMPRESSOR: ZstdCompressor = ZstdCompressor;
24
25pub struct CompressorFactory;
26
27impl CompressorFactory {
28    pub fn get_compressor(compressor_type: CompressionType) -> &'static (dyn Compressor + Send + Sync) {
29        match compressor_type {
30            CompressionType::LZ4 => &LZ4_COMPRESSOR,
31            CompressionType::Zlib => &ZLIB_COMPRESSOR,
32            CompressionType::Zstd => &ZSTD_COMPRESSOR,
33        }
34    }
35}
36
37#[cfg(test)]
38mod tests {
39    use rocketmq_error::fields;
40    use rocketmq_error::ViewValueRef;
41
42    use super::*;
43
44    const ALL_TYPES: [CompressionType; 3] = [CompressionType::LZ4, CompressionType::Zlib, CompressionType::Zstd];
45
46    const COMPRESSORS: [(&str, &dyn Compressor); 3] = [
47        ("lz4", &Lz4Compressor),
48        ("zlib", &ZlibCompressor),
49        ("zstd", &ZstdCompressor),
50    ];
51
52    const TYPICAL_PAYLOAD: &[u8] =
53        b"Hello RocketMQ Rust! {\"topic\":\"TopicTest\",\"tags\":\"TagA\",\"keys\":\"KEY-1\"} 0123456789";
54
55    /// Deterministic pseudo-random bytes (xorshift64), so no compressor can shrink them.
56    fn incompressible_payload(len: usize) -> Vec<u8> {
57        let mut state: u64 = 0x9E37_79B9_7F4A_7C15;
58        (0..len)
59            .map(|_| {
60                state ^= state << 13;
61                state ^= state >> 7;
62                state ^= state << 17;
63                state as u8
64            })
65            .collect()
66    }
67
68    fn assert_round_trip(payload: &[u8], level: i32) {
69        for (name, compressor) in COMPRESSORS {
70            let compressed = compressor
71                .compress(payload, level)
72                .unwrap_or_else(|e| panic!("{name} compress at level {level} failed: {e}"));
73            let decompressed = compressor
74                .decompress(&compressed)
75                .unwrap_or_else(|e| panic!("{name} decompress at level {level} failed: {e}"));
76            assert_eq!(decompressed.as_ref(), payload, "{name} round trip at level {level}");
77        }
78    }
79
80    #[test]
81    fn get_compressor_returns_same_singleton_for_same_type() {
82        for compression_type in ALL_TYPES {
83            let first = CompressorFactory::get_compressor(compression_type);
84            let second = CompressorFactory::get_compressor(compression_type);
85            assert!(
86                std::ptr::addr_eq(first, second),
87                "{compression_type:?} compressor should be a process-wide singleton"
88            );
89        }
90    }
91
92    #[test]
93    fn get_compressor_maps_each_type_to_matching_algorithm() {
94        for compression_type in ALL_TYPES {
95            let compressed = CompressorFactory::get_compressor(compression_type)
96                .compress(TYPICAL_PAYLOAD, 5)
97                .unwrap_or_else(|e| panic!("{compression_type:?} compress failed: {e}"));
98            let decompressed = compression_type.try_decompression(&compressed).unwrap_or_else(|e| {
99                panic!("{compression_type:?} output should be decodable by its own algorithm: {e}")
100            });
101            assert_eq!(
102                decompressed.as_ref(),
103                TYPICAL_PAYLOAD,
104                "{compression_type:?} factory mapping"
105            );
106        }
107    }
108
109    #[test]
110    fn round_trips_typical_payload() {
111        assert_round_trip(TYPICAL_PAYLOAD, 5);
112    }
113
114    #[test]
115    fn round_trips_empty_payload() {
116        assert_round_trip(&[], 5);
117    }
118
119    #[test]
120    fn round_trips_repetitive_payload_and_shrinks_it() {
121        let payload = vec![b'a'; 64 * 1024];
122        assert_round_trip(&payload, 5);
123        for (name, compressor) in COMPRESSORS {
124            let compressed = compressor
125                .compress(&payload, 5)
126                .expect("repetitive payload should compress");
127            assert!(
128                compressed.len() < payload.len(),
129                "{name} should shrink a repetitive payload, got {} bytes from {}",
130                compressed.len(),
131                payload.len()
132            );
133        }
134    }
135
136    #[test]
137    fn round_trips_incompressible_payload() {
138        assert_round_trip(&incompressible_payload(4096), 5);
139    }
140
141    #[test]
142    fn round_trips_at_level_bounds() {
143        for level in [0, 9] {
144            assert_round_trip(TYPICAL_PAYLOAD, level);
145            assert_round_trip(&incompressible_payload(1024), level);
146        }
147    }
148
149    #[test]
150    fn decompress_rejects_invalid_input_without_panic() {
151        for (name, compressor) in COMPRESSORS {
152            let error = compressor
153                .decompress(b"definitely not compressed data")
154                .expect_err("garbage input should be rejected");
155            // The old free-form "decompression failed" detail is dropped by canonical error
156            // rendering. The stable contract is the serialization descriptor, the decode of the
157            // compression format in the diagnostic context, and the retained compressor source.
158            assert_eq!(
159                error.descriptor(),
160                &rocketmq_error::CORE_SERIALIZATION_FAILED,
161                "{name} garbage input should be a serialization failure"
162            );
163            let diagnostic = error.diagnostic_view().expect("valid diagnostic error context");
164            let diagnostic_field = |field_name: &str| {
165                diagnostic
166                    .fields()
167                    .find(|field| field.name() == field_name)
168                    .map(|field| field.value())
169            };
170            assert_eq!(
171                diagnostic_field(fields::OPERATION_DIAGNOSTIC.schema().name()),
172                Some(ViewValueRef::Text("decode")),
173                "{name} failure should be classified as a decode operation"
174            );
175            assert_eq!(
176                diagnostic_field(fields::FORMAT.schema().name()),
177                Some(ViewValueRef::Text("compression")),
178                "{name} failure should be classified against the compression format"
179            );
180            assert!(
181                std::error::Error::source(&error).is_some(),
182                "{name} compressor detail should stay available as a typed source"
183            );
184        }
185    }
186}
rocketmq_protocol/common/compression/compressor_factory.rs

rocketmq_protocol/common/compression/
compressor_factory.rs
