Fix borrow mut panic #204 (#205)

* Fix borrow mut panic #204

ntex-connect/Cargo.toml

@@ -45,7 +45,7 @@ log = "0.4"
 thiserror = "1.0"
 
 ntex-tokio = { version = "0.2.3", optional = true }
-ntex-glommio = { version = "0.2.3", optional = true }
+ntex-glommio = { version = "0.2.4", optional = true }
 ntex-async-std = { version = "0.2.2", optional = true }
 
 # openssl

ntex-glommio/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.2.4] - 2023-05-30
+
+* Fix borrow mut panic #204
+
 ## [0.2.3] - 2023-04-11
 
 * Chore upgrade glommio to 0.8

ntex-glommio/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-glommio"
-version = "0.2.3"
+version = "0.2.4"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "glommio intergration for ntex framework"
 keywords = ["network", "framework", "async", "futures"]

ntex-glommio/src/io.rs

@@ -106,7 +106,8 @@ enum IoWriteState {
 }
 
 enum Shutdown {
-    None(Pin<Box<dyn Future<Output = glommio::Result<(), ()>>>>),
+    Flush,
+    Close(Pin<Box<dyn Future<Output = glommio::Result<(), ()>>>>),
     Stopping(u16),
 }
 
@@ -177,11 +178,7 @@ impl Future for WriteTask {
                             sleep(time)
                         };
 
-                        let io = this.io.clone();
+                        this.st = IoWriteState::Shutdown(timeout, Shutdown::Flush);
-                        let fut = Box::pin(async move {
-                            io.0.borrow().shutdown(std::net::Shutdown::Write).await
-                        });
-                        this.st = IoWriteState::Shutdown(timeout, Shutdown::None(fut));
                         self.poll(cx)
                     }
                     Poll::Ready(WriteStatus::Terminate) => {
@@ -199,16 +196,18 @@ impl Future for WriteTask {
                 // use disconnect timeout, otherwise it could hang forever.
                 loop {
                     match st {
-                        Shutdown::None(ref mut fut) => {
+                        Shutdown::Flush => {
                             // flush write buffer
                             let mut io = this.io.0.borrow_mut();
                             match this.state.with_buf(|buf| flush_io(&mut *io, buf, cx)) {
                                 Poll::Ready(Ok(())) => {
-                                    if ready!(fut.poll(cx)).is_err() {
+                                    let io = this.io.clone();
-                                        this.state.close(None);
+                                    let fut = Box::pin(async move {
-                                        return Poll::Ready(());
+                                        io.0.borrow()
-                                    }
+                                            .shutdown(std::net::Shutdown::Write)
-                                    *st = Shutdown::Stopping(0);
+                                            .await
+                                    });
+                                    *st = Shutdown::Close(fut);
                                     continue;
                                 }
                                 Poll::Ready(Err(err)) => {
@@ -222,6 +221,14 @@ impl Future for WriteTask {
                                 Poll::Pending => (),
                             }
                         }
+                        Shutdown::Close(ref mut fut) => {
+                            if ready!(fut.poll(cx)).is_err() {
+                                this.state.close(None);
+                                return Poll::Ready(());
+                            }
+                            *st = Shutdown::Stopping(0);
+                            continue;
+                        }
                         Shutdown::Stopping(ref mut count) => {
                             // read until 0 or err
                             let mut buf = [0u8; 512];
@@ -479,11 +486,7 @@ impl Future for UnixWriteTask {
                             sleep(time)
                         };
 
-                        let io = this.io.clone();
+                        this.st = IoWriteState::Shutdown(timeout, Shutdown::Flush);
-                        let fut = Box::pin(async move {
-                            io.0.borrow().shutdown(std::net::Shutdown::Write).await
-                        });
-                        this.st = IoWriteState::Shutdown(timeout, Shutdown::None(fut));
                         self.poll(cx)
                     }
                     Poll::Ready(WriteStatus::Terminate) => {
@@ -501,16 +504,18 @@ impl Future for UnixWriteTask {
                 // use disconnect timeout, otherwise it could hang forever.
                 loop {
                     match st {
-                        Shutdown::None(ref mut fut) => {
+                        Shutdown::Flush => {
                             // flush write buffer
                             let mut io = this.io.0.borrow_mut();
                             match this.state.with_buf(|buf| flush_io(&mut *io, buf, cx)) {
                                 Poll::Ready(Ok(())) => {
-                                    if ready!(fut.poll(cx)).is_err() {
+                                    let io = this.io.clone();
-                                        this.state.close(None);
+                                    let fut = Box::pin(async move {
-                                        return Poll::Ready(());
+                                        io.0.borrow()
-                                    }
+                                            .shutdown(std::net::Shutdown::Write)
-                                    *st = Shutdown::Stopping(0);
+                                            .await
+                                    });
+                                    *st = Shutdown::Close(fut);
                                     continue;
                                 }
                                 Poll::Ready(Err(err)) => {
@@ -524,6 +529,14 @@ impl Future for UnixWriteTask {
                                 Poll::Pending => (),
                             }
                         }
+                        Shutdown::Close(ref mut fut) => {
+                            if ready!(fut.poll(cx)).is_err() {
+                                this.state.close(None);
+                                return Poll::Ready(());
+                            }
+                            *st = Shutdown::Stopping(0);
+                            continue;
+                        }
                         Shutdown::Stopping(ref mut count) => {
                             // read until 0 or err
                             let mut buf = [0u8; 512];

ntex-util/src/future/mod.rs

@@ -1,7 +1,7 @@
 //! Utilities for futures
 use std::{future::Future, mem, pin::Pin, task::Context, task::Poll};
 
-pub use futures_core::Stream;
+pub use futures_core::{Stream, TryFuture};
 pub use futures_sink::Sink;
 
 mod either;

ntex/Cargo.toml

@@ -61,7 +61,7 @@ ntex-rt = "0.4.9"
 ntex-io = "0.2.10"
 ntex-tls = "0.2.4"
 ntex-tokio = { version = "0.2.3", optional = true }
-ntex-glommio = { version = "0.2.3", optional = true }
+ntex-glommio = { version = "0.2.4", optional = true }
 ntex-async-std = { version = "0.2.1", optional = true }
 
 async-oneshot = "0.5.0"