port fix from futures to tokio

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 => Ok(true), // Waiting for more input.
+            Status::BufError => Err(io::Error::new(io::ErrorKind::Other, "unexpected BufError")),
         }
     }
 

src/tokio/bufread/generic/decoder.rs

@@ -65,49 +65,43 @@ impl<R: AsyncBufRead, D: Decode> Decoder<R, D> {
     ) -> Poll<Result<()>> {
         let mut this = self.project();
 
+        let mut first = true;
+
         loop {
             *this.state = match this.state {
                 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;
-                            }
+                    let input = if first {
+                        &[][..]
+                    } else {
+                        ready!(this.reader.as_mut().poll_fill_buf(cx))?
+                    };
 
-                            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
+                    if input.is_empty() && !first {
+                        // 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).or_else(|err| {
+                            // ignore the first error, occurs when input is empty
+                            // but we need to run decode to flush
+                            if first {
+                                Ok(false)
                             } 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
-                                }
+                                Err(err)
                             }
+                        })?;
+
+                        first = false;
+
+                        let len = input.written().len();
+                        this.reader.as_mut().consume(len);
+                        if done {
+                            State::Flushing
+                        } else {
+                            State::Decoding
                         }
                     }
                 }