add Service::poll_shutdown

.travis.yml

@@ -26,7 +26,7 @@ before_install:
 
 before_cache: |
   if [[ "$TRAVIS_RUST_VERSION" == "nightly-2020-02-16" ]]; then
-    RUSTFLAGS="--cfg procmacro2_semver_exempt" cargo install --version 0.6.11 cargo-tarpaulin
+    RUSTFLAGS="--cfg procmacro2_semver_exempt" cargo install --version 0.10.2 cargo-tarpaulin
   fi
 
 # Add clippy

Cargo.toml

@@ -3,3 +3,7 @@ members = [
   "ntex",
   "ntex-web-macros",
 ]
+
+[patch.crates-io]
+actix-server = { path = "actix-net/actix-server" }
+actix-service = { path = "actix-net/actix-service" }

actix-net/Cargo.toml

@@ -2,7 +2,6 @@
 members = [
   "actix-codec",
   "actix-connect",
-  "actix-ioframe",
   "actix-rt",
   "actix-macros",
   "actix-service",
@@ -19,7 +18,6 @@ members = [
 [patch.crates-io]
 actix-codec = { path = "actix-codec" }
 actix-connect = { path = "actix-connect" }
-actix-ioframe = { path = "actix-ioframe" }
 actix-rt = { path = "actix-rt" }
 actix-macros = { path = "actix-macros" }
 actix-server = { path = "actix-server" }

actix-net/actix-service/src/and_then.rs

@@ -49,6 +49,19 @@ where
         }
     }
 
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        let srv = self.0.get_mut();
+
+        if srv.0.poll_shutdown(cx, is_error).is_ready()
+            && srv.1.poll_shutdown(cx, is_error).is_ready()
+        {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+
+    #[inline]
     fn call(&mut self, req: A::Request) -> Self::Future {
         AndThenServiceResponse {
             state: State::A(self.0.get_mut().0.call(req), Some(self.0.clone())),

actix-net/actix-service/src/and_then_apply_fn.rs

@@ -76,6 +76,18 @@ where
         }
     }
 
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        let srv = self.srv.get_mut();
+
+        if srv.0.poll_shutdown(cx, is_error).is_ready()
+            && srv.1.poll_shutdown(cx, is_error).is_ready()
+        {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+
     fn call(&mut self, req: A::Request) -> Self::Future {
         let fut = self.srv.get_mut().0.call(req);
         AndThenApplyFnFuture {

actix-net/actix-service/src/apply.rs

@@ -82,10 +82,17 @@ where
     type Error = Err;
     type Future = R;
 
+    #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(futures_util::ready!(self.service.poll_ready(cx)))
     }
 
+    #[inline]
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        self.service.poll_shutdown(cx, is_error)
+    }
+
+    #[inline]
     fn call(&mut self, req: In) -> Self::Future {
         (self.f)(req, &mut self.service)
     }

actix-net/actix-service/src/boxed.rs

@@ -135,10 +135,17 @@ where
     type Error = Err;
     type Future = BoxFuture<Res, Err>;
 
+    #[inline]
     fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.0.poll_ready(ctx)
     }
 
+    #[inline]
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        self.0.poll_shutdown(cx, is_error)
+    }
+
+    #[inline]
     fn call(&mut self, req: Self::Request) -> Self::Future {
         Box::pin(self.0.call(req))
     }

actix-net/actix-service/src/cell.rs

@@ -51,6 +51,10 @@ impl<T: crate::Service> crate::Service for Cell<T> {
         self.get_mut().poll_ready(cx)
     }
 
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        self.get_mut().poll_shutdown(cx, is_error)
+    }
+
     fn call(&mut self, req: Self::Request) -> Self::Future {
         self.get_mut().call(req)
     }

actix-net/actix-service/src/lib.rs

@@ -97,6 +97,16 @@ pub trait Service {
     /// 2. In case of chained services, `.poll_ready()` get called for all services at once.
     fn poll_ready(&mut self, ctx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>>;
 
+    #[inline]
+    #[allow(unused_variables)]
+    /// Shutdown service.
+    ///
+    /// Returns `Ready` when the service is properly shutdowned. This method might be called
+    /// after it returns `Ready`.
+    fn poll_shutdown(&mut self, ctx: &mut task::Context<'_>, is_error: bool) -> Poll<()> {
+        Poll::Ready(())
+    }
+
     /// Process the request and return the response asynchronously.
     ///
     /// This function is expected to be callable off task. As such,
@@ -108,6 +118,7 @@ pub trait Service {
     /// implementation must be resilient to this fact.
     fn call(&mut self, req: Self::Request) -> Self::Future;
 
+    #[inline]
     /// Map this service's output to a different type, returning a new service
     /// of the resulting type.
     ///
@@ -125,6 +136,7 @@ pub trait Service {
         crate::dev::Map::new(self, f)
     }
 
+    #[inline]
     /// Map this service's error to a different error, returning a new service.
     ///
     /// This function is similar to the `Result::map_err` where it will change

actix-net/actix-service/src/map.rs

@@ -53,10 +53,17 @@ where
     type Error = A::Error;
     type Future = MapFuture<A, F, Response>;
 
-    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    #[inline]
-        self.service.poll_ready(ctx)
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.service.poll_ready(cx)
     }
 
+    #[inline]
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        self.service.poll_shutdown(cx, is_error)
+    }
+
+    #[inline]
     fn call(&mut self, req: A::Request) -> Self::Future {
         MapFuture::new(self.service.call(req), self.f.clone())
     }

actix-net/actix-service/src/map_err.rs

@@ -54,10 +54,17 @@ where
     type Error = E;
     type Future = MapErrFuture<A, F, E>;
 
-    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    #[inline]
-        self.service.poll_ready(ctx).map_err(&self.f)
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.service.poll_ready(cx).map_err(&self.f)
     }
 
+    #[inline]
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        self.service.poll_shutdown(cx, is_error)
+    }
+
+    #[inline]
     fn call(&mut self, req: A::Request) -> Self::Future {
         MapErrFuture::new(self.service.call(req), self.f.clone())
     }

actix-net/actix-service/src/pipeline.rs

@@ -161,8 +161,13 @@ impl<T: Service> Service for Pipeline<T> {
     type Future = T::Future;
 
     #[inline]
-    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), T::Error>> {
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), T::Error>> {
-        self.service.poll_ready(ctx)
+        self.service.poll_ready(cx)
+    }
+
+    #[inline]
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        self.service.poll_shutdown(cx, is_error)
     }
 
     #[inline]

actix-net/actix-service/src/then.rs

@@ -49,6 +49,18 @@ where
         }
     }
 
+    fn poll_shutdown(&mut self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
+        let srv = self.0.get_mut();
+
+        if srv.0.poll_shutdown(cx, is_error).is_ready()
+            && srv.1.poll_shutdown(cx, is_error).is_ready()
+        {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+
     fn call(&mut self, req: A::Request) -> Self::Future {
         ThenServiceResponse {
             state: State::A(self.0.get_mut().0.call(req), Some(self.0.clone())),

ntex/src/server/test.rs

@@ -59,10 +59,7 @@ pub fn test_server<F: ServiceFactory<TcpStream>>(factory: F) -> TestServer {
 
     let (system, addr) = rx.recv().unwrap();
 
-    TestServer {
+    TestServer { addr, system }
-        addr,
-        system,
-    }
 }
 
 /// Test server controller