Merge branch 'main' into main

Author: surban

CHANGELOG.md

@@ -6,6 +6,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0),
 
 ## Unreleased
 
+## [0.4.19](https://github.com/Nullus157/async-compression/compare/v0.4.18...v0.4.19) - 2025-02-27
+
+### Changed
+
+- Update `bzip2` dependency to `0.5`.
+
+### Fixed
+
+- Ensure that flush finishes before continuing.
+
 ## [0.4.18](https://github.com/Nullus157/async-compression/compare/v0.4.17...v0.4.18) - 2024-11-23
 
 ### Fixed

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "async-compression"
-version = "0.4.18"
+version = "0.4.19"
 authors = ["Wim Looman <[email protected]>", "Allen Bui <[email protected]>"]
 edition = "2018"
 resolver = "2"

src/tokio/write/generic/encoder.rs

@@ -16,6 +16,7 @@ use tokio::io::{AsyncBufRead, AsyncRead, AsyncWrite, ReadBuf};
 #[derive(Debug)]
 enum State {
     Encoding,
+    Flushing,
     Finishing,
     Done,
 }
@@ -80,6 +81,12 @@ impl<W: AsyncWrite, E: Encode> Encoder<W, E> {
                     State::Encoding
                 }
 
+                // Once a flush has been started, it must be completed.
+                State::Flushing => match this.encoder.flush(&mut output)? {
+                    true => State::Encoding,
+                    false => State::Flushing,
+                },
+
                 State::Finishing | State::Done => {
                     return Poll::Ready(Err(io::Error::new(
                         io::ErrorKind::Other,
@@ -105,7 +112,7 @@ impl<W: AsyncWrite, E: Encode> Encoder<W, E> {
             let mut output = PartialBuffer::new(output);
 
             let done = match this.state {
-                State::Encoding => this.encoder.flush(&mut output)?,
+                State::Encoding | State::Flushing => this.encoder.flush(&mut output)?,
 
                 State::Finishing | State::Done => {
                     return Poll::Ready(Err(io::Error::new(
@@ -114,11 +121,13 @@ impl<W: AsyncWrite, E: Encode> Encoder<W, E> {
                     )))
                 }
             };
+            *this.state = State::Flushing;
 
             let produced = output.written().len();
             this.writer.as_mut().produce(produced);
 
             if done {
+                *this.state = State::Encoding;
                 return Poll::Ready(Ok(()));
             }
         }
@@ -140,6 +149,12 @@ impl<W: AsyncWrite, E: Encode> Encoder<W, E> {
                     }
                 }
 
+                // Once a flush has been started, it must be completed.
+                State::Flushing => match this.encoder.flush(&mut output)? {
+                    true => State::Finishing,
+                    false => State::Flushing,
+                },
+
                 State::Done => State::Done,
             };