fix: attempt to decode from internal state even if nothing was read

Unlike zlib, miniz_oxide consumes data into internal state
even if there is not enough space in the output buffer.
Next time poll_fill_buf() is called we should try
to decode from internal state even if no new compressed data
was read.

src/codec/flate/decoder.rs

@@ -51,7 +51,7 @@ impl Decode for FlateDecoder {
         match self.decode(input, output, FlushDecompress::None)? {
             Status::Ok => Ok(false),
             Status::StreamEnd => Ok(true),
-            Status::BufError => Err(io::Error::new(io::ErrorKind::Other, "unexpected BufError")),
+            Status::BufError => Ok(true), // Waiting for more input.
         }
     }
 

src/tokio/bufread/generic/decoder.rs

@@ -68,21 +68,46 @@ impl<R: AsyncBufRead, D: Decode> Decoder<R, D> {
         loop {
             *this.state = match this.state {
                 State::Decoding => {
-                    let input = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
-                    if input.is_empty() {
-                        // Avoid attempting to reinitialise the decoder if the reader
-                        // has returned EOF.
-                        *this.multiple_members = false;
-                        State::Flushing
-                    } else {
-                        let mut input = PartialBuffer::new(input);
-                        let done = this.decoder.decode(&mut input, output)?;
-                        let len = input.written().len();
-                        this.reader.as_mut().consume(len);
-                        if done {
-                            State::Flushing
-                        } else {
-                            State::Decoding
+                    let fill_buf_result = this.reader.as_mut().poll_fill_buf(cx);
+
+                    match fill_buf_result {
+                        Poll::Pending => {
+                            // Try to decode even if there is no new data.
+                            // Some data may be left in the internal state of the decoder
+                            // because there was not enough space in the output buffer.
+                            let written_before = output.written().len();
+
+                            let mut input: Vec<u8> = vec![];
+                            let mut input = PartialBuffer::new(input);
+                            let done = this.decoder.decode(&mut input, output)?;
+                            if output.written().len() == written_before {
+                                return Poll::Pending;
+                            }
+
+                            if done {
+                                State::Flushing
+                            } else {
+                                State::Decoding
+                            }
+                        }
+                        Poll::Ready(input) => {
+                            let input = input?;
+                            if input.is_empty() {
+                                // Avoid attempting to reinitialise the decoder if the reader
+                                // has returned EOF.
+                                *this.multiple_members = false;
+                                State::Flushing
+                            } else {
+                                let mut input = PartialBuffer::new(input);
+                                let done = this.decoder.decode(&mut input, output)?;
+                                let len = input.written().len();
+                                this.reader.as_mut().consume(len);
+                                if done {
+                                    State::Flushing
+                                } else {
+                                    State::Decoding
+                                }
+                            }
                         }
                     }
                 }