Fix flush framed write task

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.2.0-b.2] - 2021-01-20
+
+* Fix flush framed write task
+
 ## [0.2.0-b.1] - 2021-01-19
 
 * Introduce ntex::framed module

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "0.2.0-b.1"
+version = "0.2.0-b.2"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"

ntex/src/framed/dispatcher.rs

@@ -433,7 +433,6 @@ mod tests {
     use bytes::Bytes;
     use futures::future::FutureExt;
 
-    use crate::channel::condition::Condition;
     use crate::codec::BytesCodec;
     use crate::rt::time::delay_for;
     use crate::testing::Io;

ntex/src/framed/write.rs

@@ -178,46 +178,50 @@ pub(super) fn flush<T>(
     io: &mut T,
     buf: &mut BytesMut,
     cx: &mut Context<'_>,
-) -> Poll<Result<(), io::Error>>
+) -> Poll<io::Result<()>>
 where
     T: AsyncRead + AsyncWrite + Unpin,
 {
     // log::trace!("flushing framed transport: {}", len);
     let len = buf.len();
-    if len == 0 {
-        return Poll::Ready(Ok(()));
-    }
 
-    let mut written = 0;
+    if len != 0 {
-    while written < len {
+        let mut written = 0;
-        match Pin::new(&mut *io).poll_write(cx, &buf[written..]) {
+
-            Poll::Pending => break,
+        while written < len {
-            Poll::Ready(Ok(n)) => {
+            match Pin::new(&mut *io).poll_write(cx, &buf[written..]) {
-                if n == 0 {
+                Poll::Pending => break,
-                    log::trace!("Disconnected during flush, written {}", written);
+                Poll::Ready(Ok(n)) => {
-                    return Poll::Ready(Err(io::Error::new(
+                    if n == 0 {
-                        io::ErrorKind::WriteZero,
+                        log::trace!("Disconnected during flush, written {}", written);
-                        "failed to write frame to transport",
+                        return Poll::Ready(Err(io::Error::new(
-                    )));
+                            io::ErrorKind::WriteZero,
-                } else {
+                            "failed to write frame to transport",
-                    written += n
+                        )));
+                    } else {
+                        written += n
+                    }
+                }
+                Poll::Ready(Err(e)) => {
+                    log::trace!("Error during flush: {}", e);
+                    return Poll::Ready(Err(e));
                 }
             }
-            Poll::Ready(Err(e)) => {
+        }
-                log::trace!("Error during flush: {}", e);
+        // log::trace!("flushed {} bytes", written);
-                return Poll::Ready(Err(e));
+
-            }
+        // remove written data
+        if written == len {
+            // flushed same amount as in buffer, we dont need to reallocate
+            unsafe { buf.set_len(0) }
+        } else {
+            buf.advance(written);
         }
     }
-    // log::trace!("flushed {} bytes", written);
 
-    // remove written data
+    // flush
-    if written == len {
+    futures::ready!(Pin::new(&mut *io).poll_flush(cx))?;
-        // flushed same amount as in buffer, we dont need to reallocate
+
-        unsafe { buf.set_len(0) }
-    } else {
-        buf.advance(written);
-    }
     if buf.is_empty() {
         Poll::Ready(Ok(()))
     } else {