Repository: kafka-rust/kafka-rust (language: rust)

Issue:
Potentially switch to an external snappy crate?
I saw there are a few snappy crates:
- https://crates.io/crates/snappy_framed
- https://crates.io/crates/snappy
- https://crates.io/crates/rsnappy

The last one is especially interesting since it's a pure-rust implementation of snappy. Would any of them work for this project?



Discussion / hints:
hello,

i did some work on "unpacking" snappy compressed messages in `kafka-rust` and feel confident to provide an answer here.

first of all, `kafka-rust` is bound to the format spilled out by [SnappyOutputStream](https://github.com/xerial/snappy-java/blob/develop/src/main/java/org/xerial/snappy/SnappyOutputStream.java) which is used at least by Kafka 0.8.2.x.  this is basically a concatenation of snappy compressed blocks - [unfortunately not snappy frames.](https://github.com/google/snappy/blob/master/framing_format.txt)
- so far i can see, we cannot use [snappy_framed](https://crates.io/crates/snappy_framed) at the moment due to the format of the compressed data Kafka produces.
- [snappy](https://crates.io/crates/snappy) is a thin ffi wrapper around the c/c++ library which is exactly what `kafka-rust` does. in other words, with some minor changes this library could in fact be used as a drop-in replacement for the [corresponding code](https://github.com/spicavigo/kafka-rust/blob/master/src/snappy.rs#L77) in kafka-rust. however, in [branch-0.2.x](https://github.com/spicavigo/kafka-rust/blob/branch-0.2.x/src/compression/snappy.rs#L27) - hopefully to be released in a few weeks - i extended the core decompression function a little bit to avoid some extra copying. so far, i'm quite OK with the performance. however, such changes could be intergrated into `snappy`, of course. for me, `snappy` looks like a good candidate.
- i have not looked in detail at [rsnappy](https://crates.io/crates/rsnappy) yet, but right now it seems to me like it provides an API which would be suitable to us. the fact it's written in rust is indeed very interesting. i might have some experiments with it.

if you'd like to experiment yourself, feel free to go ahead and submit a pull request. for decompressing snappy messages we have unit tests as well as benchmarks (requires rust nightly) set up in `master` as well as the already mentioned `branch-0.2.x`.

1) merely replacing the existing snappy decompression code in `kafka-rust` with `rsnappy` leads to benchmarks indicating a performance loss by a factor of two to three on my machine. but it might well be i've not used `rsnappy` effectively.

2) further it seems, that `rsnappy` is not very robust against invalid input data.

3) i've pushed the changes to the `rsnappy` branch for the moment.

@veddan do you have any measurements comparing the performance of `rsnappy` to https://github.com/google/snappy ?

fyi: once JeffBelgum/rust-snappy#1 is merged and released, we'll switch over to that `snappy` crate to get rid of our own snappy native bindings. the corresponding changes to kafka-rust are prepared in the `external-snappy-dep` branch for the moment.

Switched over to the `snappy` crate with 930efc9e2336b6ab9a920839dbacbf3c7188b4ec. This change will be released as part of 0.2, since it requires an update of the libc dependency.

just pushed version 0.2.0 to crates.io with a dependency on `snappy:0.3.0`. feel free to re-open if there's anything left from your side.


Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 2d343f461..bf1788a32 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -19,10 +19,9 @@ ref_slice = "1.0"
 fnv = "1.0"
 
 flate2 = { version = "0.2", optional = true }
-snappy = { version = "0.4", optional = true }
+snap = { version = "0.2", optional = true }
 openssl = { version = "0.9", optional = true }
 
-
 [dev-dependencies]
 getopts = "0.2"
 env_logger = { version = "0.3", default-features = false }
@@ -31,6 +30,7 @@ time = "0.1"
 
 [features]
 default = ["snappy", "gzip", "security"]
+snappy = ["snap"]
 gzip = ["flate2"]
 security = ["openssl"]
 nightly = []
diff --git a/src/compression/snappy.rs b/src/compression/snappy.rs
index 1260642e3..155a457cf 100644
--- a/src/compression/snappy.rs
+++ b/src/compression/snappy.rs
@@ -1,18 +1,37 @@
 use std::io::{self, Read};
 
 use byteorder::{BigEndian, ByteOrder};
-use snappy;
+use snap;
 
 use error::{Result, Error};
 
 pub fn compress(src: &[u8]) -> Result<Vec<u8>> {
-    Ok(snappy::compress(src))
+    let mut buf = vec![0; snap::max_compress_len(src.len())];
+
+    snap::Encoder::new()
+        .compress(src, &mut buf)
+        .map(|len| {
+                 buf.truncate(len);
+                 buf
+             })
+        .map_err(|err| Error::InvalidSnappy(err))
 }
 
 fn uncompress_to(src: &[u8], dst: &mut Vec<u8>) -> Result<()> {
-    snappy::uncompress_to(src, dst)
-        .map(|_| ())
-        .map_err(|_| Error::InvalidInputSnappy)
+    snap::decompress_len(src)
+        .and_then(|min_len| {
+            if min_len > 0 {
+                let off = dst.len();
+                dst.resize(off + min_len, 0);
+                let uncompressed_len = {
+                    let buf = &mut dst.as_mut_slice()[off..off + min_len];
+                    try!(snap::Decoder::new().decompress(src, buf))
+                };
+                dst.truncate(off + uncompressed_len);
+            }
+            Ok(())
+        })
+        .map_err(|err| Error::InvalidSnappy(err))
 }
 
 // --------------------------------------------------------------------
@@ -43,18 +62,18 @@ fn validate_stream(mut stream: &[u8]) -> Result<&[u8]> {
         return Err(Error::UnexpectedEOF);
     }
     if &stream[..MAGIC.len()] != MAGIC {
-        return Err(Error::InvalidInputSnappy);
+        return Err(Error::InvalidSnappy(snap::Error::Header));
     }
     stream = &stream[MAGIC.len()..];
     // ~ let's be assertive and (for the moment) restrict ourselves to
     // version == 1 and compatibility == 1.
     let version = next_i32!(stream);
     if version != 1 {
-        return Err(Error::InvalidInputSnappy);
+        return Err(Error::InvalidSnappy(snap::Error::Header));
     }
     let compat = next_i32!(stream);
     if compat != 1 {
-        return Err(Error::InvalidInputSnappy);
+        return Err(Error::InvalidSnappy(snap::Error::Header));
     }
     Ok(stream)
 }
@@ -87,10 +106,10 @@ impl<'a> SnappyReader<'a> {
     pub fn new(mut stream: &[u8]) -> Result<SnappyReader> {
         stream = try!(validate_stream(stream));
         Ok(SnappyReader {
-            compressed_data: stream,
-            uncompressed_pos: 0,
-            uncompressed_chunk: Vec::new(),
-        })
+               compressed_data: stream,
+               uncompressed_pos: 0,
+               uncompressed_chunk: Vec::new(),
+           })
     }
 
     fn _read(&mut self, buf: &mut [u8]) -> Result<usize> {
@@ -116,7 +135,10 @@ impl<'a> SnappyReader<'a> {
         self.uncompressed_pos = 0;
         let chunk_size = next_i32!(self.compressed_data);
         if chunk_size <= 0 {
-            return Err(Error::InvalidInputSnappy);
+            return Err(Error::InvalidSnappy(snap::Error::UnsupportedChunkLength {
+                                                len: chunk_size as u64,
+                                                header: false,
+                                            }));
         }
         let chunk_size = chunk_size as usize;
         self.uncompressed_chunk.clear();
@@ -137,7 +159,10 @@ impl<'a> SnappyReader<'a> {
         while !self.compressed_data.is_empty() {
             let chunk_size = next_i32!(self.compressed_data);
             if chunk_size <= 0 {
-                return Err(Error::InvalidInputSnappy);
+                return Err(Error::InvalidSnappy(snap::Error::UnsupportedChunkLength {
+                                                    len: chunk_size as u64,
+                                                    header: false,
+                                                }));
             }
             let (c1, c2) = self.compressed_data.split_at(chunk_size as usize);
             try!(uncompress_to(c1, buf));
@@ -173,6 +198,7 @@ impl<'a> Read for SnappyReader<'a> {
 
 #[cfg(test)]
 mod tests {
+    use std::str;
     use std::io::Read;
 
     use error::{Error, Result};
@@ -208,11 +234,11 @@ mod tests {
         let compressed = &[12, 42, 84, 104, 105, 115, 32, 105, 115, 32, 116, 101, 115, 116];
         let uncompressed = uncompress(compressed);
         assert!(uncompressed.is_err());
-        assert!(if let Some(Error::InvalidInputSnappy) = uncompressed.err() {
-            true
-        } else {
-            false
-        });
+        assert!(if let Some(Error::InvalidSnappy(_)) = uncompressed.err() {
+                    true
+                } else {
+                    false
+                });
     }
 
     static ORIGINAL: &'static str = include_str!("../../test-data/fetch1.txt");
@@ -230,7 +256,7 @@ mod tests {
                 n => buf.extend_from_slice(&tmp_buf[..n]),
             }
         }
-        assert_eq!(ORIGINAL.as_bytes(), &buf[..]);
+        assert_eq!(ORIGINAL, str::from_utf8(&buf[..]).unwrap());
     }
 
     #[test]
@@ -238,6 +264,6 @@ mod tests {
         let mut buf = Vec::new();
         let mut r = SnappyReader::new(COMPRESSED).unwrap();
         r.read_to_end(&mut buf).unwrap();
-        assert_eq!(ORIGINAL.as_bytes(), &buf[..]);
+        assert_eq!(ORIGINAL, str::from_utf8(&buf[..]).unwrap());
     }
 }
diff --git a/src/error.rs b/src/error.rs
index d633b7bcd..0ecac46ae 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -48,9 +48,9 @@ pub enum Error {
     /// due to an unsupported compression format of the data
     UnsupportedCompression,
 
-    /// Failure to decode a snappy compressed response from Kafka
+    /// Failure to encode/decode a snappy compressed response from Kafka
     #[cfg(feature = "snappy")]
-    InvalidInputSnappy,
+    InvalidSnappy(::snap::Error),
 
     /// Failure to decode a response due to an insufficient number of bytes available
     UnexpectedEOF,
@@ -224,6 +224,13 @@ impl<S> From<ssl::HandshakeError<S>> for Error {
     }
 }
 
+#[cfg(feature = "snappy")]
+impl From<::snap::Error> for Error {
+    fn from(err: ::snap::Error) -> Error {
+        Error::InvalidSnappy(err)
+    }
+}
+
 impl Clone for Error {
     fn clone(&self) -> Error {
         match self {
@@ -239,7 +246,7 @@ impl Clone for Error {
             &Error::UnsupportedProtocol => Error::UnsupportedProtocol,
             &Error::UnsupportedCompression => Error::UnsupportedCompression,
             #[cfg(feature = "snappy")]
-            &Error::InvalidInputSnappy => Error::InvalidInputSnappy,
+            &Error::InvalidSnappy(ref err) => from_snap_error_ref(err),
             &Error::UnexpectedEOF => Error::UnexpectedEOF,
             &Error::CodecError => Error::CodecError,
             &Error::StringDecodeError => Error::StringDecodeError,
@@ -262,6 +269,85 @@ fn from_sslerror_ref(err: &ssl::Error) -> Error {
     }
 }
 
+#[cfg(feature = "snappy")]
+fn from_snap_error_ref(err: &::snap::Error) -> Error {
+    match err {
+        &::snap::Error::TooBig { given, max } => {
+            Error::InvalidSnappy(::snap::Error::TooBig {
+                                     given: given,
+                                     max: max,
+                                 })
+        }
+        &::snap::Error::BufferTooSmall { given, min } => {
+            Error::InvalidSnappy(::snap::Error::BufferTooSmall {
+                                     given: given,
+                                     min: min,
+                                 })
+        }
+        &::snap::Error::Empty => Error::InvalidSnappy(::snap::Error::Empty),
+        &::snap::Error::Header => Error::InvalidSnappy(::snap::Error::Header),
+        &::snap::Error::HeaderMismatch {
+             expected_len,
+             got_len,
+         } => {
+            Error::InvalidSnappy(::snap::Error::HeaderMismatch {
+                                     expected_len: expected_len,
+                                     got_len: got_len,
+                                 })
+        }
+        &::snap::Error::Literal {
+             len,
+             src_len,
+             dst_len,
+         } => {
+            Error::InvalidSnappy(::snap::Error::Literal {
+                                     len: len,
+                                     src_len: src_len,
+                                     dst_len: dst_len,
+                                 })
+        }
+        &::snap::Error::CopyRead { len, src_len } => {
+            Error::InvalidSnappy(::snap::Error::CopyRead {
+                                     len: len,
+                                     src_len: src_len,
+                                 })
+        }
+        &::snap::Error::CopyWrite { len, dst_len } => {
+            Error::InvalidSnappy(::snap::Error::CopyWrite {
+                                     len: len,
+                                     dst_len: dst_len,
+                                 })
+        }
+        &::snap::Error::Offset { offset, dst_pos } => {
+            Error::InvalidSnappy(::snap::Error::Offset {
+                                     offset: offset,
+                                     dst_pos: dst_pos,
+                                 })
+        }
+        &::snap::Error::StreamHeader { byte } => {
+            Error::InvalidSnappy(::snap::Error::StreamHeader { byte: byte })
+        }
+        &::snap::Error::StreamHeaderMismatch { ref bytes } => {
+            Error::InvalidSnappy(::snap::Error::StreamHeaderMismatch { bytes: bytes.clone() })
+        }
+        &::snap::Error::UnsupportedChunkType { byte } => {
+            Error::InvalidSnappy(::snap::Error::UnsupportedChunkType { byte: byte })
+        }
+        &::snap::Error::UnsupportedChunkLength { len, header } => {
+            Error::InvalidSnappy(::snap::Error::UnsupportedChunkLength {
+                                     len: len,
+                                     header: header,
+                                 })
+        }
+        &::snap::Error::Checksum { expected, got } => {
+            Error::InvalidSnappy(::snap::Error::Checksum {
+                                     expected: expected,
+                                     got: got,
+                                 })
+        }
+    }
+}
+
 /// Attempt to clone `io::Error`.
 fn clone_ioe(e: &io::Error) -> io::Error {
     match e.raw_os_error() {
@@ -283,7 +369,7 @@ impl error::Error for Error {
             Error::UnsupportedProtocol => "Unsupported protocol version",
             Error::UnsupportedCompression => "Unsupported compression format",
             #[cfg(feature = "snappy")]
-            Error::InvalidInputSnappy => "Snappy decode error",
+            Error::InvalidSnappy(ref err) => error::Error::description(err),
             Error::UnexpectedEOF => "Unexpected EOF",
             Error::CodecError => "Encoding/Decoding error",
             Error::StringDecodeError => "String decoding error",
@@ -318,7 +404,7 @@ impl fmt::Display for Error {
             Error::UnsupportedProtocol => write!(f, "Unsupported protocol version"),
             Error::UnsupportedCompression => write!(f, "Unsupported compression format"),
             #[cfg(feature = "snappy")]
-            Error::InvalidInputSnappy => write!(f, "Snappy decode error"),
+            Error::InvalidSnappy(ref err) => write!(f, "Snappy error, {}", err),
             Error::UnexpectedEOF => write!(f, "Unexpected EOF"),
             Error::CodecError => write!(f, "Encoding/Decoding Error"),
             Error::StringDecodeError => write!(f, "String decoding error"),
diff --git a/src/lib.rs b/src/lib.rs
index 9c2052b56..7c63d5bea 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -25,7 +25,7 @@ extern crate openssl;
 extern crate flate2;
 
 #[cfg(feature = "snappy")]
-extern crate snappy;
+extern crate snap;
 
 #[cfg(all(test, feature = "nightly"))]
 extern crate test;


Test patch: