correct multistream decompression EOF handling (fix #61) (#74)

* correct multistream decompression EOF handling (fix #61)

when the input buffer has an end-of-stream marker right at the
beginning, decompress() will return StreamEnd and total_in will not
advance. We cannot return Ok(read) as this would signal EOF. Instead, we
rely on the next loop iteration to really return EOF when the input
buffer did not fill again.

* add regression test for issue #61

* fix formatting

src/bufread.rs

@@ -192,33 +192,44 @@ impl<R> BzDecoder<R> {
 
 impl<R: BufRead> Read for BzDecoder<R> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        if self.done {
-            return Ok(0);
-        }
         loop {
-            let (read, consumed, eof, ret);
+            if self.done && !self.multi {
+                return Ok(0);
+            }
+            let (read, consumed, remaining, ret);
             {
                 let input = self.obj.fill_buf()?;
-                eof = input.is_empty();
+                if self.done {
+                    assert!(self.multi);
+                    if input.is_empty() {
+                        // beyond last stream in multi-stream case
+                        return Ok(0);
+                    } else {
+                        // previous stream ended, more data follows => create new decompressor
+                        self.data = Decompress::new(false);
+                        self.done = false;
+                    }
+                }
                 let before_out = self.data.total_out();
                 let before_in = self.data.total_in();
                 ret = self.data.decompress(input, buf);
                 read = (self.data.total_out() - before_out) as usize;
                 consumed = (self.data.total_in() - before_in) as usize;
+                remaining = input.len() - consumed;
             }
             self.obj.consume(consumed);
 
             let ret = ret.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
             if ret == Status::StreamEnd {
-                if !eof && self.multi {
-                    self.data = Decompress::new(false);
-                } else {
-                    self.done = true;
-                }
-
-                return Ok(read);
+                self.done = true;
+            } else if consumed == 0 && remaining == 0 && read == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "decompression not finished but EOF reached",
+                ));
             }
-            if read > 0 || eof || buf.len() == 0 {
+
+            if read > 0 || buf.len() == 0 {
                 return Ok(read);
             }
         }
@@ -304,3 +315,22 @@ impl<R: AsyncWrite + BufRead> AsyncWrite for MultiBzDecoder<R> {
         self.get_mut().shutdown()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::MultiBzDecoder;
+    use std::io::{BufReader, Read};
+
+    #[test]
+    fn bug_61() {
+        let compressed_bytes = include_bytes!("../tests/bug_61.bz2");
+        let uncompressed_bytes = include_bytes!("../tests/bug_61.raw");
+        let reader = BufReader::with_capacity(8192, compressed_bytes.as_ref());
+
+        let mut d = MultiBzDecoder::new(reader);
+        let mut data = Vec::new();
+
+        assert_eq!(d.read_to_end(&mut data).unwrap(), uncompressed_bytes.len());
+        assert_eq!(data, uncompressed_bytes);
+    }
+}