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

src/tokio/bufread/generic/decoder.rs

@@ -65,18 +65,37 @@ 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 input = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
-                    if input.is_empty() {
+                    let input = if first {
+                        &[][..]
+                    } else {
+                        ready!(this.reader.as_mut().poll_fill_buf(cx))?
+                    };
+
+                    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)?;
+                        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 {
+                                Err(err)
+                            }
+                        })?;
+
+                        first = false;
+
                         let len = input.written().len();
                         this.reader.as_mut().consume(len);
                         if done {