Use updated Service trait for ntex (#456)

ntex-io/src/dispatcher.rs

@@ -131,6 +131,7 @@ bitflags::bitflags! {
         const KA_ENABLED    = 0b00100;
         const KA_TIMEOUT    = 0b01000;
         const READ_TIMEOUT  = 0b10000;
+        const READY        = 0b100000;
     }
 }
 
@@ -471,13 +472,19 @@ where
 
     fn poll_service(&mut self, cx: &mut Context<'_>) -> Poll<PollService<U>> {
         // check service readiness
-        if self.shared.service.poll_not_ready(cx).is_pending() {
+        if self.flags.contains(Flags::READY) {
-            return Poll::Ready(self.check_error());
+            if self.shared.service.poll_not_ready(cx).is_pending() {
+                return Poll::Ready(self.check_error());
+            }
+            self.flags.remove(Flags::READY);
         }
 
         // wait until service becomes ready
         match self.shared.service.poll_ready(cx) {
-            Poll::Ready(Ok(_)) => Poll::Ready(self.check_error()),
+            Poll::Ready(Ok(_)) => {
+                self.flags.insert(Flags::READY);
+                Poll::Ready(self.check_error())
+            }
             // pause io read task
             Poll::Pending => {
                 log::trace!(

ntex-net/Cargo.toml

@@ -31,15 +31,15 @@ glommio = ["ntex-rt/glommio", "ntex-glommio"]
 async-std = ["ntex-rt/async-std", "ntex-async-std"]
 
 [dependencies]
-ntex-service = "3"
+ntex-service = "3.3"
 ntex-bytes = "0.1"
 ntex-http = "0.1"
-ntex-io = "2.5"
+ntex-io = "2.8"
 ntex-rt = "0.4.18"
-ntex-util = "2"
+ntex-util = "2.5"
 
-ntex-tokio = { version = "0.5.2", optional = true }
+ntex-tokio = { version = "0.5.3", optional = true }
-ntex-compio = { version = "0.2.0", optional = true }
+ntex-compio = { version = "0.2.1", optional = true }
 ntex-glommio = { version = "0.5.2", optional = true }
 ntex-async-std = { version = "0.5.1", optional = true }
 

ntex-service/src/lib.rs

@@ -120,6 +120,10 @@ pub trait Service<Req> {
 
     #[inline]
     /// Returns when the service is not able to process requests.
+    ///
+    /// Unlike the "ready()" method, the "not_ready()" method returns
+    /// only when the service becomes unready. This method is intended
+    /// for actively monitoring and maintaining the service state.
     async fn not_ready(&self) {
         std::future::pending().await
     }

ntex-tokio/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.5.3] - 2024-11-04
+
+* Use std::task::ready
+
 ## [0.5.2] - 2024-09-11
 
 * Use new io api

ntex-tokio/Cargo.toml

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

ntex/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [2.8.0] - 2024-11-04
+
+* Use updated Service trait
+
+* Don't swallow error when calling panic for read_response_json #443
+
 ## [2.7.0] - 2024-10-16
 
 * Better handling for h2 remote payload

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "2.7.0"
+version = "2.8.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -73,7 +73,7 @@ ntex-h2 = "1.2"
 ntex-rt = "0.4.19"
 ntex-io = "2.8"
 ntex-net = "2.4"
-ntex-tls = "2.1"
+ntex-tls = "2.3"
 
 base64 = "0.22"
 bitflags = "2"

ntex/src/http/client/connector.rs

@@ -5,7 +5,7 @@ use ntex_h2::{self as h2};
 use crate::connect::{Connect as TcpConnect, Connector as TcpConnector};
 use crate::service::{apply_fn, boxed, Service, ServiceCtx};
 use crate::time::{Millis, Seconds};
-use crate::util::{join, timeout::TimeoutError, timeout::TimeoutService};
+use crate::util::{join, select, timeout::TimeoutError, timeout::TimeoutService};
 use crate::{http::Uri, io::IoBoxed};
 
 use super::{connection::Connection, error::ConnectError, pool::ConnectionPool, Connect};
@@ -273,6 +273,7 @@ where
     type Response = <ConnectionPool<T> as Service<Connect>>::Response;
     type Error = ConnectError;
 
+    #[inline]
     async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         if let Some(ref ssl_pool) = self.ssl_pool {
             let (r1, r2) = join(ctx.ready(&self.tcp_pool), ctx.ready(ssl_pool)).await;
@@ -283,6 +284,15 @@ where
         }
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        if let Some(ref ssl_pool) = self.ssl_pool {
+            select(self.tcp_pool.not_ready(), ssl_pool.not_ready()).await;
+        } else {
+            self.tcp_pool.not_ready().await
+        }
+    }
+
     async fn shutdown(&self) {
         self.tcp_pool.shutdown().await;
         if let Some(ref ssl_pool) = self.ssl_pool {

ntex/src/http/client/pool.rs

@@ -117,10 +117,16 @@ where
     type Response = Connection;
     type Error = ConnectError;
 
+    #[inline]
     async fn ready(&self, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         self.connector.ready().await
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        self.connector.not_ready().await
+    }
+
     async fn shutdown(&self) {
         self.connector.shutdown().await
     }

ntex/src/http/h1/service.rs

@@ -6,7 +6,7 @@ use crate::http::error::{DispatchError, ResponseError};
 use crate::http::{request::Request, response::Response};
 use crate::io::{types, Filter, Io, IoRef};
 use crate::service::{IntoServiceFactory, Service, ServiceCtx, ServiceFactory};
-use crate::{channel::oneshot, util::join, util::HashSet};
+use crate::{channel::oneshot, util::join, util::select, util::HashSet};
 
 use super::control::{Control, ControlAck};
 use super::default::DefaultControlService;
@@ -230,6 +230,12 @@ where
         })
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        let cfg = self.config.as_ref();
+        select(cfg.control.not_ready(), cfg.service.not_ready()).await;
+    }
+
     async fn shutdown(&self) {
         self.config.shutdown();
 

ntex/src/http/h2/service.rs

@@ -226,6 +226,11 @@ where
         })
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        self.config.service.not_ready().await;
+    }
+
     #[inline]
     async fn shutdown(&self) {
         self.config.shutdown();

ntex/src/http/service.rs

@@ -2,7 +2,7 @@ use std::{cell::Cell, cell::RefCell, error, fmt, marker, rc::Rc};
 
 use crate::io::{types, Filter, Io, IoRef};
 use crate::service::{IntoServiceFactory, Service, ServiceCtx, ServiceFactory};
-use crate::{channel::oneshot, util::join, util::HashSet};
+use crate::{channel::oneshot, util::join, util::select, util::HashSet};
 
 use super::body::MessageBody;
 use super::builder::HttpServiceBuilder;
@@ -311,6 +311,12 @@ where
         })
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        let cfg = self.config.as_ref();
+        select(cfg.control.not_ready(), cfg.service.not_ready()).await;
+    }
+
     #[inline]
     async fn shutdown(&self) {
         self.config.shutdown();

ntex/src/web/app_service.rs

@@ -5,7 +5,7 @@ use crate::router::{Path, ResourceDef, Router};
 use crate::service::boxed::{self, BoxService, BoxServiceFactory};
 use crate::service::dev::ServiceChainFactory;
 use crate::service::{fn_service, Middleware, Service, ServiceCtx, ServiceFactory};
-use crate::util::{join, BoxFuture, Extensions};
+use crate::util::{join, select, BoxFuture, Extensions};
 
 use super::config::AppConfig;
 use super::error::ErrorRenderer;
@@ -301,6 +301,11 @@ where
         ready2
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        select(self.filter.not_ready(), self.routing.not_ready()).await;
+    }
+
     async fn call(
         &self,
         req: WebRequest<Err>,

ntex/src/web/scope.rs

@@ -5,7 +5,7 @@ use crate::router::{IntoPattern, ResourceDef, Router};
 use crate::service::boxed::{self, BoxService, BoxServiceFactory};
 use crate::service::{chain_factory, dev::ServiceChainFactory, IntoServiceFactory};
 use crate::service::{Identity, Middleware, Service, ServiceCtx, ServiceFactory};
-use crate::util::{join, Extensions};
+use crate::util::{join, select, Extensions};
 
 use super::app::Filter;
 use super::config::ServiceConfig;
@@ -494,6 +494,11 @@ where
         ready2
     }
 
+    #[inline]
+    async fn not_ready(&self) {
+        select(self.filter.not_ready(), self.routing.not_ready()).await;
+    }
+
     async fn call(
         &self,
         req: WebRequest<Err>,

ntex/src/web/test.rs

@@ -244,8 +244,9 @@ where
 {
     let body = read_response::<S>(app, req).await;
 
-    serde_json::from_slice(&body)
+    serde_json::from_slice(&body).unwrap_or_else(|e| {
-        .unwrap_or_else(|_| panic!("read_response_json failed during deserialization"))
+        panic!("read_response_json failed during deserialization, {:?}", e)
+    })
 }
 
 /// Helper method for extractors testing