feat: Add Snappy support

crates/async-compression/Cargo.toml

@@ -30,6 +30,7 @@ all-algorithms = [
     "xz-parallel",
     "zlib",
     "zstd",
+    "snappy",
 ]
 
 # algorithms
@@ -46,6 +47,7 @@ xz2 = ["compression-codecs/xz2", "xz"]
 zlib = ["compression-codecs/zlib"]
 zstd = ["compression-codecs/zstd"]
 zstdmt = ["compression-codecs/zstdmt", "zstd"]
+snappy = ["compression-codecs/snappy"]
 
 
 [dependencies]
@@ -72,6 +74,7 @@ tokio = { version = "1.38.2", default-features = false, features = [
     "macros",
     "rt-multi-thread",
     "io-std",
+    "fs"
 ] }
 tokio-util = { version = "0.7", default-features = false, features = ["io"] }
 
@@ -83,6 +86,7 @@ lz4 = "1.28.1"
 liblzma = "0.4.2"
 zstd-safe = { version = "7", default-features = false }
 deflate64 = "0.1.5"
+snap = "1"
 
 [lints]
 workspace = true
@@ -131,6 +135,10 @@ required-features = ["zstd", "tokio"]
 name = "zstd-window-size"
 required-features = ["zstd", "tokio"]
 
+[[test]]
+name = "snappy"
+required-features = ["snappy"]
+
 [[example]]
 name = "zlib_tokio_write"
 required-features = ["zlib", "tokio"]

crates/async-compression/src/macros.rs

@@ -375,5 +375,19 @@ macro_rules! algos {
         { @dec }
         );
 
+        algos!(@algo snappy ["snappy"] SnappyDecoder SnappyEncoder <$inner>
+        { @enc
+
+            pub fn with_quality(inner: $inner, _level: crate::core::Level) -> Self {
+                Self {
+                    inner: crate::$($mod::)+generic::Encoder::new(
+                        inner,
+                        crate::codecs::SnappyEncoder::new()
+                    ),
+                }
+            }
+        }
+        { @dec }
+        );
     }
 }

crates/async-compression/tests/proptest.rs

@@ -142,4 +142,7 @@ mod proptest {
 
     #[cfg(feature = "zstd")]
     tests!(zstd);
+
+    #[cfg(feature = "snappy")]
+    tests!(snappy);
 }

crates/async-compression/tests/snappy.proptest-regressions

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 3782cfa243f6ba3a6ca7c7480392b0df7d8c859c44643222851eceba064dd625 # shrinks to ref input = InputStream([[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]), limit = 1

crates/async-compression/tests/snappy.rs

@@ -0,0 +1,55 @@
+#[macro_use]
+mod utils;
+
+test_cases!(snappy);
+
+#[cfg(all(feature = "snappy", feature = "tokio"))]
+mod snappy_additional_tests {
+    use crate::utils::algos::snappy::sync;
+    use crate::utils::algos::snappy::tokio;
+    use crate::utils::InputStream;
+    use proptest::arbitrary::any;
+    use proptest::collection::vec as prop_vec;
+    use proptest::prelude::Strategy;
+    use proptest::proptest;
+    use std::iter::FromIterator;
+
+    /// A random short seed repeated a random number of times to create a compressible payload
+    fn compressible_bytes() -> impl Strategy<Value = Vec<u8>> {
+        (prop_vec(any::<u8>(), 1..=32), 1usize..=8192)
+            .prop_map(|(seed, n)| seed.iter().copied().cycle().take(seed.len() * n).collect())
+    }
+
+    /// The same payload, sliced into an `InputStream` with a random chunk size
+    fn compressible_input_stream() -> impl Strategy<Value = InputStream> {
+        (compressible_bytes(), 1usize..=16_384).prop_map(|(bytes, chunk)| {
+            InputStream::new(bytes.chunks(chunk).map(<[u8]>::to_vec).collect())
+        })
+    }
+
+    proptest! {
+
+        #[test]
+        fn write_compress_compressible_input(ref input in compressible_input_stream(), limit in 1..20usize) {
+            let compressed = tokio::write::compress(input.as_ref(), limit);
+            let output = sync::decompress(&compressed);
+            assert_eq!(output, input.bytes())
+        }
+
+        #[test]
+        fn bufread_compress_compressible_input(ref input in compressible_input_stream()) {
+            let buf = tokio::bufread::from(input);
+            let compressed = tokio::bufread::compress(buf);
+            let output = sync::decompress(&compressed);
+            assert_eq!(output, input.bytes())
+        }
+
+        #[test]
+        fn decompressed_compressible_input(ref bytes in compressible_bytes(), chunk_size in 1..20usize) {
+            let compressed = sync::compress(bytes);
+            let input = InputStream::from(Vec::from_iter(compressed.chunks(chunk_size).map(Vec::from)));
+            let output = tokio::bufread::decompress(tokio::bufread::from(&input));
+            assert_eq!(&output, bytes);
+        }
+    }
+}

crates/async-compression/tests/utils/algos.rs

@@ -229,6 +229,33 @@ algos! {
             }
         }
     }
+
+    pub mod snappy("snappy", SnappyEncoder, SnappyDecoder) {
+        pub mod sync {
+            pub use crate::utils::impls::sync::to_vec;
+
+            pub fn compress(bytes: &[u8]) -> Vec<u8> {
+                if bytes.is_empty() {
+                    return vec![0xff, 0x06, 0x00, 0x00, b's', b'N', b'a', b'P', b'p', b'Y'];
+                }
+
+                use std::io::Write;
+                use snap::write::FrameEncoder;
+
+                let mut output = Vec::new();
+                {
+                    let mut encoder = FrameEncoder::new(&mut output);
+                    encoder.write_all(bytes).unwrap();
+                }
+                output
+            }
+
+            pub fn decompress(bytes: &[u8]) -> Vec<u8> {
+                use snap::read::FrameDecoder;
+                to_vec(FrameDecoder::new(bytes))
+            }
+        }
+    }
 }
 
 macro_rules! io_algo_parallel {

crates/async-compression/tests/utils/test_cases.rs

@@ -176,6 +176,7 @@ macro_rules! io_test_cases {
 
                     #[test]
                     #[ntest::timeout(1000)]
+                    #[cfg(not(feature = "snappy"))]
                     fn trailer() {
                         let mut compressed = sync::compress(&[1, 2, 3, 4, 5, 6]);
 

crates/compression-codecs/Cargo.toml

@@ -28,6 +28,7 @@ all-algorithms = [
     "zlib",
     "zstd",
     "deflate64",
+    "snappy"
 ]
 
 # algorithms
@@ -42,6 +43,7 @@ zlib = ["flate2"]
 zstd = ["libzstd", "zstd-safe"]
 zstdmt = ["zstd", "zstd-safe/zstdmt"]
 deflate64 = ["dep:deflate64"]
+snappy = ["snap", "crc32c"]
 
 [dependencies]
 # Workspace dependencies.
@@ -56,6 +58,8 @@ lz4 = { version = "1.28.1", optional = true }
 liblzma = { version = "0.4.5", optional = true }
 memchr = { version = "2", optional = true }
 zstd-safe = { version = "7", optional = true, default-features = false }
+snap = { version = "1", optional = true, default-features = false }
+crc32c = { version = "0.6.8", optional = true, default-features = false }
 
 [lints]
 workspace = true

crates/compression-codecs/src/lib.rs

@@ -22,6 +22,8 @@ pub mod gzip;
 pub mod lz4;
 #[cfg(feature = "lzma")]
 pub mod lzma;
+#[cfg(feature = "snappy")]
+pub mod snappy;
 #[cfg(feature = "xz")]
 pub mod xz;
 #[cfg(feature = "lzma")]
@@ -49,6 +51,8 @@ pub use self::gzip::{GzipDecoder, GzipEncoder};
 pub use self::lz4::{Lz4Decoder, Lz4Encoder};
 #[cfg(feature = "lzma")]
 pub use self::lzma::{LzmaDecoder, LzmaEncoder};
+#[cfg(feature = "snappy")]
+pub use self::snappy::{SnappyDecoder, SnappyEncoder};
 #[cfg(feature = "xz")]
 pub use self::xz::{XzDecoder, XzEncoder};
 #[cfg(feature = "lzma")]