Refactor service pipelines

ntex-async-std/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-async-std"
-version = "0.3.0-beta.0"
+version = "0.3.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "async-std intergration for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -17,8 +17,8 @@ path = "src/lib.rs"
 
 [dependencies]
 ntex-bytes = "0.1.19"
-ntex-io = "0.3.0-beta.0"
-ntex-util = "0.3.0-beta.0"
+ntex-io = "0.3.0"
+ntex-util = "0.3.0"
 async-oneshot = "0.5.0"
 log = "0.4"
 pin-project-lite = "0.2"

ntex-bytes/Cargo.toml

@@ -27,4 +27,4 @@ simdutf8 = { version = "0.1.4", optional = true }
 [dev-dependencies]
 serde_test = "1.0"
 serde_json = "1.0"
-ntex = { version = "0.7.0-beta.0", features = ["tokio"] }
+ntex = { version = "0.7.0", features = ["tokio"] }

ntex-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-connect"
-version = "0.3.0-beta.1"
+version = "0.3.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntexwork connect utils for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -34,19 +34,19 @@ glommio = ["ntex-rt/glommio", "ntex-glommio"]
 async-std = ["ntex-rt/async-std", "ntex-async-std"]
 
 [dependencies]
-ntex-service = "1.2.0-beta.1"
+ntex-service = "1.2.0"
 ntex-bytes = "0.1.19"
 ntex-http = "0.1.8"
-ntex-io = "0.3.0-beta.1"
+ntex-io = "0.3.0"
 ntex-rt = "0.4.7"
-ntex-tls = "0.3.0-beta.1"
-ntex-util = "0.3.0-beta.1"
+ntex-tls = "0.3.0"
+ntex-util = "0.3.0"
 log = "0.4"
 thiserror = "1.0"
 
-ntex-tokio = { version = "0.3.0-beta.0", optional = true }
-ntex-glommio = { version = "0.3.0-beta.0", optional = true }
-ntex-async-std = { version = "0.3.0-beta.0", optional = true }
+ntex-tokio = { version = "0.3.0", optional = true }
+ntex-glommio = { version = "0.3.0", optional = true }
+ntex-async-std = { version = "0.3.0", optional = true }
 
 # openssl
 tls-openssl = { version="0.10", package = "openssl", optional = true }
@@ -58,4 +58,4 @@ webpki-roots = { version = "0.23", optional = true }
 [dev-dependencies]
 rand = "0.8"
 env_logger = "0.10"
-ntex = { version = "0.7.0-beta.0", features = ["tokio"] }
+ntex = { version = "0.7.0", features = ["tokio"] }

ntex-connect/src/openssl.rs

@@ -5,14 +5,14 @@ pub use tls_openssl::ssl::{Error as SslError, HandshakeError, SslConnector, SslM
 
 use ntex_bytes::PoolId;
 use ntex_io::{FilterFactory, Io, Layer};
-use ntex_service::{Container, Service, ServiceCtx, ServiceFactory};
+use ntex_service::{Pipeline, Service, ServiceCtx, ServiceFactory};
 use ntex_tls::openssl::SslConnector as IoSslConnector;
 use ntex_util::future::{BoxFuture, Ready};
 
 use super::{Address, Connect, ConnectError, Connector as BaseConnector};
 
 pub struct Connector<T> {
-    connector: Container<BaseConnector<T>>,
+    connector: Pipeline<BaseConnector<T>>,
     openssl: SslConnector,
 }
 
@@ -126,7 +126,7 @@ mod tests {
         let ssl = SslConnector::builder(SslMethod::tls()).unwrap();
         let factory = Connector::new(ssl.build()).memory_pool(PoolId::P5).clone();
 
-        let srv = factory.container(&()).await.unwrap();
+        let srv = factory.pipeline(&()).await.unwrap();
         let result = srv
             .call(Connect::new("").set_addr(Some(server.addr())))
             .await;

ntex-connect/src/resolve.rs

@@ -129,7 +129,7 @@ mod tests {
     async fn resolver() {
         let resolver = Resolver::default().clone();
         assert!(format!("{:?}", resolver).contains("Resolver"));
-        let srv = resolver.container(()).await.unwrap();
+        let srv = resolver.pipeline(()).await.unwrap();
         assert!(lazy(|cx| srv.poll_ready(cx)).await.is_ready());
 
         let res = srv.call(Connect::new("www.rust-lang.org")).await;

ntex-connect/src/rustls.rs

@@ -5,15 +5,15 @@ pub use tls_rustls::{ClientConfig, ServerName};
 
 use ntex_bytes::PoolId;
 use ntex_io::{FilterFactory, Io, Layer};
-use ntex_service::{Container, Service, ServiceCtx, ServiceFactory};
+use ntex_service::{Pipeline, Service, ServiceCtx, ServiceFactory};
 use ntex_tls::rustls::TlsConnector;
 use ntex_util::future::{BoxFuture, Ready};
 
 use super::{Address, Connect, ConnectError, Connector as BaseConnector};
 
 /// Rustls connector factory
 pub struct Connector<T> {
-    connector: Container<BaseConnector<T>>,
+    connector: Pipeline<BaseConnector<T>>,
     inner: TlsConnector,
 }
 
@@ -149,7 +149,7 @@ mod tests {
             .memory_pool(PoolId::P5)
             .clone();
 
-        let srv = factory.container(&()).await.unwrap();
+        let srv = factory.pipeline(&()).await.unwrap();
         // always ready
         assert!(lazy(|cx| srv.poll_ready(cx)).await.is_ready());
         let result = srv

ntex-glommio/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-glommio"
-version = "0.3.0-beta.0"
+version = "0.3.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "glommio intergration for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -17,8 +17,8 @@ path = "src/lib.rs"
 
 [dependencies]
 ntex-bytes = "0.1.19"
-ntex-io = "0.3.0-beta.0"
-ntex-util = "0.3.0-beta.0"
+ntex-io = "0.3.0"
+ntex-util = "0.3.0"
 async-oneshot = "0.5.0"
 futures-lite = "1.12"
 log = "0.4"

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "0.3.0-beta.2"
+version = "0.3.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]
@@ -18,8 +18,8 @@ path = "src/lib.rs"
 [dependencies]
 ntex-codec = "0.6.2"
 ntex-bytes = "0.1.19"
-ntex-util = "0.3.0-beta.1"
-ntex-service = "1.2.0-beta.3"
+ntex-util = "0.3.0"
+ntex-service = "1.2.0"
 
 bitflags = "1.3"
 log = "0.4"
@@ -29,4 +29,4 @@ pin-project-lite = "0.2"
 rand = "0.8"
 env_logger = "0.10"
 
-ntex = { version = "0.7.0-beta.0", features = ["tokio"] }
+ntex = { version = "0.7.0", features = ["tokio"] }

ntex-io/src/dispatcher.rs

@@ -3,7 +3,7 @@ use std::{cell::Cell, future, pin::Pin, rc::Rc, task::Context, task::Poll, time}
 
 use ntex_bytes::Pool;
 use ntex_codec::{Decoder, Encoder};
-use ntex_service::{Container, IntoService, Service};
+use ntex_service::{IntoService, Pipeline, Service};
 use ntex_util::time::Seconds;
 use ntex_util::{future::Either, ready, spawn};
 
@@ -51,7 +51,7 @@ where
 {
     io: IoBoxed,
     codec: U,
-    service: Container<S>,
+    service: Pipeline<S>,
     error: Cell<Option<DispatcherError<S::Error, <U as Encoder>::Error>>>,
     inflight: Cell<usize>,
 }
@@ -107,7 +107,7 @@ where
             codec,
             error: Cell::new(None),
             inflight: Cell::new(0),
-            service: Container::new(service.into_service()),
+            service: Pipeline::new(service.into_service()),
         });
 
         Dispatcher {
@@ -250,7 +250,7 @@ where
                     // call service
                     let shared = slf.shared.clone();
                     shared.inflight.set(shared.inflight.get() + 1);
-                    let fut = shared.service.container_call(item).into_static();
+                    let fut = shared.service.call(item).into_static();
                     spawn(async move {
                         let result = fut.await;
                         shared.handle_result(result, &shared.io);
@@ -276,7 +276,7 @@ where
                     // call service
                     let shared = slf.shared.clone();
                     shared.inflight.set(shared.inflight.get() + 1);
-                    let fut = shared.service.container_call(item).into_static();
+                    let fut = shared.service.call(item).into_static();
                     spawn(async move {
                         let result = fut.await;
                         shared.handle_result(result, &shared.io);
@@ -342,7 +342,7 @@ where
 {
     fn poll_service(
         &self,
-        srv: &Container<S>,
+        srv: &Pipeline<S>,
         cx: &mut Context<'_>,
         io: &IoBoxed,
     ) -> Poll<PollService<U>> {
@@ -478,7 +478,7 @@ mod tests {
                 io: state.into(),
                 error: Cell::new(None),
                 inflight: Cell::new(0),
-                service: Container::new(service.into_service()),
+                service: Pipeline::new(service.into_service()),
             });
 
             (

ntex-io/src/utils.rs

@@ -1,6 +1,6 @@
 use std::marker::PhantomData;
 
-use ntex_service::{fn_service, pipeline_factory, Service, ServiceCtx, ServiceFactory};
+use ntex_service::{chain_factory, fn_service, Service, ServiceCtx, ServiceFactory};
 use ntex_util::future::Ready;
 
 use crate::{Filter, FilterFactory, Io, IoBoxed, Layer};
@@ -20,10 +20,9 @@ where
     S: ServiceFactory<IoBoxed, C>,
     C: Clone,
 {
-    pipeline_factory(
-        fn_service(|io: Io<F>| Ready::Ok(IoBoxed::from(io))).map_init_err(|_| panic!()),
-    )
-    .and_then(srv)
+    chain_factory(fn_service(|io: Io<F>| Ready::Ok(IoBoxed::from(io))))
+        .map_init_err(|_| panic!())
+        .and_then(srv)
 }
 
 /// Create filter factory service
@@ -106,7 +105,7 @@ mod tests {
                 .unwrap();
             Ok::<_, ()>(())
         }))
-        .container(())
+        .pipeline(())
         .await
         .unwrap();
         let _ = svc.call(Io::new(server)).await;
@@ -143,7 +142,7 @@ mod tests {
     #[ntex::test]
     async fn test_utils_filter() {
         let (_, server) = IoTest::create();
-        let svc = pipeline_factory(
+        let svc = chain_factory(
             filter::<_, crate::filter::Base>(TestFilterFactory)
                 .map_err(|_| ())
                 .map_init_err(|_| ()),
@@ -152,7 +151,7 @@ mod tests {
             let _ = io.recv(&BytesCodec).await;
             Ok::<_, ()>(())
         })))
-        .container(())
+        .pipeline(())
         .await
         .unwrap();
         let _ = svc.call(Io::new(server)).await;

ntex-macros/Cargo.toml

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

ntex-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-service"
-version = "1.2.0-beta.4"
+version = "1.2.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex service"
 keywords = ["network", "framework", "async", "futures"]
@@ -20,5 +20,5 @@ pin-project-lite = "0.2.6"
 slab = "0.4"
 
 [dev-dependencies]
-ntex = { version = "0.7.0-beta.1", features = ["tokio"] }
-ntex-util = "0.3.0-beta.1"
+ntex = { version = "0.7.0", features = ["tokio"] }
+ntex-util = "0.3.0"

ntex-service/src/and_then.rs

@@ -234,9 +234,7 @@ where
 mod tests {
     use std::{cell::Cell, rc::Rc, task::Context, task::Poll};
 
-    use crate::{
-        fn_factory, pipeline, pipeline_factory, Service, ServiceCtx, ServiceFactory,
-    };
+    use crate::{chain, chain_factory, fn_factory, Service, ServiceCtx};
     use ntex_util::future::{lazy, Ready};
 
     #[derive(Clone)]
@@ -286,9 +284,7 @@ mod tests {
     #[ntex::test]
     async fn test_poll_ready() {
         let cnt = Rc::new(Cell::new(0));
-        let srv = pipeline(Srv1(cnt.clone()))
-            .and_then(Srv2(cnt.clone()))
-            .clone();
+        let srv = chain(Srv1(cnt.clone())).and_then(Srv2(cnt.clone())).clone();
         let res = lazy(|cx| srv.poll_ready(cx)).await;
         assert_eq!(res, Poll::Ready(Ok(())));
         assert_eq!(cnt.get(), 2);
@@ -299,7 +295,7 @@ mod tests {
     #[ntex::test]
     async fn test_call() {
         let cnt = Rc::new(Cell::new(0));
-        let srv = pipeline(Srv1(cnt.clone())).and_then(Srv2(cnt)).container();
+        let srv = chain(Srv1(cnt.clone())).and_then(Srv2(cnt)).pipeline();
         let res = srv.call("srv1").await;
         assert!(res.is_ok());
         assert_eq!(res.unwrap(), ("srv1", "srv2"));
@@ -309,13 +305,13 @@ mod tests {
     async fn test_factory() {
         let cnt = Rc::new(Cell::new(0));
         let cnt2 = cnt.clone();
-        let new_srv = pipeline_factory(fn_factory(move || {
+        let new_srv = chain_factory(fn_factory(move || {
             Ready::from(Ok::<_, ()>(Srv1(cnt2.clone())))
         }))
         .and_then(move || Ready::from(Ok(Srv2(cnt.clone()))))
         .clone();
 
-        let srv = new_srv.container(&()).await.unwrap();
+        let srv = new_srv.pipeline(&()).await.unwrap();
         let res = srv.call("srv1").await;
         assert!(res.is_ok());
         assert_eq!(res.unwrap(), ("srv1", "srv2"));