add decompressed gzip stream support, tests

and a bit of refactoring adding the bytes

Author: fairingrey

src/stream/flate.rs

@@ -94,7 +94,7 @@ impl<S: Stream<Item = Result<Bytes>>> Stream for DecompressedStream<S> {
             if *this.flushing {
                 return Poll::Ready(None);
             } else if let Some(bytes) = ready!(this.inner.poll_next(cx)) {
-                *this.input_buffer = bytes?;
+                this.input_buffer.extend_from_slice(&bytes?);
             } else {
                 *this.flushing = true;
             }

src/stream/gzip.rs

@@ -2,11 +2,11 @@ use core::{
     pin::Pin,
     task::{Context, Poll},
 };
-use std::io::Result;
+use std::io::{Error, ErrorKind, Result};
 
 use bytes::{Bytes, BytesMut};
 pub use flate2::Compression;
-use flate2::{Compress, Crc, Decompress, FlushCompress};
+use flate2::{Compress, Crc, Decompress, FlushCompress, FlushDecompress};
 use futures::{ready, stream::Stream};
 use pin_project::unsafe_project;
 
@@ -42,15 +42,15 @@ impl<S: Stream<Item = Result<Bytes>>> Stream for GzipStream<S> {
             if *this.flushing {
                 if !*this.footer_appended {
                     let mut footer = Bytes::from(&this.crc.sum().to_le_bytes()[..]);
-                    let length_read = &this.crc.amount().to_le_bytes()[..];
-                    footer.extend_from_slice(length_read);
+                    let bytes_read = &this.crc.amount().to_le_bytes()[..];
+                    footer.extend_from_slice(bytes_read);
                     *this.footer_appended = true;
                     return Poll::Ready(Some(Ok(footer)));
                 } else {
                     return Poll::Ready(None);
                 }
             } else if let Some(bytes) = ready!(this.inner.poll_next(cx)) {
-                *this.input_buffer = bytes?;
+                this.input_buffer.extend_from_slice(&bytes?);
             } else {
                 *this.flushing = true;
             }
@@ -121,7 +121,6 @@ pub struct DecompressedGzipStream<S: Stream<Item = Result<Bytes>>> {
     output_buffer: BytesMut,
     crc: Crc,
     header_stripped: bool,
-    footer_stripped: bool,
     decompress: Decompress,
 }
 
@@ -133,17 +132,63 @@ impl<S: Stream<Item = Result<Bytes>>> Stream for DecompressedGzipStream<S> {
 
         let this = self.project();
 
-        if this.input_buffer.is_empty() {
+        if this.input_buffer.len() <= 8 {
             if *this.flushing {
+                // check crc and len in the footer
+                let crc = &this.crc.sum().to_le_bytes()[..];
+                let bytes_read = &this.crc.amount().to_le_bytes()[..];
+                if crc != this.input_buffer.slice(0, 4) {
+                    return Poll::Ready(Some(Err(Error::new(
+                        ErrorKind::InvalidData,
+                        "CRC computed does not match",
+                    ))));
+                } else if bytes_read != this.input_buffer.slice(4, 8) {
+                    return Poll::Ready(Some(Err(Error::new(
+                        ErrorKind::InvalidData,
+                        "amount of bytes read does not match",
+                    ))));
+                }
                 return Poll::Ready(None);
             } else if let Some(bytes) = ready!(this.inner.poll_next(cx)) {
-                *this.input_buffer = bytes?;
+                this.input_buffer.extend_from_slice(&bytes?);
             } else {
                 *this.flushing = true;
             }
         }
 
-        unimplemented!()
+        if !*this.header_stripped {
+            let header = this.input_buffer.split_to(10);
+            if header[0..3] != [0x1f, 0x8b, 0x08] {
+                return Poll::Ready(Some(Err(Error::new(
+                    ErrorKind::InvalidData,
+                    "Invalid file header",
+                ))));
+            }
+            *this.header_stripped = true;
+            return Poll::Ready(Some(Ok(Bytes::new())));
+        }
+
+        this.output_buffer.resize(OUTPUT_BUFFER_SIZE, 0);
+
+        let flush = if *this.flushing {
+            FlushDecompress::Finish
+        } else {
+            FlushDecompress::None
+        };
+
+        let (prior_in, prior_out) = (this.decompress.total_in(), this.decompress.total_out());
+        this.decompress
+            .decompress(this.input_buffer, this.output_buffer, flush)?;
+        let input = this.decompress.total_in() - prior_in;
+        let output = this.decompress.total_out() - prior_out;
+
+        this.crc.update(&this.output_buffer[..output as usize]);
+        this.input_buffer.advance(input as usize);
+
+        Poll::Ready(Some(Ok(this
+            .output_buffer
+            .split_to(output as usize)
+            .freeze())))
     }
 }
 
@@ -156,7 +201,6 @@ impl<S: Stream<Item = Result<Bytes>>> DecompressedGzipStream<S> {
             output_buffer: BytesMut::new(),
             crc: Crc::new(),
             header_stripped: false,
-            footer_stripped: false,
             decompress: Decompress::new(false),
         }
     }

tests/brotli.rs

@@ -2,7 +2,6 @@ use brotli2::bufread::BrotliDecoder;
 use bytes::Bytes;
 use futures::{
     executor::block_on,
-    //io::AsyncReadExt,
     stream::{self, StreamExt},
 };
 use std::io::{self, Read};
@@ -27,18 +26,19 @@ fn brotli_stream() {
     assert_eq!(output, vec![1, 2, 3, 4, 5, 6]);
 }
 
-//#[test]
-//fn brotli_read() {
-//    use async_compression::read::brotli;
-//
-//    let input = &[1, 2, 3, 4, 5, 6];
-//    let compress = brotli::Compress::new();
-//    let mut compressed = brotli::compress_read(&input[..], compress);
-//    let mut data = vec![];
-//    block_on(compressed.read_to_end(&mut data)).unwrap();
-//    let mut output = vec![];
-//    BrotliDecoder::new(&data[..])
-//        .read_to_end(&mut output)
-//        .unwrap();
-//    assert_eq!(output, input);
-//}
+#[test]
+fn decompressed_brotli_stream() {
+    use async_compression::stream::brotli;
+
+    let stream = stream::iter(vec![
+        Bytes::from_static(&[1, 2, 3]),
+        Bytes::from_static(&[4, 5, 6]),
+    ]);
+    let compress = brotli::Compress::new();
+    let compressed = brotli::BrotliStream::new(stream.map(Ok), compress);
+    let decompressed = brotli::DecompressedBrotliStream::new(compressed);
+    let data: Vec<_> = block_on(decompressed.collect());
+    let data: io::Result<Vec<_>> = data.into_iter().collect();
+    let data: Vec<u8> = data.unwrap().into_iter().flatten().collect();
+    assert_eq!(data, vec![1, 2, 3, 4, 5, 6]);
+}

tests/deflate.rs

@@ -26,6 +26,23 @@ fn deflate_stream() {
     assert_eq!(output, vec![1, 2, 3, 4, 5, 6]);
 }
 
+#[test]
+fn decompressed_deflate_stream() {
+    use async_compression::stream::deflate;
+
+    let stream = stream::iter(vec![
+        Bytes::from_static(&[1, 2, 3]),
+        Bytes::from_static(&[4, 5, 6]),
+    ]);
+    let compressed = deflate::DeflateStream::new(stream.map(Ok), deflate::Compression::default());
+    let decompressed = deflate::DecompressedDeflateStream::new(compressed);
+    let data: Vec<_> = block_on(decompressed.collect());
+    let data: io::Result<Vec<_>> = data.into_iter().collect();
+    let data: Vec<u8> = data.unwrap().into_iter().flatten().collect();
+
+    assert_eq!(data, vec![1, 2, 3, 4, 5, 6]);
+}
+
 #[test]
 fn deflate_read() {
     use async_compression::read::deflate;

tests/gzip.rs

@@ -22,3 +22,51 @@ fn gzip_stream() {
     GzDecoder::new(&data[..]).read_to_end(&mut output).unwrap();
     assert_eq!(output, vec![1, 2, 3, 4, 5, 6]);
 }
+
+#[test]
+fn decompressed_gzip_stream_1() {
+    use async_compression::stream::gzip;
+
+    let stream = stream::iter(vec![
+        Bytes::from_static(&[1, 2, 3]),
+        Bytes::from_static(&[4, 5, 6]),
+    ]);
+    let compressed = gzip::GzipStream::new(stream.map(Ok), gzip::Compression::default());
+    let decompressed = gzip::DecompressedGzipStream::new(compressed);
+    let data: Vec<_> = block_on(decompressed.collect());
+    let data: io::Result<Vec<_>> = data.into_iter().collect();
+    let data: Vec<u8> = data.unwrap().into_iter().flatten().collect();
+    assert_eq!(data, vec![1, 2, 3, 4, 5, 6]);
+}
+
+#[test]
+fn decompressed_gzip_stream_2() {
+    use async_compression::stream::gzip;
+
+    let stream = stream::iter(vec![
+        Bytes::from_static(&[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]),
+        Bytes::from_static(&[11, 12, 13, 14, 15]),
+    ]);
+    let compressed = gzip::GzipStream::new(stream.map(Ok), gzip::Compression::default());
+    let decompressed = gzip::DecompressedGzipStream::new(compressed);
+    let data: Vec<_> = block_on(decompressed.collect());
+    let data: io::Result<Vec<_>> = data.into_iter().collect();
+    let data: Vec<u8> = data.unwrap().into_iter().flatten().collect();
+    assert_eq!(
+        data,
+        vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
+    );
+}
+
+#[test]
+fn decompressed_gzip_stream_3() {
+    use async_compression::stream::gzip;
+
+    let stream = stream::iter(vec![Bytes::from_static(&[])]);
+    let compressed = gzip::GzipStream::new(stream.map(Ok), gzip::Compression::default());
+    let decompressed = gzip::DecompressedGzipStream::new(compressed);
+    let data: Vec<_> = block_on(decompressed.collect());
+    let data: io::Result<Vec<_>> = data.into_iter().collect();
+    let data: Vec<u8> = data.unwrap().into_iter().flatten().collect();
+    assert_eq!(data, vec![]);
+}

tests/zlib.rs

@@ -26,6 +26,23 @@ fn zlib_stream() {
     assert_eq!(output, vec![1, 2, 3, 4, 5, 6]);
 }
 
+#[test]
+fn decompressed_zlib_stream() {
+    use async_compression::stream::zlib;
+
+    let stream = stream::iter(vec![
+        Bytes::from_static(&[1, 2, 3]),
+        Bytes::from_static(&[4, 5, 6]),
+    ]);
+    let compressed = zlib::ZlibStream::new(stream.map(Ok), zlib::Compression::default());
+    let decompressed = zlib::DecompressedZlibStream::new(compressed);
+    let data: Vec<_> = block_on(decompressed.collect());
+    let data: io::Result<Vec<_>> = data.into_iter().collect();
+    let data: Vec<u8> = data.unwrap().into_iter().flatten().collect();
+
+    assert_eq!(data, vec![1, 2, 3, 4, 5, 6]);
+}
+
 #[test]
 fn zlib_read() {
     use async_compression::read::zlib;