fix: encoder not flushing data

crates/async-compression/src/futures/bufread/generic/encoder.rs

@@ -23,8 +23,11 @@ impl<R: AsyncBufRead, E: EncodeV2> AsyncRead for Encoder<R, E> {
         }
 
         let mut output = WriteBuffer::new_initialized(buf);
-        self.do_poll_read(cx, &mut output)
-            .map_ok(|()| output.written_len())
+        match self.do_poll_read(cx, &mut output) {
+            Poll::Pending if output.written().is_empty() => Poll::Pending,
+            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
+            _ => Poll::Ready(Ok(output.written_len())),
+        }
     }
 }
 

crates/async-compression/src/generic/bufread/encoder.rs

@@ -58,7 +58,7 @@ impl Encoder {
 
                             read += input.written().len();
 
-                            // Poll for more data
+                            self.state = State::Encoding(read);
                             break;
                         }
                     }

crates/async-compression/src/tokio/bufread/generic/mod.rs

@@ -30,5 +30,9 @@ fn poll_read(
     unsafe { buf.assume_init(initialized) };
     buf.advance(written);
 
-    res
+    match res {
+        Poll::Pending if written == 0 => Poll::Pending,
+        Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
+        _ => Poll::Ready(Ok(())),
+    }
 }