Use async fn for Service::ready() and Service::shutdown()

README.md

@@ -29,7 +29,7 @@ Starting ntex v0.5 async runtime must be selected as a feature. Available option
 
 ```toml
 [dependencies]
-ntex = { version = "1.0", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }
 ```
 
 ## Documentation & community resources

ntex-bytes/Cargo.toml

@@ -29,5 +29,5 @@ simdutf8 = { version = "0.1.4", optional = true }
 [dev-dependencies]
 serde_test = "1"
 serde_json = "1"
-ntex = { version = "1", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }
 ntex-bytes = { path = ".", features = ["mpool"] }

ntex-io/Cargo.toml

@@ -29,4 +29,4 @@ pin-project-lite = "0.2"
 rand = "0.8"
 env_logger = "0.11"
 
-ntex = { version = "1", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }

ntex-macros/Cargo.toml

@@ -16,6 +16,6 @@ syn = { version = "^1", features = ["full", "parsing"] }
 proc-macro2 = "^1"
 
 [dev-dependencies]
-ntex = { version = "1", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }
 futures = "0.3"
 env_logger = "0.11"

ntex-net/Cargo.toml

@@ -44,4 +44,4 @@ thiserror = "1.0"
 
 [dev-dependencies]
 env_logger = "0.11"
-ntex = { version = "1", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }

ntex-server/Cargo.toml

@@ -30,7 +30,7 @@ socket2 = "0.5"
 oneshot = { version = "0.1", default-features = false, features = ["async"] }
 
 [dev-dependencies]
-ntex = { version = "1", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }
 ntex-macros = "0.1.3"
 
 [target.'cfg(target_family = "unix")'.dependencies]

ntex-service/Cargo.toml

@@ -19,5 +19,5 @@ path = "src/lib.rs"
 slab = "0.4"
 
 [dev-dependencies]
-ntex = { version = "1", features = ["tokio"] }
-ntex-util = "1"
+ntex = { version = "2", features = ["tokio"] }
+ntex-util = "2"

ntex-tls/Cargo.toml

@@ -40,7 +40,7 @@ tls_openssl = { version = "0.10", package = "openssl", optional = true }
 tls_rust = { version = "0.23", package = "rustls", optional = true }
 
 [dev-dependencies]
-ntex = { version = "1", features = ["openssl", "rustls", "tokio"] }
+ntex = { version = "2", features = ["openssl", "rustls", "tokio"] }
 env_logger = "0.11"
 rustls-pemfile = "2"
 webpki-roots = "0.26"

ntex-tls/examples/rustls-client.rs

@@ -16,7 +16,7 @@ async fn main() -> io::Result<()> {
         .with_no_client_auth();
 
     // rustls connector
-    let connector = connect::rustls::Connector::new(config.clone());
+    let connector = connect::rustls::TlsConnector::new(config.clone());
 
     //let io = connector.connect("www.rust-lang.org:443").await.unwrap();
     let io = connector.connect("127.0.0.1:8443").await.unwrap();

ntex-tls/examples/simple-client.rs

@@ -21,7 +21,7 @@ async fn main() -> io::Result<()> {
     builder.set_verify(SslVerifyMode::NONE);
 
     // openssl connector
-    let connector = connect::openssl::Connector::new(builder.build());
+    let connector = connect::openssl::SslConnector::new(builder.build());
 
     let io = connector.connect("127.0.0.1:8443").await.unwrap();
     println!("Connected to ssl server");

ntex-tls/src/counter.rs

@@ -1,3 +1,4 @@
+#![allow(dead_code)]
 use std::{cell::Cell, future::poll_fn, rc::Rc, task, task::Poll};
 
 use ntex_util::task::LocalWaker;

ntex-util/Cargo.toml

@@ -28,6 +28,6 @@ futures-sink = { version = "0.3", default-features = false, features = ["alloc"]
 pin-project-lite = "0.2"
 
 [dev-dependencies]
-ntex = { version = "1", features = ["tokio"] }
+ntex = { version = "2", features = ["tokio"] }
 ntex-macros = "0.1.3"
 futures-util = { version = "0.3", default-features = false, features = ["alloc"] }

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [2.0.0] - 2024-05-28
+
+* Use "async fn" for Service::ready() and Service::shutdown()
+
 ## [1.2.1] - 2024-03-28
 
 * Feature gate websocket support #320

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "1.2.1"
+version = "2.0.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -60,16 +60,16 @@ brotli = ["dep:brotli2"]
 ntex-codec = "0.6.2"
 ntex-http = "0.1.12"
 ntex-router = "0.5.3"
-ntex-service = "2.0.1"
+ntex-service = "3.0"
 ntex-macros = "0.1.3"
-ntex-util = "1.0.1"
+ntex-util = "2.0"
 ntex-bytes = "0.1.25"
-ntex-server = "1.0.5"
-ntex-h2 = "0.5.4"
+ntex-server = "2.0"
+ntex-h2 = "1.0"
 ntex-rt = "0.4.12"
-ntex-io = "1.2.0"
-ntex-net = "1.0.1"
-ntex-tls = "1.1.0"
+ntex-io = "2.0"
+ntex-net = "2.0"
+ntex-tls = "2.0"
 
 base64 = "0.22"
 bitflags = "2"

ntex/src/http/client/connector.rs

@@ -1,11 +1,11 @@
-use std::{fmt, task::Context, task::Poll, time::Duration};
+use std::{fmt, time::Duration};
 
 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::{timeout::TimeoutError, timeout::TimeoutService};
+use crate::util::{join, timeout::TimeoutError, timeout::TimeoutService};
 use crate::{http::Uri, io::IoBoxed};
 
 use super::{connection::Connection, error::ConnectError, pool::ConnectionPool, Connect};
@@ -273,31 +273,20 @@
     type Response = <ConnectionPool<T> as Service<Connect>>::Response;
     type Error = ConnectError;
 
-    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        let ready = self.tcp_pool.poll_ready(cx)?.is_ready();
-        let ready = if let Some(ref ssl_pool) = self.ssl_pool {
-            ssl_pool.poll_ready(cx)?.is_ready() && ready
+    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;
+            r1?;
+            r2
         } else {
-            ready
-        };
-        if ready {
-            Poll::Ready(Ok(()))
-        } else {
-            Poll::Pending
+            ctx.ready(&self.tcp_pool).await
         }
     }
 
-    fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
-        let tcp_ready = self.tcp_pool.poll_shutdown(cx).is_ready();
-        let ssl_ready = self
-            .ssl_pool
-            .as_ref()
-            .map(|pool| pool.poll_shutdown(cx).is_ready())
-            .unwrap_or(true);
-        if tcp_ready && ssl_ready {
-            Poll::Ready(())
-        } else {
-            Poll::Pending
+    async fn shutdown(&self) {
+        self.tcp_pool.shutdown().await;
+        if let Some(ref ssl_pool) = self.ssl_pool {
+            ssl_pool.shutdown().await;
         }
     }
 
@@ -322,11 +311,11 @@
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::util::lazy;
+    use crate::{service::Pipeline, util::lazy};
 
     #[crate::rt_test]
     async fn test_readiness() {
-        let conn = Connector::default().finish();
+        let conn = Pipeline::new(Connector::default().finish()).bind();
         assert!(lazy(|cx| conn.poll_ready(cx).is_ready()).await);
         assert!(lazy(|cx| conn.poll_shutdown(cx).is_ready()).await);
     }

ntex/src/http/client/h2proto.rs

@@ -80,7 +80,7 @@ where
     if !eof {
         // sending body is async process, we can handle upload and download
         // at the same time
-        crate::rt::spawn(async move {
+        let _ = crate::rt::spawn(async move {
             if let Err(e) = send_body(body, &snd_stream).await {
                 log::error!("Cannot send body: {:?}", e);
                 snd_stream.reset(frame::Reason::INTERNAL_ERROR);
@@ -125,7 +125,7 @@ async fn get_response(
                         log::debug!("Creating local payload stream for {:?}", stream.id());
                         let (mut pl, payload) =
                             payload::Payload::create(stream.empty_capacity());
-                        crate::rt::spawn(async move {
+                        let _ = crate::rt::spawn(async move {
                             loop {
                                 let h2::Message { stream, kind } =
                                     match rcv_stream.recv().await {

ntex/src/http/client/pool.rs

@@ -80,7 +80,7 @@ where
         }));
 
         // start pool support future
-        crate::rt::spawn(ConnectionPoolSupport {
+        let _ = crate::rt::spawn(ConnectionPoolSupport {
             connector: connector.clone(),
             inner: inner.clone(),
             waiters: waiters.clone(),
@@ -117,8 +117,13 @@ where
     type Response = Connection;
     type Error = ConnectError;
 
-    crate::forward_poll_ready!(connector);
-    crate::forward_poll_shutdown!(connector);
+    async fn ready(&self, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
+        self.connector.ready().await
+    }
+
+    async fn shutdown(&self) {
+        self.connector.shutdown().await
+    }
 
     async fn call(
         &self,
@@ -252,7 +257,7 @@ impl Inner {
                     || (now - conn.created) > self.conn_lifetime
                 {
                     if let ConnectionType::H1(io) = conn.io {
-                        spawn(async move {
+                        let _ = spawn(async move {
                             let _ = io.shutdown().await;
                         });
                     }
@@ -419,7 +424,7 @@ where
         let disconnect_timeout = inner.borrow().disconnect_timeout;
 
         #[allow(clippy::redundant_async_block)]
-        spawn(async move {
+        let _ = spawn(async move {
             OpenConnection::<T> {
                 tx: Some(tx),
                 key: key.clone(),
@@ -576,7 +581,7 @@ impl Acquired {
                 );
                 match io {
                     ConnectionType::H1(io) => {
-                        spawn(async move {
+                        let _ = spawn(async move {
                             let _ = io.shutdown().await;
                         });
                     }
@@ -634,7 +639,8 @@ mod tests {
                 h2::Config::client(),
             )
             .clone(),
-        );
+        )
+        .bind();
 
         // uri must contain authority
         let req = Connect {

ntex/src/http/config.rs

@@ -325,7 +325,7 @@ impl DateService {
 
             // periodic date update
             let s = self.clone();
-            crate::rt::spawn(async move {
+            let _ = crate::rt::spawn(async move {
                 sleep(Millis(500)).await;
                 s.0.current.set(false);
             });

ntex/src/http/h1/control.rs

@@ -210,7 +210,7 @@ impl<F: Filter> Upgrade<F> {
         H: FnOnce(Request, Io<F>, Codec) -> R + 'static,
         R: Future<Output = O>,
     {
-        crate::rt::spawn(async move {
+        let _ = crate::rt::spawn(async move {
             let _ = f(self.req, self.io, self.codec).await;
         });
         ControlAck {

ntex/src/http/h1/service.rs

@@ -1,11 +1,12 @@
-use std::{error::Error, fmt, marker, rc::Rc, task::Context, task::Poll};
+use std::{error::Error, fmt, marker, rc::Rc};
 
 use crate::http::body::MessageBody;
 use crate::http::config::{DispatcherConfig, ServiceConfig};
 use crate::http::error::{DispatchError, ResponseError};
 use crate::http::{request::Request, response::Response};
 use crate::io::{types, Filter, Io};
 use crate::service::{IntoServiceFactory, Service, ServiceCtx, ServiceFactory};
+use crate::util::join;
 
 use super::control::{Control, ControlAck};
 use super::default::DefaultControlService;
@@ -208,43 +209,23 @@
     type Response = ();
     type Error = DispatchError;
 
-    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    async fn ready(&self, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
         let cfg = self.config.as_ref();
 
-        let ready1 = cfg
-            .control
-            .poll_ready(cx)
-            .map_err(|e| {
-                log::error!("Http control service readiness error: {:?}", e);
-                DispatchError::Control(Box::new(e))
-            })?
-            .is_ready();
-
-        let ready2 = cfg
-            .service
-            .poll_ready(cx)
-            .map_err(|e| {
-                log::error!("Http service readiness error: {:?}", e);
-                DispatchError::Service(Box::new(e))
-            })?
-            .is_ready();
-
-        if ready1 && ready2 {
-            Poll::Ready(Ok(()))
-        } else {
-            Poll::Pending
-        }
+        let (ready1, ready2) = join(cfg.control.ready(), cfg.service.ready()).await;
+        ready1.map_err(|e| {
+            log::error!("Http control service readiness error: {:?}", e);
+            DispatchError::Control(Box::new(e))
+        })?;
+        ready2.map_err(|e| {
+            log::error!("Http service readiness error: {:?}", e);
+            DispatchError::Service(Box::new(e))
+        })
     }
 
-    fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
-        let ready1 = self.config.control.poll_shutdown(cx).is_ready();
-        let ready2 = self.config.service.poll_shutdown(cx).is_ready();
-
-        if ready1 && ready2 {
-            Poll::Ready(())
-        } else {
-            Poll::Pending
-        }
+    async fn shutdown(&self) {
+        self.config.control.shutdown().await;
+        self.config.service.shutdown().await;
     }
 
     async fn call(&self, io: Io<F>, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {