Merge pull request #75 from ntex-rs/refactor-server-factory

Refactor server factory trait

ntex-tls/examples/rustls-server.rs

@@ -35,7 +35,7 @@ async fn main() -> io::Result<()> {
 
     // start server
     server::ServerBuilder::new()
-        .bind("basic", "127.0.0.1:8443", move || {
+        .bind("basic", "127.0.0.1:8443", move |_| {
             pipeline_factory(filter_factory(TlsAcceptor::new(tls_config.clone())))
                 .and_then(fn_service(|io: Io<_>| async move {
                     println!("New client is connected");

ntex-tls/examples/server.rs

@@ -24,7 +24,7 @@ async fn main() -> io::Result<()> {
 
     // start server
     server::ServerBuilder::new()
-        .bind("basic", "127.0.0.1:8443", move || {
+        .bind("basic", "127.0.0.1:8443", move |_| {
             pipeline_factory(filter_factory(SslAcceptor::new(acceptor.clone())))
                 .and_then(fn_service(|io: Io<_>| async move {
                     println!("New client is connected");

ntex-tls/examples/webserver.rs

@@ -37,7 +37,7 @@ async fn main() -> io::Result<()> {
 
     // start server
     server::ServerBuilder::new()
-        .bind("basic", "127.0.0.1:8443", move || {
+        .bind("basic", "127.0.0.1:8443", move |_| {
             HttpService::build()
                 .client_timeout(Seconds(1))
                 .disconnect_timeout(Seconds(1))

ntex/CHANGES.md

@@ -4,6 +4,8 @@
 
 * Refactor http/1 dispatcher
 
+* Refactor Server service configuration
+
 ## [0.5.0-b.0] - 2021-12-19
 
 * Migrate io to ntex-io

ntex/examples/echo.rs

@@ -13,7 +13,7 @@ async fn main() -> io::Result<()> {
     env_logger::init();
 
     Server::build()
-        .bind("echo", "127.0.0.1:8080", || {
+        .bind("echo", "127.0.0.1:8080", |_| {
             HttpService::build()
                 .client_timeout(Seconds(1))
                 .disconnect_timeout(Seconds(1))

ntex/examples/echo2.rs

@@ -25,7 +25,7 @@ async fn main() -> io::Result<()> {
     env_logger::init();
 
     Server::build()
-        .bind("echo", "127.0.0.1:8080", || {
+        .bind("echo", "127.0.0.1:8080", |_| {
             HttpService::build().finish(handle_request)
         })?
         .run()

ntex/examples/hello-world.rs

@@ -11,7 +11,7 @@ async fn main() -> io::Result<()> {
     env_logger::init();
 
     Server::build()
-        .bind("hello-world", "127.0.0.1:8080", || {
+        .bind("hello-world", "127.0.0.1:8080", |_| {
             HttpService::build()
                 .client_timeout(Seconds(1))
                 .disconnect_timeout(Seconds(1))

ntex/src/http/test.rs

@@ -4,8 +4,7 @@ use std::{convert::TryFrom, io, net, str::FromStr, sync::mpsc, thread};
 #[cfg(feature = "cookie")]
 use coo_kie::{Cookie, CookieJar};
 
-use crate::rt::System;
+use crate::{io::Io, rt::System, server::Server, service::ServiceFactory};
-use crate::server::{Server, StreamServiceFactory};
 use crate::{time::Millis, time::Seconds, util::Bytes};
 
 use super::client::error::WsClientError;
@@ -209,7 +208,11 @@ fn parts(parts: &mut Option<Inner>) -> &mut Inner {
 ///     assert!(response.status().is_success());
 /// }
 /// ```
-pub fn server<F: StreamServiceFactory>(factory: F) -> TestServer {
+pub fn server<F, R>(factory: F) -> TestServer
+where
+    F: Fn() -> R + Send + Clone + 'static,
+    R: ServiceFactory<Config = (), Request = Io>,
+{
     let (tx, rx) = mpsc::channel();
 
     // run server in separate thread
@@ -220,10 +223,10 @@ pub fn server<F: StreamServiceFactory>(factory: F) -> TestServer {
 
         sys.exec(|| {
             Server::build()
-                .listen("test", tcp, factory)?
+                .listen("test", tcp, move |_| factory())?
                 .workers(1)
                 .disable_signals()
-                .start();
+                .run();
             Ok::<_, io::Error>(())
         })?;
 

ntex/src/server/builder.rs

@@ -8,12 +8,16 @@ use futures_core::Stream;
 use log::{error, info};
 use socket2::{Domain, SockAddr, Socket, Type};
 
+use crate::io::Io;
 use crate::rt::{spawn, Signal, System};
-use crate::{time::sleep, time::Millis, util::join_all, util::PoolId};
+use crate::service::ServiceFactory;
+use crate::{time::sleep, time::Millis, util::join_all};
 
 use super::accept::{AcceptLoop, AcceptNotify, Command};
-use super::config::{ConfigWrapper, ConfiguredService, ServiceConfig, ServiceRuntime};
+use super::config::{
-use super::service::{Factory, InternalServiceFactory, StreamServiceFactory};
+    ConfigWrapper, Configuration, ConfiguredService, RuntimeConfiguration, ServiceConfig,
+};
+use super::service::{Factory, InternalServiceFactory};
 use super::socket::Listener;
 use super::worker::{self, Worker, WorkerAvailability, WorkerClient};
 use super::{Server, ServerCommand, ServerStatus, Token};
@@ -147,9 +151,9 @@ impl ServerBuilder {
     /// different module or even library.
     pub fn configure<F>(mut self, f: F) -> io::Result<ServerBuilder>
     where
-        F: Fn(&mut ServiceConfig) -> io::Result<()>,
+        F: Fn(&mut Configuration) -> io::Result<()>,
     {
-        let mut cfg = ServiceConfig::new(self.threads, self.backlog);
+        let mut cfg = Configuration::new(self.threads, self.backlog);
 
         f(&mut cfg)?;
 
@@ -172,7 +176,7 @@ impl ServerBuilder {
     /// It get executed in the worker thread.
     pub fn on_worker_start<F, R, E>(mut self, f: F) -> Self
     where
-        F: Fn(ServiceRuntime) -> R + Send + Clone + 'static,
+        F: Fn(RuntimeConfiguration) -> R + Send + Clone + 'static,
         R: Future<Output = Result<(), E>> + 'static,
         E: fmt::Display + 'static,
     {
@@ -184,27 +188,17 @@ impl ServerBuilder {
         self
     }
 
-    /// Set memory pool for name dservice.
-    ///
-    /// Use specified memory pool for memory allocations. By default P0
-    /// memory pool is used.
-    pub fn memory_pool<N: AsRef<str>>(mut self, name: N, id: PoolId) -> Self {
-        for srv in &mut self.services {
-            srv.set_memory_pool(name.as_ref(), id)
-        }
-        self
-    }
-
     /// Add new service to the server.
-    pub fn bind<F, U, N: AsRef<str>>(
+    pub fn bind<F, U, N: AsRef<str>, R>(
         mut self,
         name: N,
         addr: U,
         factory: F,
     ) -> io::Result<Self>
     where
-        F: StreamServiceFactory,
         U: net::ToSocketAddrs,
+        F: Fn(ServiceConfig) -> R + Send + Clone + 'static,
+        R: ServiceFactory<Config = (), Request = Io>,
     {
         let sockets = bind_addr(addr, self.backlog)?;
 
@@ -227,11 +221,12 @@ impl ServerBuilder {
 
     #[cfg(all(unix))]
     /// Add new unix domain service to the server.
-    pub fn bind_uds<F, U, N>(self, name: N, addr: U, factory: F) -> io::Result<Self>
+    pub fn bind_uds<F, U, N, R>(self, name: N, addr: U, factory: F) -> io::Result<Self>
     where
-        F: StreamServiceFactory,
         N: AsRef<str>,
         U: AsRef<std::path::Path>,
+        F: Fn(ServiceConfig) -> R + Send + Clone + 'static,
+        R: ServiceFactory<Config = (), Request = Io>,
     {
         use std::os::unix::net::UnixListener;
 
@@ -252,14 +247,15 @@ impl ServerBuilder {
     /// 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.
-    pub fn listen_uds<F, N: AsRef<str>>(
+    pub fn listen_uds<F, N: AsRef<str>, R>(
         mut self,
         name: N,
         lst: std::os::unix::net::UnixListener,
         factory: F,
     ) -> io::Result<Self>
     where
-        F: StreamServiceFactory,
+        F: Fn(ServiceConfig) -> R + Send + Clone + 'static,
+        R: ServiceFactory<Config = (), Request = Io>,
     {
         use std::net::{IpAddr, Ipv4Addr, SocketAddr};
         let token = self.token.next();
@@ -276,14 +272,15 @@ impl ServerBuilder {
     }
 
     /// Add new service to the server.
-    pub fn listen<F, N: AsRef<str>>(
+    pub fn listen<F, N: AsRef<str>, R>(
         mut self,
         name: N,
         lst: net::TcpListener,
         factory: F,
     ) -> io::Result<Self>
     where
-        F: StreamServiceFactory,
+        F: Fn(ServiceConfig) -> R + Send + Clone + 'static,
+        R: ServiceFactory<Config = (), Request = Io>,
     {
         let token = self.token.next();
         self.services.push(Factory::create(
@@ -297,11 +294,6 @@ impl ServerBuilder {
         Ok(self)
     }
 
-    #[doc(hidden)]
-    pub fn start(self) -> Server {
-        self.run()
-    }
-
     /// Starts processing incoming connections and return server controller.
     pub fn run(mut self) -> Server {
         if self.sockets.is_empty() {

ntex/src/server/config.rs

@@ -1,6 +1,6 @@
 use std::{
-    cell::RefCell, fmt, future::Future, io, marker::PhantomData, mem, net, pin::Pin,
+    cell::Cell, cell::RefCell, fmt, future::Future, io, marker::PhantomData, mem, net,
-    rc::Rc,
+    pin::Pin, rc::Rc,
 };
 
 use log::error;
@@ -14,7 +14,32 @@ use super::service::{
 };
 use super::Token;
 
-pub struct ServiceConfig {
+#[derive(Clone)]
+pub struct ServiceConfig(pub(super) Rc<InnerServiceConfig>);
+
+pub(super) struct InnerServiceConfig {
+    pub(super) pool: Cell<PoolId>,
+}
+
+impl Default for ServiceConfig {
+    fn default() -> Self {
+        Self(Rc::new(InnerServiceConfig {
+            pool: Cell::new(PoolId::DEFAULT),
+        }))
+    }
+}
+
+impl ServiceConfig {
+    /// Set memory pool for the service.
+    ///
+    /// Use specified memory pool for memory allocations.
+    pub fn memory_pool(&self, id: PoolId) -> &Self {
+        self.0.pool.set(id);
+        self
+    }
+}
+
+pub struct Configuration {
     pub(super) services: Vec<(String, net::TcpListener)>,
     pub(super) apply: Box<dyn ServiceRuntimeConfiguration + Send>,
     pub(super) threads: usize,
@@ -22,9 +47,9 @@ pub struct ServiceConfig {
     applied: bool,
 }
 
-impl ServiceConfig {
+impl Configuration {
-    pub(super) fn new(threads: usize, backlog: i32) -> ServiceConfig {
+    pub(super) fn new(threads: usize, backlog: i32) -> Self {
-        ServiceConfig {
+        Configuration {
             threads,
             backlog,
             services: Vec::new(),
@@ -78,7 +103,7 @@ impl ServiceConfig {
     /// It get executed in the worker thread.
     pub fn on_worker_start<F, R, E>(&mut self, f: F) -> io::Result<()>
     where
-        F: Fn(ServiceRuntime) -> R + Send + Clone + 'static,
+        F: Fn(RuntimeConfiguration) -> R + Send + Clone + 'static,
         R: Future<Output = Result<(), E>> + 'static,
         E: fmt::Display + 'static,
     {
@@ -126,15 +151,13 @@ impl InternalServiceFactory for ConfiguredService {
         })
     }
 
-    fn set_memory_pool(&self, _: &str, _: PoolId) {}
-
     fn create(
         &self,
     ) -> Pin<Box<dyn Future<Output = Result<Vec<(Token, BoxedServerService)>, ()>>>>
     {
         // configure services
-        let rt = ServiceRuntime::new(self.topics.clone());
+        let rt = RuntimeConfiguration::new(self.topics.clone());
-        let cfg_fut = self.rt.configure(ServiceRuntime(rt.0.clone()));
+        let cfg_fut = self.rt.configure(RuntimeConfiguration(rt.0.clone()));
         let mut names = self.names.clone();
         let tokens = self.services.clone();
 
@@ -185,7 +208,7 @@ pub(super) trait ServiceRuntimeConfiguration {
 
     fn configure(
         &self,
-        rt: ServiceRuntime,
+        rt: RuntimeConfiguration,
     ) -> Pin<Box<dyn Future<Output = Result<(), ()>>>>;
 }
 
@@ -199,7 +222,7 @@ unsafe impl<F: Send, R, E> Send for ConfigWrapper<F, R, E> {}
 
 impl<F, R, E> ServiceRuntimeConfiguration for ConfigWrapper<F, R, E>
 where
-    F: Fn(ServiceRuntime) -> R + Send + Clone + 'static,
+    F: Fn(RuntimeConfiguration) -> R + Send + Clone + 'static,
     R: Future<Output = Result<(), E>> + 'static,
     E: fmt::Display + 'static,
 {
@@ -212,7 +235,7 @@ where
 
     fn configure(
         &self,
-        rt: ServiceRuntime,
+        rt: RuntimeConfiguration,
     ) -> Pin<Box<dyn Future<Output = Result<(), ()>>>> {
         let f = self.f.clone();
         Box::pin(async move {
@@ -227,7 +250,7 @@ fn not_configured() {
     error!("Service is not configured");
 }
 
-pub struct ServiceRuntime(Rc<RefCell<ServiceRuntimeInner>>);
+pub struct RuntimeConfiguration(Rc<RefCell<ServiceRuntimeInner>>);
 
 struct ServiceRuntimeInner {
     names: HashMap<String, Token>,
@@ -235,9 +258,9 @@ struct ServiceRuntimeInner {
     onstart: Vec<Pin<Box<dyn Future<Output = ()>>>>,
 }
 
-impl ServiceRuntime {
+impl RuntimeConfiguration {
     fn new(names: HashMap<String, Token>) -> Self {
-        ServiceRuntime(Rc::new(RefCell::new(ServiceRuntimeInner {
+        RuntimeConfiguration(Rc::new(RefCell::new(ServiceRuntimeInner {
             names,
             services: HashMap::default(),
             onstart: Vec::new(),

ntex/src/server/mod.rs

@@ -14,8 +14,7 @@ mod worker;
 
 pub(crate) use self::builder::create_tcp_listener;
 pub use self::builder::ServerBuilder;
-pub use self::config::{ServiceConfig, ServiceRuntime};
+pub use self::config::{Configuration, RuntimeConfiguration, ServiceConfig};
-pub use self::service::StreamServiceFactory;
 pub use self::test::{build_test_server, test_server, TestServer};
 
 #[non_exhaustive]

ntex/src/server/service.rs

@@ -1,7 +1,5 @@
 use std::convert::TryInto;
-use std::{
+use std::{future::Future, net::SocketAddr, pin::Pin, task::Context, task::Poll};
-    cell::Cell, future::Future, net::SocketAddr, pin::Pin, task::Context, task::Poll,
-};
 
 use log::error;
 
@@ -10,7 +8,7 @@ use crate::service::{Service, ServiceFactory};
 use crate::util::{counter::CounterGuard, Pool, PoolId, Ready};
 use crate::{rt::spawn, time::Millis};
 
-use super::{socket::Stream, Token};
+use super::{socket::Stream, ServiceConfig, Token};
 
 /// Server message
 pub(super) enum ServerMessage {
@@ -22,10 +20,10 @@ pub(super) enum ServerMessage {
     ForceShutdown,
 }
 
-pub trait StreamServiceFactory: Send + Clone + 'static {
+pub(super) trait StreamServiceFactory: Send + Clone + 'static {
     type Factory: ServiceFactory<Config = (), Request = Io>;
 
-    fn create(&self) -> Self::Factory;
+    fn create(&self, _: ServiceConfig) -> Self::Factory;
 }
 
 pub(super) trait InternalServiceFactory: Send {
@@ -33,8 +31,6 @@ pub(super) trait InternalServiceFactory: Send {
 
     fn clone_factory(&self) -> Box<dyn InternalServiceFactory>;
 
-    fn set_memory_pool(&self, name: &str, pool: PoolId);
-
     fn create(
         &self,
     ) -> Pin<Box<dyn Future<Output = Result<Vec<(Token, BoxedServerService)>, ()>>>>;
@@ -120,7 +116,6 @@ pub(super) struct Factory<F: StreamServiceFactory> {
     inner: F,
     token: Token,
     addr: SocketAddr,
-    pool: Cell<PoolId>,
 }
 
 impl<F> Factory<F>
@@ -138,7 +133,6 @@ where
             token,
             inner,
             addr,
-            pool: Cell::new(PoolId::P0),
         })
     }
 }
@@ -157,29 +151,22 @@ where
             inner: self.inner.clone(),
             token: self.token,
             addr: self.addr,
-            pool: self.pool.clone(),
         })
     }
 
-    fn set_memory_pool(&self, name: &str, pool: PoolId) {
-        if self.name == name {
-            self.pool.set(pool)
-        }
-    }
-
     fn create(
         &self,
     ) -> Pin<Box<dyn Future<Output = Result<Vec<(Token, BoxedServerService)>, ()>>>>
     {
         let token = self.token;
-        let pool = self.pool.get();
+        let cfg = ServiceConfig::default();
-        let fut = self.inner.create().new_service(());
+        let fut = self.inner.create(cfg.clone()).new_service(());
 
         Box::pin(async move {
             match fut.await {
                 Ok(inner) => {
                     let service: BoxedServerService =
-                        Box::new(StreamService::new(inner, pool));
+                        Box::new(StreamService::new(inner, cfg.0.pool.get()));
                     Ok(vec![(token, service)])
                 }
                 Err(_) => Err(()),
@@ -197,10 +184,6 @@ impl InternalServiceFactory for Box<dyn InternalServiceFactory> {
         self.as_ref().clone_factory()
     }
 
-    fn set_memory_pool(&self, name: &str, pool: PoolId) {
-        self.as_ref().set_memory_pool(name, pool)
-    }
-
     fn create(
         &self,
     ) -> Pin<Box<dyn Future<Output = Result<Vec<(Token, BoxedServerService)>, ()>>>>
@@ -211,13 +194,13 @@ impl InternalServiceFactory for Box<dyn InternalServiceFactory> {
 
 impl<F, T> StreamServiceFactory for F
 where
-    F: Fn() -> T + Send + Clone + 'static,
+    F: Fn(ServiceConfig) -> T + Send + Clone + 'static,
     T: ServiceFactory<Config = (), Request = Io>,
 {
     type Factory = T;
 
     #[inline]
-    fn create(&self) -> T {
+    fn create(&self, cfg: ServiceConfig) -> T {
-        (self)()
+        (self)(cfg)
     }
 }

ntex/src/server/test.rs

@@ -3,9 +3,9 @@ use std::{io, net, sync::mpsc, thread};
 
 use socket2::{Domain, SockAddr, Socket, Type};
 
-use crate::io::Io;
 use crate::rt::{tcp_connect, System};
-use crate::server::{Server, ServerBuilder, StreamServiceFactory};
+use crate::server::{Server, ServerBuilder};
+use crate::{io::Io, service::ServiceFactory};
 
 /// Start test server
 ///
@@ -38,7 +38,11 @@ use crate::server::{Server, ServerBuilder, StreamServiceFactory};
 ///     assert!(response.status().is_success());
 /// }
 /// ```
-pub fn test_server<F: StreamServiceFactory>(factory: F) -> TestServer {
+pub fn test_server<F, R>(factory: F) -> TestServer
+where
+    F: Fn() -> R + Send + Clone + 'static,
+    R: ServiceFactory<Config = (), Request = Io>,
+{
     let (tx, rx) = mpsc::channel();
 
     // run server in separate thread
@@ -49,10 +53,10 @@ pub fn test_server<F: StreamServiceFactory>(factory: F) -> TestServer {
 
         sys.exec(|| {
             Server::build()
-                .listen("test", tcp, factory)?
+                .listen("test", tcp, move |_| factory())?
                 .workers(1)
                 .disable_signals()
-                .start();
+                .run();
             Ok::<_, io::Error>(())
         })?;
 
@@ -77,10 +81,7 @@ where
         let sys = System::new("ntex-test-server");
 
         sys.exec(|| {
-            factory(Server::build())
+            factory(Server::build()).workers(1).disable_signals().run();
-                .workers(1)
-                .disable_signals()
-                .start();
         });
 
         tx.send(System::current()).unwrap();

ntex/src/server/worker.rs

@@ -592,7 +592,7 @@ mod tests {
             vec![Factory::create(
                 "test".to_string(),
                 Token(0),
-                move || f.clone(),
+                move |_| f.clone(),
                 "127.0.0.1:8080".parse().unwrap(),
             )],
             avail.clone(),
@@ -664,7 +664,7 @@ mod tests {
             vec![Factory::create(
                 "test".to_string(),
                 Token(0),
-                move || f.clone(),
+                move |_| f.clone(),
                 "127.0.0.1:8080".parse().unwrap(),
             )],
             avail.clone(),

ntex/src/web/app.rs

@@ -468,7 +468,7 @@ where
     ///
     /// #[ntex::main]
     /// async fn main() -> std::io::Result<()> {
-    ///     server::build().bind("http", "127.0.0.1:0", ||
+    ///     server::build().bind("http", "127.0.0.1:0", |_|
     ///         http::HttpService::build().finish(
     ///             web::App::new()
     ///                 .route("/index.html", web::get().to(|| async { "hello_world" }))
@@ -498,7 +498,7 @@ where
     ///
     /// #[ntex::main]
     /// async fn main() -> std::io::Result<()> {
-    ///     server::build().bind("http", "127.0.0.1:0", ||
+    ///     server::build().bind("http", "127.0.0.1:0", |_|
     ///         http::HttpService::build().finish(
     ///             web::App::new()
     ///                 .route("/index.html", web::get().to(|| async { "hello_world" }))

ntex/src/web/server.rs

@@ -9,8 +9,8 @@ use crate::http::{
     body::MessageBody, HttpService, KeepAlive, Request, Response, ResponseError,
 };
 use crate::server::{Server, ServerBuilder};
-use crate::time::Seconds;
 use crate::{service::map_config, IntoServiceFactory, Service, ServiceFactory};
+use crate::{time::Seconds, util::PoolId};
 
 use super::config::AppConfig;
 
@@ -20,6 +20,7 @@ struct Config {
     client_timeout: Seconds,
     client_disconnect: Seconds,
     handshake_timeout: Seconds,
+    pool: PoolId,
 }
 
 /// An HTTP Server.
@@ -79,6 +80,7 @@ where
                 client_timeout: Seconds(5),
                 client_disconnect: Seconds(5),
                 handshake_timeout: Seconds(5),
+                pool: PoolId::P0,
             })),
             backlog: 1024,
             builder: ServerBuilder::default(),
@@ -219,6 +221,14 @@ where
         self
     }
 
+    /// Set memory pool.
+    ///
+    /// Use specified memory pool for memory allocations.
+    pub fn memory_pool(&mut self, id: PoolId) -> &mut Self {
+        self.config.lock().unwrap().pool = id;
+        self
+    }
+
     /// Use listener for accepting incoming connection requests
     ///
     /// HttpServer does not change any configuration for TcpListener,
@@ -231,13 +241,14 @@ where
         self.builder = self.builder.listen(
             format!("ntex-web-service-{}", addr),
             lst,
-            move || {
+            move |r| {
                 let c = cfg.lock().unwrap();
                 let cfg = AppConfig::new(
                     false,
                     addr,
                     c.host.clone().unwrap_or_else(|| format!("{}", addr)),
                 );
+                r.memory_pool(c.pool);
 
                 HttpService::build()
                     .keep_alive(c.keep_alive)
@@ -274,13 +285,15 @@ where
         self.builder = self.builder.listen(
             format!("ntex-web-service-{}", addr),
             lst,
-            move || {
+            move |r| {
                 let c = cfg.lock().unwrap();
                 let cfg = AppConfig::new(
                     true,
                     addr,
                     c.host.clone().unwrap_or_else(|| format!("{}", addr)),
                 );
+                r.memory_pool(c.pool);
+
                 HttpService::build()
                     .keep_alive(c.keep_alive)
                     .client_timeout(c.client_timeout)
@@ -318,13 +331,15 @@ where
         self.builder = self.builder.listen(
             format!("ntex-web-rustls-service-{}", addr),
             lst,
-            move || {
+            move |r| {
                 let c = cfg.lock().unwrap();
                 let cfg = AppConfig::new(
                     true,
                     addr,
                     c.host.clone().unwrap_or_else(|| format!("{}", addr)),
                 );
+                r.memory_pool(c.pool);
+
                 HttpService::build()
                     .keep_alive(c.keep_alive)
                     .client_timeout(c.client_timeout)
@@ -436,13 +451,15 @@ where
 
         let addr = format!("ntex-web-service-{:?}", lst.local_addr()?);
 
-        self.builder = self.builder.listen_uds(addr, lst, move || {
+        self.builder = self.builder.listen_uds(addr, lst, move |r| {
             let c = cfg.lock().unwrap();
             let config = AppConfig::new(
                 false,
                 socket_addr,
                 c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
             );
+            r.memory_pool(c.pool);
+
             HttpService::build()
                 .keep_alive(c.keep_alive)
                 .client_timeout(c.client_timeout)
@@ -469,13 +486,15 @@ where
         self.builder = self.builder.bind_uds(
             format!("ntex-web-service-{:?}", addr.as_ref()),
             addr,
-            move || {
+            move |r| {
                 let c = cfg.lock().unwrap();
                 let config = AppConfig::new(
                     false,
                     socket_addr,
                     c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
                 );
+                r.memory_pool(c.pool);
+
                 HttpService::build()
                     .keep_alive(c.keep_alive)
                     .client_timeout(c.client_timeout)
@@ -520,7 +539,7 @@ where
     /// }
     /// ```
     pub fn run(self) -> Server {
-        self.builder.start()
+        self.builder.run()
     }
 }
 

ntex/src/web/test.rs

@@ -624,21 +624,21 @@ where
 
             match cfg.stream {
                 StreamType::Tcp => match cfg.tp {
-                    HttpVer::Http1 => builder.listen("test", tcp, move || {
+                    HttpVer::Http1 => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(false, local_addr, format!("{}", local_addr));
                         HttpService::build()
                             .client_timeout(ctimeout)
                             .h1(map_config(factory(), move |_| cfg.clone()))
                     }),
-                    HttpVer::Http2 => builder.listen("test", tcp, move || {
+                    HttpVer::Http2 => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(false, local_addr, format!("{}", local_addr));
                         HttpService::build()
                             .client_timeout(ctimeout)
                             .h2(map_config(factory(), move |_| cfg.clone()))
                     }),
-                    HttpVer::Both => builder.listen("test", tcp, move || {
+                    HttpVer::Both => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(false, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -648,7 +648,7 @@ where
                 },
                 #[cfg(feature = "openssl")]
                 StreamType::Openssl(acceptor) => match cfg.tp {
-                    HttpVer::Http1 => builder.listen("test", tcp, move || {
+                    HttpVer::Http1 => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(true, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -656,7 +656,7 @@ where
                             .h1(map_config(factory(), move |_| cfg.clone()))
                             .openssl(acceptor.clone())
                     }),
-                    HttpVer::Http2 => builder.listen("test", tcp, move || {
+                    HttpVer::Http2 => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(true, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -664,7 +664,7 @@ where
                             .h2(map_config(factory(), move |_| cfg.clone()))
                             .openssl(acceptor.clone())
                     }),
-                    HttpVer::Both => builder.listen("test", tcp, move || {
+                    HttpVer::Both => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(true, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -675,7 +675,7 @@ where
                 },
                 #[cfg(feature = "rustls")]
                 StreamType::Rustls(config) => match cfg.tp {
-                    HttpVer::Http1 => builder.listen("test", tcp, move || {
+                    HttpVer::Http1 => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(true, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -683,7 +683,7 @@ where
                             .h1(map_config(factory(), move |_| cfg.clone()))
                             .rustls(config.clone())
                     }),
-                    HttpVer::Http2 => builder.listen("test", tcp, move || {
+                    HttpVer::Http2 => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(true, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -691,7 +691,7 @@ where
                             .h2(map_config(factory(), move |_| cfg.clone()))
                             .rustls(config.clone())
                     }),
-                    HttpVer::Both => builder.listen("test", tcp, move || {
+                    HttpVer::Both => builder.listen("test", tcp, move |_| {
                         let cfg =
                             AppConfig::new(true, local_addr, format!("{}", local_addr));
                         HttpService::build()
@@ -702,7 +702,7 @@ where
                 },
             }
             .unwrap()
-            .start()
+            .run()
         });
 
         tx.send((System::current(), srv, local_addr)).unwrap();

ntex/tests/server.rs

@@ -21,9 +21,9 @@ fn test_bind() {
             Server::build()
                 .workers(1)
                 .disable_signals()
-                .bind("test", addr, move || fn_service(|_| ok::<_, ()>(())))
+                .bind("test", addr, move |_| fn_service(|_| ok::<_, ()>(())))
                 .unwrap()
-                .start()
+                .run()
         });
         let _ = tx.send((srv, ntex::rt::System::current()));
         let _ = sys.run();
@@ -48,9 +48,9 @@ fn test_listen() {
             Server::build()
                 .disable_signals()
                 .workers(1)
-                .listen("test", lst, move || fn_service(|_| ok::<_, ()>(())))
+                .listen("test", lst, move |_| fn_service(|_| ok::<_, ()>(())))
                 .unwrap()
-                .start()
+                .run()
         });
         let _ = tx.send(ntex::rt::System::current());
         let _ = sys.run();
@@ -65,7 +65,7 @@ fn test_listen() {
 
 #[test]
 #[cfg(unix)]
-fn test_start() {
+fn test_run() {
     let addr = TestServer::unused_addr();
     let (tx, rx) = mpsc::channel();
 
@@ -75,7 +75,7 @@ fn test_start() {
             Server::build()
                 .backlog(100)
                 .disable_signals()
-                .bind("test", addr, move || {
+                .bind("test", addr, move |_| {
                     fn_service(|io: Io| async move {
                         io.send(Bytes::from_static(b"test"), &BytesCodec)
                             .await
@@ -84,7 +84,7 @@ fn test_start() {
                     })
                 })
                 .unwrap()
-                .start()
+                .run()
         });
 
         let _ = tx.send((srv, ntex::rt::System::current()));
@@ -170,7 +170,7 @@ fn test_on_worker_start() {
                     Ready::Ok::<_, io::Error>(())
                 })
                 .workers(1)
-                .start()
+                .run()
         });
         let _ = tx.send((srv, ntex::rt::System::current()));
         let _ = sys.run();
@@ -203,7 +203,7 @@ fn test_panic_in_worker() {
             Server::build()
                 .workers(1)
                 .disable_signals()
-                .bind("test", addr, move || {
+                .bind("test", addr, move |_| {
                     let counter = counter.clone();
                     fn_service(move |_| {
                         counter.fetch_add(1, Relaxed);
@@ -212,7 +212,7 @@ fn test_panic_in_worker() {
                     })
                 })
                 .unwrap()
-                .start()
+                .run()
         });
         let _ = tx.send((srv.clone(), ntex::rt::System::current()));
         sys.exec(move || ntex::rt::spawn(srv.map(|_| ())));