Stop timer before handling UPGRADE h1 requests (#270)

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.7.16] - 2023-12-15
+
+* Stop timer before handling UPGRADE h1 requests
+
 ## [0.7.15] - 2023-12-14
 
 * Better io tags handling

ntex/Cargo.toml

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

ntex/src/http/h1/dispatcher.rs

@@ -383,8 +383,9 @@ where
                 }
                 // stop io tasks and call upgrade service
                 State::Upgrade(ref mut req) => {
-                    let io = this.inner.io.take();
                     let req = req.take().unwrap();
+                    let io = this.inner.io.take();
+                    io.stop_timer();
 
                     log::trace!(
                         "{}: Switching to upgrade service for {:?}",
@@ -485,6 +486,7 @@ where
     fn service_upgrade(&mut self, mut req: Request) -> CallState<S, X> {
         // Move io into request
         let io: IoBoxed = self.io.take().into();
+        self.io.stop_timer();
         req.head_mut().io = CurrentIo::Io(Rc::new((
             io.get_ref(),
             RefCell::new(Some(Box::new((io, self.codec.clone())))),
@@ -551,7 +553,7 @@ where
 
                     if upgrade {
                         // Handle UPGRADE request
-                        log::trace!("{}: Prepare io for upgrade handler", self.io.tag(),);
+                        log::trace!("{}: Prepare io for upgrade handler", self.io.tag());
                         Poll::Ready(State::Upgrade(Some(req)))
                     } else {
                         if req.upgrade() {

ntex/tests/http_ws.rs

@@ -5,6 +5,7 @@ use ntex::http::test::server as test_server;
 use ntex::http::{body, h1, test, HttpService, Request, Response, StatusCode};
 use ntex::io::{DispatchItem, Dispatcher, Io};
 use ntex::service::{fn_factory, Service, ServiceCtx};
+use ntex::time::{sleep, Millis, Seconds};
 use ntex::util::{BoxFuture, ByteString, Bytes, Ready};
 use ntex::ws::{self, handshake, handshake_response};
 
@@ -51,11 +52,15 @@ impl Service<(Request, Io, h1::Codec)> for WsService {
             io.encode((res, body::BodySize::None).into(), &codec)
                 .unwrap();
 
+            let cfg = ntex_io::DispatcherConfig::default();
+            cfg.set_keepalive_timeout(Seconds(0));
+
             Dispatcher::with_config(
                 io.seal(),
                 ws::Codec::new(),
                 service,
-                &Default::default(),
+                //&Default::default(),
+                &cfg,
             )
             .await
             .map_err(|_| panic!())
@@ -90,6 +95,9 @@ async fn test_simple() {
         move || {
             let ws_service = ws_service.clone();
             HttpService::build()
+                .keep_alive(1)
+                .headers_read_rate(Seconds(1), Seconds::ZERO, 16)
+                .payload_read_rate(Seconds(1), Seconds::ZERO, 16)
                 .upgrade(fn_factory(move || {
                     Ready::Ok::<_, io::Error>(ws_service.clone())
                 }))