bump service to stable v2

Cargo.toml

@@ -61,7 +61,7 @@ actix-macros = "0.2.0"
 actix-router = "0.2.7"
 actix-rt = "2.2"
 actix-server = "2.0.0-beta.3"
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0"
 actix-utils = "3.0.0-beta.4"
 actix-tls = { version = "3.0.0-beta.5", default-features = false, optional = true }
 

actix-files/Cargo.toml

@@ -18,7 +18,7 @@ path = "src/lib.rs"
 
 [dependencies]
 actix-web = { version = "4.0.0-beta.5", default-features = false }
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0"
 actix-utils = "3.0.0-beta.4"
 
 askama_escape = "0.10"

actix-http-test/Cargo.toml

@@ -29,7 +29,7 @@ default = []
 openssl = ["tls-openssl", "awc/openssl"]
 
 [dependencies]
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0"
 actix-codec = "0.4.0-beta.1"
 actix-tls = "3.0.0-beta.5"
 actix-utils = "3.0.0-beta.4"

actix-http/Cargo.toml

@@ -38,7 +38,7 @@ compress = ["flate2", "brotli2"]
 trust-dns = ["trust-dns-resolver"]
 
 [dependencies]
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0"
 actix-codec = "0.4.0-beta.1"
 actix-utils = "3.0.0-beta.4"
 actix-rt = "2.2"

actix-http/src/h1/service.rs

@@ -5,7 +5,9 @@ use std::{fmt, net};
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
 use actix_rt::net::TcpStream;
-use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
+use actix_service::{
+    fn_service, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _,
+};
 use actix_utils::future::ready;
 use futures_core::future::LocalBoxFuture;
 
@@ -82,7 +84,7 @@ where
         Error = DispatchError,
         InitError = (),
     > {
-        pipeline_factory(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
             let peer_addr = io.peer_addr().ok();
             ready(Ok((io, peer_addr)))
         })
@@ -132,16 +134,14 @@ mod openssl {
             Error = TlsError<SslError, DispatchError>,
             InitError = (),
         > {
-            pipeline_factory(
+            Acceptor::new(acceptor)
-                Acceptor::new(acceptor)
+                .map_err(TlsError::Tls)
-                    .map_err(TlsError::Tls)
+                .map_init_err(|_| panic!())
-                    .map_init_err(|_| panic!()),
+                .and_then(|io: TlsStream<TcpStream>| {
-            )
+                    let peer_addr = io.get_ref().peer_addr().ok();
-            .and_then(|io: TlsStream<TcpStream>| {
+                    ready(Ok((io, peer_addr)))
-                let peer_addr = io.get_ref().peer_addr().ok();
+                })
-                ready(Ok((io, peer_addr)))
+                .and_then(self.map_err(TlsError::Service))
-            })
-            .and_then(self.map_err(TlsError::Service))
         }
     }
 }
@@ -190,16 +190,14 @@ mod rustls {
             Error = TlsError<io::Error, DispatchError>,
             InitError = (),
         > {
-            pipeline_factory(
+            Acceptor::new(config)
-                Acceptor::new(config)
+                .map_err(TlsError::Tls)
-                    .map_err(TlsError::Tls)
+                .map_init_err(|_| panic!())
-                    .map_init_err(|_| panic!()),
+                .and_then(|io: TlsStream<TcpStream>| {
-            )
+                    let peer_addr = io.get_ref().0.peer_addr().ok();
-            .and_then(|io: TlsStream<TcpStream>| {
+                    ready(Ok((io, peer_addr)))
-                let peer_addr = io.get_ref().0.peer_addr().ok();
+                })
-                ready(Ok((io, peer_addr)))
+                .and_then(self.map_err(TlsError::Service))
-            })
-            .and_then(self.map_err(TlsError::Service))
         }
     }
 }

actix-http/src/h2/service.rs

@@ -7,8 +7,8 @@ use std::{net, rc::Rc};
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_rt::net::TcpStream;
 use actix_service::{
-    fn_factory, fn_service, pipeline_factory, IntoServiceFactory, Service,
+    fn_factory, fn_service, IntoServiceFactory, Service, ServiceFactory,
-    ServiceFactory,
+    ServiceFactoryExt as _,
 };
 use actix_utils::future::ready;
 use bytes::Bytes;
@@ -81,12 +81,12 @@ where
         Error = DispatchError,
         InitError = S::InitError,
     > {
-        pipeline_factory(fn_factory(|| {
+        fn_factory(|| {
             ready(Ok::<_, S::InitError>(fn_service(|io: TcpStream| {
                 let peer_addr = io.peer_addr().ok();
                 ready(Ok::<_, DispatchError>((io, peer_addr)))
             })))
-        }))
+        })
         .and_then(self)
     }
 }
@@ -119,20 +119,18 @@ mod openssl {
             Error = TlsError<SslError, DispatchError>,
             InitError = S::InitError,
         > {
-            pipeline_factory(
+            Acceptor::new(acceptor)
-                Acceptor::new(acceptor)
+                .map_err(TlsError::Tls)
-                    .map_err(TlsError::Tls)
+                .map_init_err(|_| panic!())
-                    .map_init_err(|_| panic!()),
+                .and_then(fn_factory(|| {
-            )
+                    ready(Ok::<_, S::InitError>(fn_service(
-            .and_then(fn_factory(|| {
+                        |io: TlsStream<TcpStream>| {
-                ready(Ok::<_, S::InitError>(fn_service(
+                            let peer_addr = io.get_ref().peer_addr().ok();
-                    |io: TlsStream<TcpStream>| {
+                            ready(Ok((io, peer_addr)))
-                        let peer_addr = io.get_ref().peer_addr().ok();
+                        },
-                        ready(Ok((io, peer_addr)))
+                    )))
-                    },
+                }))
-                )))
+                .and_then(self.map_err(TlsError::Service))
-            }))
-            .and_then(self.map_err(TlsError::Service))
         }
     }
 }
@@ -168,20 +166,18 @@ mod rustls {
             let protos = vec!["h2".to_string().into()];
             config.set_protocols(&protos);
 
-            pipeline_factory(
+            Acceptor::new(config)
-                Acceptor::new(config)
+                .map_err(TlsError::Tls)
-                    .map_err(TlsError::Tls)
+                .map_init_err(|_| panic!())
-                    .map_init_err(|_| panic!()),
+                .and_then(fn_factory(|| {
-            )
+                    ready(Ok::<_, S::InitError>(fn_service(
-            .and_then(fn_factory(|| {
+                        |io: TlsStream<TcpStream>| {
-                ready(Ok::<_, S::InitError>(fn_service(
+                            let peer_addr = io.get_ref().0.peer_addr().ok();
-                    |io: TlsStream<TcpStream>| {
+                            ready(Ok((io, peer_addr)))
-                        let peer_addr = io.get_ref().0.peer_addr().ok();
+                        },
-                        ready(Ok((io, peer_addr)))
+                    )))
-                    },
+                }))
-                )))
+                .and_then(self.map_err(TlsError::Service))
-            }))
-            .and_then(self.map_err(TlsError::Service))
         }
     }
 }

actix-http/src/service.rs

@@ -10,7 +10,9 @@ use std::{
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
 use actix_rt::net::TcpStream;
-use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
+use actix_service::{
+    fn_service, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _,
+};
 use bytes::Bytes;
 use futures_core::{future::LocalBoxFuture, ready};
 use h2::server::{handshake, Handshake};
@@ -180,7 +182,7 @@ where
         Error = DispatchError,
         InitError = (),
     > {
-        pipeline_factory(|io: TcpStream| async {
+        fn_service(|io: TcpStream| async {
             let peer_addr = io.peer_addr().ok();
             Ok((io, Protocol::Http1, peer_addr))
         })
@@ -232,25 +234,23 @@ mod openssl {
             Error = TlsError<SslError, DispatchError>,
             InitError = (),
         > {
-            pipeline_factory(
+            Acceptor::new(acceptor)
-                Acceptor::new(acceptor)
+                .map_err(TlsError::Tls)
-                    .map_err(TlsError::Tls)
+                .map_init_err(|_| panic!())
-                    .map_init_err(|_| panic!()),
+                .and_then(|io: TlsStream<TcpStream>| async {
-            )
+                    let proto = if let Some(protos) = io.ssl().selected_alpn_protocol() {
-            .and_then(|io: TlsStream<TcpStream>| async {
+                        if protos.windows(2).any(|window| window == b"h2") {
-                let proto = if let Some(protos) = io.ssl().selected_alpn_protocol() {
+                            Protocol::Http2
-                    if protos.windows(2).any(|window| window == b"h2") {
+                        } else {
-                        Protocol::Http2
+                            Protocol::Http1
+                        }
                     } else {
                         Protocol::Http1
-                    }
+                    };
-                } else {
+                    let peer_addr = io.get_ref().peer_addr().ok();
-                    Protocol::Http1
+                    Ok((io, proto, peer_addr))
-                };
+                })
-                let peer_addr = io.get_ref().peer_addr().ok();
+                .and_then(self.map_err(TlsError::Service))
-                Ok((io, proto, peer_addr))
-            })
-            .and_then(self.map_err(TlsError::Service))
         }
     }
 }
@@ -304,25 +304,24 @@ mod rustls {
             let protos = vec!["h2".to_string().into(), "http/1.1".to_string().into()];
             config.set_protocols(&protos);
 
-            pipeline_factory(
+            Acceptor::new(config)
-                Acceptor::new(config)
+                .map_err(TlsError::Tls)
-                    .map_err(TlsError::Tls)
+                .map_init_err(|_| panic!())
-                    .map_init_err(|_| panic!()),
+                .and_then(|io: TlsStream<TcpStream>| async {
-            )
+                    let proto = if let Some(protos) = io.get_ref().1.get_alpn_protocol()
-            .and_then(|io: TlsStream<TcpStream>| async {
+                    {
-                let proto = if let Some(protos) = io.get_ref().1.get_alpn_protocol() {
+                        if protos.windows(2).any(|window| window == b"h2") {
-                    if protos.windows(2).any(|window| window == b"h2") {
+                            Protocol::Http2
-                        Protocol::Http2
+                        } else {
+                            Protocol::Http1
+                        }
                     } else {
                         Protocol::Http1
-                    }
+                    };
-                } else {
+                    let peer_addr = io.get_ref().0.peer_addr().ok();
-                    Protocol::Http1
+                    Ok((io, proto, peer_addr))
-                };
+                })
-                let peer_addr = io.get_ref().0.peer_addr().ok();
+                .and_then(self.map_err(TlsError::Service))
-                Ok((io, proto, peer_addr))
-            })
-            .and_then(self.map_err(TlsError::Service))
         }
     }
 }

actix-test/Cargo.toml

@@ -22,7 +22,7 @@ openssl = ["tls-openssl", "actix-http/openssl"]
 actix-codec = "0.4.0-beta.1"
 actix-http = "3.0.0-beta.5"
 actix-http-test = { version = "3.0.0-beta.4", features = [] }
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0"
 actix-utils = "3.0.0-beta.2"
 actix-web = { version = "4.0.0-beta.5", default-features = false, features = ["cookies"] }
 actix-rt = "2.1"

awc/Cargo.toml

@@ -48,7 +48,7 @@ trust-dns = ["actix-http/trust-dns"]
 
 [dependencies]
 actix-codec = "0.4.0-beta.1"
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0"
 actix-http = "3.0.0-beta.5"
 actix-rt = { version = "2.1", default-features = false }
 

awc/tests/test_client.rs

@@ -20,7 +20,7 @@ use actix_http::{
     HttpService,
 };
 use actix_http_test::test_server;
-use actix_service::{map_config, pipeline_factory};
+use actix_service::{fn_service, map_config, ServiceFactoryExt as _};
 use actix_web::{
     dev::{AppConfig, BodyEncoding},
     http::header,
@@ -239,7 +239,7 @@ async fn test_connection_reuse() {
 
     let srv = test_server(move || {
         let num2 = num2.clone();
-        pipeline_factory(move |io| {
+        fn_service(move |io| {
             num2.fetch_add(1, Ordering::Relaxed);
             ok(io)
         })
@@ -276,7 +276,7 @@ async fn test_connection_force_close() {
 
     let srv = test_server(move || {
         let num2 = num2.clone();
-        pipeline_factory(move |io| {
+        fn_service(move |io| {
             num2.fetch_add(1, Ordering::Relaxed);
             ok(io)
         })
@@ -313,7 +313,7 @@ async fn test_connection_server_close() {
 
     let srv = test_server(move || {
         let num2 = num2.clone();
-        pipeline_factory(move |io| {
+        fn_service(move |io| {
             num2.fetch_add(1, Ordering::Relaxed);
             ok(io)
         })
@@ -353,7 +353,7 @@ async fn test_connection_wait_queue() {
 
     let srv = test_server(move || {
         let num2 = num2.clone();
-        pipeline_factory(move |io| {
+        fn_service(move |io| {
             num2.fetch_add(1, Ordering::Relaxed);
             ok(io)
         })
@@ -401,7 +401,7 @@ async fn test_connection_wait_queue_force_close() {
 
     let srv = test_server(move || {
         let num2 = num2.clone();
-        pipeline_factory(move |io| {
+        fn_service(move |io| {
             num2.fetch_add(1, Ordering::Relaxed);
             ok(io)
         })

awc/tests/test_ssl_client.rs

@@ -7,7 +7,7 @@ use std::sync::Arc;
 
 use actix_http::HttpService;
 use actix_http_test::test_server;
-use actix_service::{map_config, pipeline_factory, ServiceFactoryExt};
+use actix_service::{map_config, fn_service, ServiceFactoryExt};
 use actix_utils::future::ok;
 use actix_web::http::Version;
 use actix_web::{dev::AppConfig, web, App, HttpResponse};
@@ -48,7 +48,7 @@ async fn test_connection_reuse_h2() {
 
     let srv = test_server(move || {
         let num2 = num2.clone();
-        pipeline_factory(move |io| {
+        fn_service(move |io| {
             num2.fetch_add(1, Ordering::Relaxed);
             ok(io)
         })

src/http/header/cache_control.rs

@@ -1,9 +1,7 @@
 use std::fmt::{self, Write};
 use std::str::FromStr;
 
-use super::{
+use super::{fmt_comma_delimited, from_comma_delimited, Header, IntoHeaderValue, Writer};
-    fmt_comma_delimited, from_comma_delimited, Header, IntoHeaderValue, Writer,
-};
 
 use crate::http::header;
 
@@ -176,9 +174,7 @@ impl FromStr for CacheDirective {
                         ("max-stale", secs) => secs.parse().map(MaxStale).map_err(Some),
                         ("min-fresh", secs) => secs.parse().map(MinFresh).map_err(Some),
                         ("s-maxage", secs) => secs.parse().map(SMaxAge).map_err(Some),
-                        (left, right) => {
+                        (left, right) => Ok(Extension(left.to_owned(), Some(right.to_owned()))),
-                            Ok(Extension(left.to_owned(), Some(right.to_owned())))
-                        }
                     }
                 }
                 Some(_) => Err(None),

src/http/header/content_disposition.rs

@@ -10,8 +10,8 @@ use once_cell::sync::Lazy;
 use regex::Regex;
 use std::fmt::{self, Write};
 
-use crate::http::header;
 use super::{ExtendedValue, Header, IntoHeaderValue, Writer};
+use crate::http::header;
 
 /// Split at the index of the first `needle` if it exists or at the end.
 fn split_once(haystack: &str, needle: char) -> (&str, &str) {
@@ -403,11 +403,7 @@ impl ContentDisposition {
     /// Returns `true` if it is [`Ext`](DispositionType::Ext) and the `disp_type` matches.
     pub fn is_ext<T: AsRef<str>>(&self, disp_type: T) -> bool {
         match self.disposition {
-            DispositionType::Ext(ref t)
+            DispositionType::Ext(ref t) if t.eq_ignore_ascii_case(disp_type.as_ref()) => true,
-                if t.eq_ignore_ascii_case(disp_type.as_ref()) =>
-            {
-                true
-            }
             _ => false,
         }
     }
@@ -521,7 +517,8 @@ impl fmt::Display for DispositionParam {
         //
         //
         // See also comments in test_from_raw_unnecessary_percent_decode.
-        static RE: Lazy<Regex> = Lazy::new(|| Regex::new("[\x00-\x08\x10-\x1F\x7F\"\\\\]").unwrap());
+        static RE: Lazy<Regex> =
+            Lazy::new(|| Regex::new("[\x00-\x08\x10-\x1F\x7F\"\\\\]").unwrap());
         match self {
             DispositionParam::Name(ref value) => write!(f, "name={}", value),
             DispositionParam::Filename(ref value) => {
@@ -618,8 +615,8 @@ mod tests {
                 charset: Charset::Ext(String::from("UTF-8")),
                 language_tag: None,
                 value: vec![
-                    0xc2, 0xa3, 0x20, b'a', b'n', b'd', 0x20, 0xe2, 0x82, 0xac, 0x20,
+                    0xc2, 0xa3, 0x20, b'a', b'n', b'd', 0x20, 0xe2, 0x82, 0xac, 0x20, b'r',
-                    b'r', b'a', b't', b'e', b's',
+                    b'a', b't', b'e', b's',
                 ],
             })],
         };
@@ -635,8 +632,8 @@ mod tests {
                 charset: Charset::Ext(String::from("UTF-8")),
                 language_tag: None,
                 value: vec![
-                    0xc2, 0xa3, 0x20, b'a', b'n', b'd', 0x20, 0xe2, 0x82, 0xac, 0x20,
+                    0xc2, 0xa3, 0x20, b'a', b'n', b'd', 0x20, 0xe2, 0x82, 0xac, 0x20, b'r',
-                    b'r', b'a', b't', b'e', b's',
+                    b'a', b't', b'e', b's',
                 ],
             })],
         };
@@ -698,26 +695,22 @@ mod tests {
 
     #[test]
     fn test_from_raw_only_disp() {
-        let a = ContentDisposition::from_raw(&HeaderValue::from_static("attachment"))
+        let a = ContentDisposition::from_raw(&HeaderValue::from_static("attachment")).unwrap();
-            .unwrap();
         let b = ContentDisposition {
             disposition: DispositionType::Attachment,
             parameters: vec![],
         };
         assert_eq!(a, b);
 
-        let a =
+        let a = ContentDisposition::from_raw(&HeaderValue::from_static("inline ;")).unwrap();
-            ContentDisposition::from_raw(&HeaderValue::from_static("inline ;")).unwrap();
         let b = ContentDisposition {
             disposition: DispositionType::Inline,
             parameters: vec![],
         };
         assert_eq!(a, b);
 
-        let a = ContentDisposition::from_raw(&HeaderValue::from_static(
+        let a = ContentDisposition::from_raw(&HeaderValue::from_static("unknown-disp-param"))
-            "unknown-disp-param",
+            .unwrap();
-        ))
-        .unwrap();
         let b = ContentDisposition {
             disposition: DispositionType::Ext(String::from("unknown-disp-param")),
             parameters: vec![],
@@ -756,8 +749,8 @@ mod tests {
         Mainstream browsers like Firefox (gecko) and Chrome use UTF-8 directly as above.
         (And now, only UTF-8 is handled by this implementation.)
         */
-        let a = HeaderValue::from_str("form-data; name=upload; filename=\"文件.webp\"")
+        let a =
-            .unwrap();
+            HeaderValue::from_str("form-data; name=upload; filename=\"文件.webp\"").unwrap();
         let a: ContentDisposition = ContentDisposition::from_raw(&a).unwrap();
         let b = ContentDisposition {
             disposition: DispositionType::FormData,
@@ -808,8 +801,7 @@ mod tests {
 
     #[test]
     fn test_from_raw_semicolon() {
-        let a =
+        let a = HeaderValue::from_static("form-data; filename=\"A semicolon here;.pdf\"");
-            HeaderValue::from_static("form-data; filename=\"A semicolon here;.pdf\"");
         let a: ContentDisposition = ContentDisposition::from_raw(&a).unwrap();
         let b = ContentDisposition {
             disposition: DispositionType::FormData,
@@ -845,9 +837,8 @@ mod tests {
         };
         assert_eq!(a, b);
 
-        let a = HeaderValue::from_static(
+        let a =
-            "form-data; name=photo; filename=\"%74%65%73%74.png\"",
+            HeaderValue::from_static("form-data; name=photo; filename=\"%74%65%73%74.png\"");
-        );
         let a: ContentDisposition = ContentDisposition::from_raw(&a).unwrap();
         let b = ContentDisposition {
             disposition: DispositionType::FormData,
@@ -892,8 +883,7 @@ mod tests {
 
     #[test]
     fn test_display_extended() {
-        let as_string =
+        let as_string = "attachment; filename*=UTF-8'en'%C2%A3%20and%20%E2%82%AC%20rates";
-            "attachment; filename*=UTF-8'en'%C2%A3%20and%20%E2%82%AC%20rates";
         let a = HeaderValue::from_static(as_string);
         let a: ContentDisposition = ContentDisposition::from_raw(&a).unwrap();
         let display_rendered = format!("{}", a);

src/http/header/content_range.rs

@@ -1,10 +1,8 @@
 use std::fmt::{self, Display, Write};
 use std::str::FromStr;
 
+use super::{HeaderValue, IntoHeaderValue, InvalidHeaderValue, Writer, CONTENT_RANGE};
 use crate::error::ParseError;
-use super::{
-    HeaderValue, IntoHeaderValue, InvalidHeaderValue, Writer, CONTENT_RANGE,
-};
 
 crate::__define_common_header! {
     /// `Content-Range` header, defined in
@@ -141,8 +139,7 @@ impl FromStr for ContentRangeSpec {
                 } else {
                     let (first_byte, last_byte) =
                         split_in_two(range, '-').ok_or(ParseError::Header)?;
-                    let first_byte =
+                    let first_byte = first_byte.parse().map_err(|_| ParseError::Header)?;
-                        first_byte.parse().map_err(|_| ParseError::Header)?;
                     let last_byte = last_byte.parse().map_err(|_| ParseError::Header)?;
                     if last_byte < first_byte {
                         return Err(ParseError::Header);

src/http/header/if_range.rs

@@ -1,11 +1,11 @@
 use std::fmt::{self, Display, Write};
 
-use crate::http::header;
-use crate::error::ParseError;
 use super::{
-    from_one_raw_str, EntityTag, Header, HeaderName, HeaderValue, HttpDate,
+    from_one_raw_str, EntityTag, Header, HeaderName, HeaderValue, HttpDate, IntoHeaderValue,
-    IntoHeaderValue, InvalidHeaderValue, Writer,
+    InvalidHeaderValue, Writer,
 };
+use crate::error::ParseError;
+use crate::http::header;
 use crate::HttpMessage;
 
 /// `If-Range` header, defined in [RFC7233](http://tools.ietf.org/html/rfc7233#section-3.2)
@@ -76,13 +76,11 @@ impl Header for IfRange {
     where
         T: HttpMessage,
     {
-        let etag: Result<EntityTag, _> =
+        let etag: Result<EntityTag, _> = from_one_raw_str(msg.headers().get(&header::IF_RANGE));
-            from_one_raw_str(msg.headers().get(&header::IF_RANGE));
         if let Ok(etag) = etag {
             return Ok(IfRange::EntityTag(etag));
         }
-        let date: Result<HttpDate, _> =
+        let date: Result<HttpDate, _> = from_one_raw_str(msg.headers().get(&header::IF_RANGE));
-            from_one_raw_str(msg.headers().get(&header::IF_RANGE));
         if let Ok(date) = date {
             return Ok(IfRange::Date(date));
         }

src/server.rs

@@ -489,7 +489,7 @@ where
     pub fn listen_uds(mut self, lst: std::os::unix::net::UnixListener) -> io::Result<Self> {
         use actix_http::Protocol;
         use actix_rt::net::UnixStream;
-        use actix_service::pipeline_factory;
+        use actix_service::{fn_service, ServiceFactoryExt as _};
 
         let cfg = self.config.clone();
         let factory = self.factory.clone();
@@ -511,22 +511,19 @@ where
                 socket_addr,
             );
 
-            pipeline_factory(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) })
+            fn_service(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) }).and_then({
-                .and_then({
+                let svc = HttpService::build()
-                    let svc = HttpService::build()
+                    .keep_alive(c.keep_alive)
-                        .keep_alive(c.keep_alive)
+                    .client_timeout(c.client_timeout);
-                        .client_timeout(c.client_timeout);
 
-                    let svc = if let Some(handler) = on_connect_fn.clone() {
+                let svc = if let Some(handler) = on_connect_fn.clone() {
-                        svc.on_connect_ext(move |io: &_, ext: _| {
+                    svc.on_connect_ext(move |io: &_, ext: _| (&*handler)(io as &dyn Any, ext))
-                            (&*handler)(io as &dyn Any, ext)
+                } else {
-                        })
+                    svc
-                    } else {
+                };
-                        svc
-                    };
 
-                    svc.finish(map_config(factory(), move |_| config.clone()))
+                svc.finish(map_config(factory(), move |_| config.clone()))
-                })
+            })
         })?;
         Ok(self)
     }
@@ -539,7 +536,7 @@ where
     {
         use actix_http::Protocol;
         use actix_rt::net::UnixStream;
-        use actix_service::pipeline_factory;
+        use actix_service::{fn_service, ServiceFactoryExt as _};
 
         let cfg = self.config.clone();
         let factory = self.factory.clone();
@@ -560,13 +557,12 @@ where
                     c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
                     socket_addr,
                 );
-                pipeline_factory(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) })
+                fn_service(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) }).and_then(
-                    .and_then(
+                    HttpService::build()
-                        HttpService::build()
+                        .keep_alive(c.keep_alive)
-                            .keep_alive(c.keep_alive)
+                        .client_timeout(c.client_timeout)
-                            .client_timeout(c.client_timeout)
+                        .finish(map_config(factory(), move |_| config.clone())),
-                            .finish(map_config(factory(), move |_| config.clone())),
+                )
-                    )
             },
         )?;
         Ok(self)