Fix KeepAlive timeout handling in default dispatcher

ntex-io/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [1.1.0] - 2024-05-01
+
+* Add IoRef::notify_timeout() helper method
+
+* Fix KeepAlive timeout handling in default dispatcher
+
 ## [1.0.2] - 2024-03-31
 
 * Add IoRef::is_wr_backpressure() method

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "1.0.2"
+version = "1.1.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]

ntex-io/src/dispatcher.rs

@@ -575,15 +575,19 @@ where
                         self.shared.io.tag()
                     );
                 }
-                return Err(DispatchItem::ReadTimeout);
+                Err(DispatchItem::ReadTimeout)
+            } else {
+                Ok(())
             }
+        } else if self.flags.contains(Flags::KA_TIMEOUT) {
+            log::trace!(
+                "{}: Keep-alive error, stopping dispatcher",
+                self.shared.io.tag()
+            );
+            Err(DispatchItem::KeepAliveTimeout)
+        } else {
+            Ok(())
         }
-
-        log::trace!(
-            "{}: Keep-alive error, stopping dispatcher",
-            self.shared.io.tag()
-        );
-        Err(DispatchItem::KeepAliveTimeout)
     }
 }
 

ntex-io/src/io.rs

@@ -93,13 +93,12 @@ impl IoState {
     }
 
     pub(super) fn notify_timeout(&self) {
-        log::trace!("{}: Timeout, notify dispatcher", self.tag.get());
-
         let mut flags = self.flags.get();
         if !flags.contains(Flags::DSP_TIMEOUT) {
             flags.insert(Flags::DSP_TIMEOUT);
             self.flags.set(flags);
             self.dispatch_task.wake();
+            log::trace!("{}: Timer, notify dispatcher", self.tag.get());
         }
     }
 

ntex-io/src/ioref.rs

@@ -228,6 +228,12 @@ impl IoRef {
         self.0.timeout.get()
     }
 
+    #[inline]
+    /// wakeup dispatcher and send keep-alive error
+    pub fn notify_timeout(&self) {
+        self.0.notify_timeout()
+    }
+
     #[inline]
     /// Start timer
     pub fn start_timer(&self, timeout: Seconds) -> timer::TimerHandle {

ntex/Cargo.toml

@@ -67,7 +67,7 @@ ntex-bytes = "0.1.25"
 ntex-server = "1.0.5"
 ntex-h2 = "0.5.4"
 ntex-rt = "0.4.12"
-ntex-io = "1.0.1"
+ntex-io = "1.1.0"
 ntex-net = "1.0.1"
 ntex-tls = "1.1.0"
 

ntex/examples/client.rs

@@ -2,24 +2,16 @@ use ntex::http::client::{error::SendRequestError, Client};
 
 #[ntex::main]
 async fn main() -> Result<(), SendRequestError> {
-    std::env::set_var("RUST_LOG", "ntex=trace");
+    std::env::set_var("RUST_LOG", "trace");
     env_logger::init();
 
     let client = Client::new();
 
     // Create request builder, configure request and send
-    let mut response = client
-        .get("https://www.rust-lang.org/")
-        .header("User-Agent", "ntex")
-        .send()
-        .await?;
-
-    // server http response
-    println!("Response: {:?}", response);
-
-    // read response body
-    let body = response.body().await.unwrap();
-    println!("Downloaded: {:?} bytes", body.len());
+    loop {
+        let _response = client.get("https://www.google.com").send().await.unwrap();
+        ntex::time::sleep(std::time::Duration::from_secs(10)).await;
+    }
 
     Ok(())
 }