Refactor pipeline call impl (#219)

* Refactor pipeline call impl

.github/workflows/linux.yml

@@ -8,7 +8,7 @@ jobs:
       fail-fast: false
       matrix:
         version:
-          - 1.66.0 # MSRV
+          - 1.67.0 # MSRV
           - stable
           - nightly
 

ntex-async-std/src/io.rs

@@ -124,7 +124,7 @@ impl Future for WriteTask {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.as_mut().get_mut();
+        let this = self.as_mut().get_mut();
 
         match this.st {
             IoWriteState::Processing(ref mut delay) => {
@@ -432,7 +432,7 @@ mod unixstream {
         type Output = ();
 
         fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-            let mut this = self.as_mut().get_mut();
+            let this = self.as_mut().get_mut();
 
             match this.st {
                 IoWriteState::Processing(ref mut delay) => {

ntex-io/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.3.2] - 2023-08-10
+
+* Replace `PipelineCall` with `ServiceCall<'static, S, R>`
+
 ## [0.3.1] - 2023-06-23
 
 * `PipelineCall` is static

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "0.3.1"
+version = "0.3.2"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]
@@ -19,7 +19,7 @@ path = "src/lib.rs"
 ntex-codec = "0.6.2"
 ntex-bytes = "0.1.19"
 ntex-util = "0.3.0"
-ntex-service = "1.2.1"
+ntex-service = "1.2.3"
 
 bitflags = "1.3"
 log = "0.4"

ntex-io/src/dispatcher.rs

@@ -250,9 +250,8 @@ where
                     // call service
                     let shared = slf.shared.clone();
                     shared.inflight.set(shared.inflight.get() + 1);
-                    let fut = shared.service.call(item);
                     spawn(async move {
-                        let result = fut.await;
+                        let result = shared.service.call(item).await;
                         shared.handle_result(result, &shared.io);
                     });
                 }
@@ -276,9 +275,8 @@ where
                     // call service
                     let shared = slf.shared.clone();
                     shared.inflight.set(shared.inflight.get() + 1);
-                    let fut = shared.service.call(item);
                     spawn(async move {
-                        let result = fut.await;
+                        let result = shared.service.call(item).await;
                         shared.handle_result(result, &shared.io);
                     });
                 }

ntex-io/src/testing.rs

@@ -468,7 +468,7 @@ impl Future for WriteTask {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.as_mut().get_mut();
+        let this = self.as_mut().get_mut();
 
         match this.st {
             IoWriteState::Processing(ref mut delay) => {

ntex-service/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.2.3] - 2023-08-10
+
+* Check readiness for pipeline calls
+
 ## [1.2.2] - 2023-06-24
 
 * Added `ServiceCall::advance_to_call`

ntex-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-service"
-version = "1.2.2"
+version = "1.2.3"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex service"
 keywords = ["network", "framework", "async", "futures"]

ntex-service/src/apply.rs

@@ -56,7 +56,7 @@ impl<S> ApplyService<S> {
     where
         S: Service<R>,
     {
-        self.service.service_call(req)
+        self.service.call(req)
     }
 }
 
@@ -85,7 +85,6 @@ where
     type Error = Err;
     type Future<'f> = R where Self: 'f, In: 'f, R: 'f;
 
-    crate::forward_poll_ready!(service);
     crate::forward_poll_shutdown!(service);
 
     #[inline]

ntex-service/src/ctx.rs

@@ -1,6 +1,6 @@
 use std::{cell::UnsafeCell, future::Future, marker, pin::Pin, rc::Rc, task};
 
-use crate::Service;
+use crate::{Pipeline, Service};
 
 pub struct ServiceCtx<'a, S: ?Sized> {
     idx: usize,
@@ -82,7 +82,7 @@ impl Drop for Waiters {
     }
 }
 
-impl<'a, S: ?Sized> ServiceCtx<'a, S> {
+impl<'a, S> ServiceCtx<'a, S> {
     pub(crate) fn new(waiters: &'a Waiters) -> Self {
         Self {
             idx: waiters.index,
@@ -107,7 +107,7 @@ impl<'a, S: ?Sized> ServiceCtx<'a, S> {
     /// Wait for service readiness and then call service
     pub fn call<T, R>(&self, svc: &'a T, req: R) -> ServiceCall<'a, T, R>
     where
-        T: Service<R> + ?Sized,
+        T: Service<R>,
         R: 'a,
     {
         ServiceCall {
@@ -125,7 +125,7 @@ impl<'a, S: ?Sized> ServiceCtx<'a, S> {
     /// Call service, do not check service readiness
     pub fn call_nowait<T, R>(&self, svc: &'a T, req: R) -> T::Future<'a>
     where
-        T: Service<R> + ?Sized,
+        T: Service<R>,
         R: 'a,
     {
         svc.call(
@@ -139,9 +139,9 @@ impl<'a, S: ?Sized> ServiceCtx<'a, S> {
     }
 }
 
-impl<'a, S: ?Sized> Copy for ServiceCtx<'a, S> {}
+impl<'a, S> Copy for ServiceCtx<'a, S> {}
 
-impl<'a, S: ?Sized> Clone for ServiceCtx<'a, S> {
+impl<'a, S> Clone for ServiceCtx<'a, S> {
     #[inline]
     fn clone(&self) -> Self {
         Self {
@@ -157,8 +157,6 @@ pin_project_lite::pin_project! {
     pub struct ServiceCall<'a, S, Req>
     where
         S: Service<Req>,
-        S: 'a,
-        S: ?Sized,
         Req: 'a,
     {
         #[pin]
@@ -166,16 +164,45 @@ pin_project_lite::pin_project! {
     }
 }
 
+pin_project_lite::pin_project! {
+    #[project = ServiceCallStateProject]
+    enum ServiceCallState<'a, S, Req>
+    where
+        S: Service<Req>,
+        Req: 'a,
+    {
+        Ready { req: Option<Req>,
+                svc: &'a S,
+                idx: usize,
+                waiters: &'a WaitersRef,
+        },
+        ReadyPl { req: Option<Req>,
+                  svc: &'a Pipeline<S>,
+                  pl: Pipeline<S>,
+        },
+        Call { #[pin] fut: S::Future<'a> },
+        Empty,
+    }
+}
+
 impl<'a, S, Req> ServiceCall<'a, S, Req>
 where
     S: Service<Req>,
-    S: 'a,
-    S: ?Sized,
     Req: 'a,
 {
+    pub(crate) fn call_pipeline(req: Req, svc: &'a Pipeline<S>) -> Self {
+        ServiceCall {
+            state: ServiceCallState::ReadyPl {
+                req: Some(req),
+                pl: svc.clone(),
+                svc,
+            },
+        }
+    }
+
     pub fn advance_to_call(self) -> ServiceCallToCall<'a, S, Req> {
         match self.state {
-            ServiceCallState::Ready { .. } => {}
+            ServiceCallState::Ready { .. } | ServiceCallState::ReadyPl { .. } => {}
             ServiceCallState::Call { .. } | ServiceCallState::Empty => {
                 panic!(
                     "`ServiceCall::advance_to_call` must be called before `ServiceCall::poll`"
@@ -186,28 +213,9 @@ where
     }
 }
 
-pin_project_lite::pin_project! {
-    #[project = ServiceCallStateProject]
-    enum ServiceCallState<'a, S, Req>
-    where
-        S: Service<Req>,
-        S: 'a,
-        S: ?Sized,
-        Req: 'a,
-    {
-        Ready { req: Option<Req>,
-                svc: &'a S,
-                idx: usize,
-                waiters: &'a WaitersRef,
-        },
-        Call { #[pin] fut: S::Future<'a> },
-        Empty,
-    }
-}
-
 impl<'a, S, Req> Future for ServiceCall<'a, S, Req>
 where
-    S: Service<Req> + ?Sized,
+    S: Service<Req>,
 {
     type Output = Result<S::Response, S::Error>;
 
@@ -243,7 +251,21 @@ where
                     task::Poll::Pending
                 }
             },
-            ServiceCallStateProject::Call { fut } => fut.poll(cx).map(|r| {
+            ServiceCallStateProject::ReadyPl { req, svc, pl } => {
+                task::ready!(pl.poll_ready(cx))?;
+
+                let ctx = ServiceCtx::new(&svc.waiters);
+                let svc_call = svc.get_ref().call(req.take().unwrap(), ctx);
+
+                // SAFETY: `svc_call` has same lifetime same as lifetime of `pl.svc`
+                // Pipeline::svc is heap allocated(Rc<S>), we keep it alive until
+                // `svc_call` get resolved to result
+                let fut = unsafe { std::mem::transmute(svc_call) };
+
+                this.state.set(ServiceCallState::Call { fut });
+                self.poll(cx)
+            }
+            ServiceCallStateProject::Call { fut, .. } => fut.poll(cx).map(|r| {
                 this.state.set(ServiceCallState::Empty);
                 r
             }),
@@ -259,8 +281,6 @@ pin_project_lite::pin_project! {
     pub struct ServiceCallToCall<'a, S, Req>
     where
         S: Service<Req>,
-        S: 'a,
-        S: ?Sized,
         Req: 'a,
     {
         #[pin]
@@ -270,7 +290,7 @@ pin_project_lite::pin_project! {
 
 impl<'a, S, Req> Future for ServiceCallToCall<'a, S, Req>
 where
-    S: Service<Req> + ?Sized,
+    S: Service<Req>,
 {
     type Output = Result<S::Future<'a>, S::Error>;
 
@@ -306,6 +326,12 @@ where
                     task::Poll::Pending
                 }
             },
+            ServiceCallStateProject::ReadyPl { req, svc, pl } => {
+                task::ready!(pl.poll_ready(cx))?;
+
+                let ctx = ServiceCtx::new(&svc.waiters);
+                task::Poll::Ready(Ok(svc.get_ref().call(req.take().unwrap(), ctx)))
+            }
             ServiceCallStateProject::Call { .. } => {
                 unreachable!("`ServiceCallToCall` can only be constructed in `Ready` state")
             }
@@ -387,13 +413,13 @@ mod tests {
         let data1 = data.clone();
         ntex::rt::spawn(async move {
             let _ = poll_fn(|cx| srv1.poll_ready(cx)).await;
-            let i = srv1.call("srv1").await.unwrap();
+            let i = srv1.call_nowait("srv1").await.unwrap();
             data1.borrow_mut().push(i);
         });
 
         let data2 = data.clone();
         ntex::rt::spawn(async move {
-            let i = srv2.service_call("srv2").await.unwrap();
+            let i = srv2.call_static("srv2").await.unwrap();
             data2.borrow_mut().push(i);
         });
         time::sleep(time::Millis(50)).await;
@@ -417,7 +443,7 @@ mod tests {
         let con = condition::Condition::new();
         let srv = Pipeline::from(Srv(cnt.clone(), con.wait()));
 
-        let mut fut = srv.service_call("test").advance_to_call();
+        let mut fut = srv.call("test").advance_to_call();
         let _ = lazy(|cx| Pin::new(&mut fut).poll(cx)).await;
         con.notify();
 
@@ -432,7 +458,7 @@ mod tests {
         let con = condition::Condition::new();
         let srv = Pipeline::from(Srv(cnt.clone(), con.wait()));
 
-        let mut fut = srv.service_call("test");
+        let mut fut = srv.call("test");
         let _ = lazy(|cx| Pin::new(&mut fut).poll(cx)).await;
         con.notify();
 

ntex-service/src/fn_shutdown.rs

@@ -85,7 +85,7 @@ mod tests {
 
         let pipe = Pipeline::new(chain(srv).and_then(on_shutdown).clone());
 
-        let res = pipe.service_call(()).await;
+        let res = pipe.call(()).await;
         assert_eq!(lazy(|cx| pipe.poll_ready(cx)).await, Poll::Ready(Ok(())));
         assert!(res.is_ok());
         assert_eq!(res.unwrap(), "pipe");

ntex-service/src/lib.rs

@@ -261,7 +261,7 @@ pub trait ServiceFactory<Req, Cfg = ()> {
 
 impl<'a, S, Req> Service<Req> for &'a S
 where
-    S: Service<Req> + ?Sized,
+    S: Service<Req>,
 {
     type Response = S::Response;
     type Error = S::Error;
@@ -285,7 +285,7 @@ where
 
 impl<S, Req> Service<Req> for Box<S>
 where
-    S: Service<Req> + ?Sized,
+    S: Service<Req>,
 {
     type Response = S::Response;
     type Error = S::Error;

ntex-service/src/pipeline.rs

@@ -1,15 +1,14 @@
-use std::{cell::Cell, future, pin::Pin, rc::Rc, task::Context, task::Poll};
+use std::{cell::Cell, future, pin::Pin, rc::Rc, task, task::Context, task::Poll};
 
-use crate::ctx::{ServiceCall, ServiceCtx, Waiters};
+use crate::{ctx::ServiceCall, ctx::Waiters, Service, ServiceCtx, ServiceFactory};
-use crate::{Service, ServiceFactory};
 
 /// Container for a service.
 ///
 /// Container allows to call enclosed service and adds support of shared readiness.
 pub struct Pipeline<S> {
     svc: Rc<S>,
-    waiters: Waiters,
     pending: Cell<bool>,
+    pub(crate) waiters: Waiters,
 }
 
 impl<S> Pipeline<S> {
@@ -55,33 +54,53 @@ impl<S> Pipeline<S> {
         self.svc.poll_shutdown(cx)
     }
 
+    #[deprecated(since = "1.2.3", note = "Use Pipeline::call() instead")]
+    #[doc(hidden)]
     #[inline]
     /// Wait for service readiness and then create future object
     /// that resolves to service result.
-    pub fn service_call<'a, R>(&'a self, req: R) -> ServiceCall<'a, S, R>
+    pub fn service_call<R>(&self, req: R) -> ServiceCall<'_, S, R>
     where
         S: Service<R>,
     {
-        ServiceCtx::<'a, S>::new(&self.waiters).call(self.svc.as_ref(), req)
+        ServiceCall::call_pipeline(req, self)
+    }
+
+    #[inline]
+    /// Wait for service readiness and then create future object
+    /// that resolves to service result.
+    pub fn call<R>(&self, req: R) -> ServiceCall<'_, S, R>
+    where
+        S: Service<R>,
+    {
+        ServiceCall::call_pipeline(req, self)
+    }
+
+    #[inline]
+    /// Wait for service readiness and then create future object
+    /// that resolves to service result.
+    pub fn call_static<R>(&self, req: R) -> PipelineCall<S, R>
+    where
+        S: Service<R> + 'static,
+    {
+        PipelineCall {
+            state: PipelineCallState::Ready { req: Some(req) },
+            pipeline: self.clone(),
+        }
     }
 
     #[inline]
     /// Call service and create future object that resolves to service result.
     ///
     /// Note, this call does not check service readiness.
-    pub fn call<R>(&self, req: R) -> PipelineCall<S, R>
+    pub fn call_nowait<R>(&self, req: R) -> PipelineCall<S, R>
     where
         S: Service<R> + 'static,
-        R: 'static,
     {
-        let pipeline = self.clone();
+        PipelineCall {
-        let svc_call = pipeline.svc.call(req, ServiceCtx::new(&pipeline.waiters));
+            state: PipelineCallState::new_call(self, req),
-
+            pipeline: self.clone(),
-        // SAFETY: `svc_call` has same lifetime same as lifetime of `pipeline.svc`
+        }
-        // Pipeline::svc is heap allocated(Rc<S>), we keep it alive until
-        // `svc_call` get resolved to result
-        let fut = unsafe { std::mem::transmute(svc_call) };
-        PipelineCall { fut, pipeline }
     }
 
     /// Extract service if container hadnt been cloned before.
@@ -119,11 +138,43 @@ pin_project_lite::pin_project! {
         R: 'static,
     {
         #[pin]
-        fut: S::Future<'static>,
+        state: PipelineCallState<S, R>,
         pipeline: Pipeline<S>,
     }
 }
 
+pin_project_lite::pin_project! {
+    #[project = PipelineCallStateProject]
+    enum PipelineCallState<S, Req>
+    where
+        S: Service<Req>,
+        S: 'static,
+        Req: 'static,
+    {
+        Ready { req: Option<Req> },
+        Call { #[pin] fut: S::Future<'static> },
+        Empty,
+    }
+}
+
+impl<S, R> PipelineCallState<S, R>
+where
+    S: Service<R> + 'static,
+    R: 'static,
+{
+    fn new_call(pl: &Pipeline<S>, req: R) -> Self {
+        let ctx = ServiceCtx::new(&pl.waiters);
+        let svc_call = pl.get_ref().call(req, ctx);
+
+        // SAFETY: `svc_call` has same lifetime same as lifetime of `pl.svc`
+        // Pipeline::svc is heap allocated(Rc<S>), we keep it alive until
+        // `svc_call` get resolved to result
+        let fut = unsafe { std::mem::transmute(svc_call) };
+
+        PipelineCallState::Call { fut }
+    }
+}
+
 impl<S, R> future::Future for PipelineCall<S, R>
 where
     S: Service<R>,
@@ -131,8 +182,25 @@ where
     type Output = Result<S::Response, S::Error>;
 
     #[inline]
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.project().fut.poll(cx)
+        let mut this = self.as_mut().project();
+
+        match this.state.as_mut().project() {
+            PipelineCallStateProject::Ready { req } => {
+                task::ready!(this.pipeline.poll_ready(cx))?;
+
+                let st = PipelineCallState::new_call(this.pipeline, req.take().unwrap());
+                this.state.set(st);
+                self.poll(cx)
+            }
+            PipelineCallStateProject::Call { fut, .. } => fut.poll(cx).map(|r| {
+                this.state.set(PipelineCallState::Empty);
+                r
+            }),
+            PipelineCallStateProject::Empty => {
+                panic!("future must not be polled after it returned `Poll::Ready`")
+            }
+        }
     }
 }
 

ntex-tokio/src/io.rs

@@ -128,7 +128,7 @@ impl Future for WriteTask {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.as_mut().get_mut();
+        let this = self.as_mut().get_mut();
 
         match this.st {
             IoWriteState::Processing(ref mut delay) => {
@@ -523,7 +523,7 @@ mod unixstream {
         type Output = ();
 
         fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-            let mut this = self.as_mut().get_mut();
+            let this = self.as_mut().get_mut();
 
             match this.st {
                 IoWriteState::Processing(ref mut delay) => {

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.7.3] - 2023-08-10
+
+* Update ntex-service
+
 ## [0.7.1] - 2023-06-23
 
 * `PipelineCall` is static

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "0.7.2"
+version = "0.7.3"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -52,13 +52,13 @@ ntex-codec = "0.6.2"
 ntex-connect = "0.3.0"
 ntex-http = "0.1.9"
 ntex-router = "0.5.1"
-ntex-service = "1.2.1"
+ntex-service = "1.2.3"
 ntex-macros = "0.1.3"
 ntex-util = "0.3.0"
 ntex-bytes = "0.1.19"
 ntex-h2 = "0.3.2"
 ntex-rt = "0.4.9"
-ntex-io = "0.3.1"
+ntex-io = "0.3.2"
 ntex-tls = "0.3.0"
 ntex-tokio = { version = "0.3.0", optional = true }
 ntex-glommio = { version = "0.3.0", optional = true }

ntex/src/http/client/connect.rs

@@ -30,8 +30,7 @@ where
     ) -> BoxFuture<'_, Result<ClientResponse, SendRequestError>> {
         Box::pin(async move {
             // connect to the host
-            let pl = self.0.clone();
+            let fut = self.0.call(ClientConnect {
-            let fut = pl.service_call(ClientConnect {
                 uri: head.as_ref().uri.clone(),
                 addr,
             });

ntex/src/http/client/pool.rs

@@ -6,7 +6,7 @@ use ntex_h2::{self as h2};
 
 use crate::http::uri::{Authority, Scheme, Uri};
 use crate::io::{types::HttpProtocol, IoBoxed};
-use crate::service::{Pipeline, Service, ServiceCall, ServiceCtx};
+use crate::service::{Pipeline, PipelineCall, Service, ServiceCtx};
 use crate::time::{now, Millis};
 use crate::util::{ready, BoxFuture, ByteString, HashMap, HashSet};
 use crate::{channel::pool, rt::spawn, task::LocalWaker};
@@ -150,7 +150,7 @@ where
                     trace!("Connecting to {:?}", req.uri);
                     let uri = req.uri.clone();
                     let (tx, rx) = waiters.borrow_mut().pool.channel();
-                    OpenConnection::spawn(key, tx, uri, inner, self.connector.clone(), req);
+                    OpenConnection::spawn(key, tx, uri, inner, &self.connector, req);
 
                     match rx.await {
                         Err(_) => Err(ConnectError::Disconnected(None)),
@@ -368,7 +368,7 @@ where
                             tx,
                             uri,
                             this.inner.clone(),
-                            this.connector.clone(),
+                            &this.connector,
                             connect,
                         );
                     }
@@ -385,12 +385,12 @@ where
 }
 
 pin_project_lite::pin_project! {
-    struct OpenConnection<'f, T: Service<Connect>>
+    struct OpenConnection<T: Service<Connect>>
-    where T: 'f
+    where T: 'static
     {
         key: Key,
         #[pin]
-        fut: ServiceCall<'f, T, Connect>,
+        fut: PipelineCall<T, Connect>,
         uri: Uri,
         tx: Option<Waiter>,
         guard: Option<OpenGuard>,
@@ -399,7 +399,7 @@ pin_project_lite::pin_project! {
     }
 }
 
-impl<'f, T> OpenConnection<'f, T>
+impl<T> OpenConnection<T>
 where
     T: Service<Connect, Response = IoBoxed, Error = ConnectError> + 'static,
 {
@@ -408,19 +408,20 @@ where
         tx: Waiter,
         uri: Uri,
         inner: Rc<RefCell<Inner>>,
-        pipeline: Pipeline<T>,
+        pipeline: &Pipeline<T>,
         msg: Connect,
     ) {
+        let fut = pipeline.call_static(msg);
         let disconnect_timeout = inner.borrow().disconnect_timeout;
 
         #[allow(clippy::redundant_async_block)]
         spawn(async move {
             OpenConnection::<T> {
-                fut: pipeline.service_call(msg),
                 tx: Some(tx),
                 key: key.clone(),
                 inner: inner.clone(),
                 guard: Some(OpenGuard::new(key, inner)),
+                fut,
                 uri,
                 disconnect_timeout,
             }
@@ -429,7 +430,7 @@ where
     }
 }
 
-impl<'f, T> Future for OpenConnection<'f, T>
+impl<T> Future for OpenConnection<T>
 where
     T: Service<Connect, Response = IoBoxed, Error = ConnectError>,
 {

ntex/src/http/client/request.rs

@@ -537,7 +537,7 @@ impl ClientRequest {
             if https {
                 slf = slf.set_header_if_none(header::ACCEPT_ENCODING, HTTPS_ENCODING)
             } else {
-                #[cfg(any(feature = "compress"))]
+                #[cfg(feature = "compress")]
                 {
                     slf = slf.set_header_if_none(header::ACCEPT_ENCODING, "gzip, deflate")
                 }

ntex/src/http/h1/dispatcher.rs

@@ -479,21 +479,21 @@ where
     fn service_call(&self, req: Request) -> CallState<S, X> {
         // Handle normal requests
         CallState::Service {
-            fut: self.config.service.call(req),
+            fut: self.config.service.call_nowait(req),
         }
     }
 
     fn service_filter(&self, req: Request, f: &Pipeline<OnRequest>) -> CallState<S, X> {
         // Handle filter fut
         CallState::Filter {
-            fut: f.call((req, self.io.get_ref())),
+            fut: f.call_nowait((req, self.io.get_ref())),
         }
     }
 
     fn service_expect(&self, req: Request) -> CallState<S, X> {
         // Handle normal requests with EXPECT: 100-Continue` header
         CallState::Expect {
-            fut: self.config.expect.call(req),
+            fut: self.config.expect.call_nowait(req),
         }
     }
 
@@ -506,7 +506,7 @@ where
         )));
         // Handle upgrade requests
         CallState::ServiceUpgrade {
-            fut: self.config.service.call(req),
+            fut: self.config.service.call_nowait(req),
         }
     }
 

ntex/src/server/builder.rs

@@ -242,7 +242,7 @@ impl ServerBuilder {
         Ok(self)
     }
 
-    #[cfg(all(unix))]
+    #[cfg(unix)]
     /// Add new unix domain service to the server.
     pub fn bind_uds<F, U, N, R>(self, name: N, addr: U, factory: F) -> io::Result<Self>
     where
@@ -266,7 +266,7 @@ impl ServerBuilder {
         self.listen_uds(name, lst, factory)
     }
 
-    #[cfg(all(unix))]
+    #[cfg(unix)]
     /// Add new unix domain service to the server.
     /// Useful when running as a systemd service and
     /// a socket FD can be acquired using the systemd crate.

ntex/src/server/socket.rs

@@ -174,7 +174,7 @@ mod tests {
     }
 
     #[test]
-    #[cfg(all(unix))]
+    #[cfg(unix)]
     fn uds() {
         use std::os::unix::net::UnixListener;
 

ntex/src/server/worker.rs

@@ -255,9 +255,11 @@ impl Worker {
             self.services.iter_mut().for_each(|srv| {
                 if srv.status == WorkerServiceStatus::Available {
                     srv.status = WorkerServiceStatus::Stopped;
-                    let svc = srv.service.clone();
+                    let fut = srv
+                        .service
+                        .call_static((None, ServerMessage::ForceShutdown));
                     spawn(async move {
-                        let _ = svc.call((None, ServerMessage::ForceShutdown)).await;
+                        let _ = fut.await;
                     });
                 }
             });
@@ -267,9 +269,11 @@ impl Worker {
                 if srv.status == WorkerServiceStatus::Available {
                     srv.status = WorkerServiceStatus::Stopping;
 
-                    let svc = srv.service.clone();
+                    let fut = srv
+                        .service
+                        .call_static((None, ServerMessage::Shutdown(timeout)));
                     spawn(async move {
-                        let _ = svc.call((None, ServerMessage::Shutdown(timeout))).await;
+                        let _ = fut.await;
                     });
                 }
             });
@@ -490,11 +494,11 @@ impl Future for Worker {
                                 self.factories[srv.factory].name(msg.token)
                             );
                         }
-                        let srv = srv.service.clone();
+                        let fut = srv
+                            .service
+                            .call_static((Some(guard), ServerMessage::Connect(msg.io)));
                         spawn(async move {
-                            let _ = srv
+                            let _ = fut.await;
-                                .call((Some(guard), ServerMessage::Connect(msg.io)))
-                                .await;
                         });
                     } else {
                         return Poll::Ready(());

ntex/src/web/test.rs

@@ -107,7 +107,7 @@ where
     S: Service<R, Response = WebResponse, Error = E>,
     E: std::fmt::Debug,
 {
-    app.service_call(req).await.unwrap()
+    app.call(req).await.unwrap()
 }
 
 /// Helper function that returns a response body of a TestRequest
@@ -140,7 +140,7 @@ where
     S: Service<Request, Response = WebResponse>,
 {
     let mut resp = app
-        .service_call(req)
+        .call(req)
         .await
         .unwrap_or_else(|_| panic!("read_response failed at application call"));
 

ntex/src/ws/client.rs

@@ -158,7 +158,7 @@ where
         let msg = Connect::new(head.uri.clone()).set_addr(self.addr);
         log::trace!("Open ws connection to {:?} addr: {:?}", head.uri, self.addr);
 
-        let io = self.connector.clone().service_call(msg).await?;
+        let io = self.connector.call(msg).await?;
 
         // create Framed and send request
         let codec = h1::ClientCodec::default();