convert Server::bind to accept a normal service factory

Author: robjtede

actix-server/CHANGES.md

@@ -1,11 +1,13 @@
 # Changes
 
 ## Unreleased - 2021-xx-xx
-* Rename `Server` to `ServerHandle`. [#???]
-* Rename `ServerBuilder::{maxconn => max_concurrent_connections}`. [#???]
+* Rename `Server` to `ServerHandle`. [#403]
+* Rename `ServerBuilder::{maxconn => max_concurrent_connections}`. [#403]
+* Remove wrapper `service::ServiceFactory` trait. [#403]
+* `Server::bind` and related methods now take a regular `ServiceFactory` (from actix-service crate). [#403]
 * Minimum supported Rust version (MSRV) is now 1.52.
 
-[#???]: https://github.com/actix/actix-net/pull/???
+[#403]: https://github.com/actix/actix-net/pull/403
 
 
 ## 2.0.0-beta.6 - 2021-10-11

actix-server/Cargo.toml

@@ -38,4 +38,4 @@ actix-rt = "2.0.0"
 bytes = "1"
 env_logger = "0.9"
 futures-util = { version = "0.3.7", default-features = false, features = ["sink"] }
-tokio = { version = "1.5.1", features = ["io-util"] }
+tokio = { version = "1.5.1", features = ["io-util", "rt-multi-thread", "macros"] }

actix-server/examples/startup-fail.rs

@@ -0,0 +1,25 @@
+use std::io;
+
+use actix_rt::net::TcpStream;
+use actix_server::Server;
+use actix_service::fn_service;
+use log::info;
+
+#[actix_rt::main]
+async fn main() -> io::Result<()> {
+    env_logger::Builder::from_env(env_logger::Env::new().default_filter_or("trace,mio=info"))
+        .init();
+
+    let addr = ("127.0.0.1", 8080);
+    info!("starting server on port: {}", &addr.0);
+
+    Server::build()
+        .bind(
+            "startup-fail",
+            addr,
+            fn_service(move |mut _stream: TcpStream| async move { Ok::<u32, u32>(42) }),
+        )?
+        .workers(2)
+        .run()
+        .await
+}

actix-server/examples/tcp-echo.rs

@@ -21,7 +21,6 @@ use actix_rt::net::TcpStream;
 use actix_server::Server;
 use actix_service::{fn_service, ServiceFactoryExt as _};
 use bytes::BytesMut;
-use futures_util::future::ok;
 use log::{error, info};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -39,11 +38,11 @@ async fn main() -> io::Result<()> {
     // logical CPU cores as the worker count. For this reason, the closure passed to bind needs
     // to return a service *factory*; so it can be created once per worker.
     Server::build()
-        .bind("echo", addr, move || {
+        .bind("echo", addr, {
             let count = Arc::clone(&count);
             let num2 = Arc::clone(&count);
 
-            fn_service(move |mut stream: TcpStream| {
+            let svc = fn_service(move |mut stream: TcpStream| {
                 let count = Arc::clone(&count);
 
                 async move {
@@ -78,11 +77,15 @@ async fn main() -> io::Result<()> {
                 }
             })
             .map_err(|err| error!("Service Error: {:?}", err))
-            .and_then(move |(_, size)| {
+            .and_then_send(move |(_, size)| {
                 let num = num2.load(Ordering::SeqCst);
                 info!("[{}] total bytes read: {}", num, size);
-                ok(size)
-            })
+                async move { Ok(size) }
+            });
+
+            let svc2 = svc.clone();
+
+            svc2
         })?
         .workers(1)
         .run()

actix-server/src/builder.rs

@@ -1,4 +1,5 @@
 use std::{
+    fmt,
     future::Future,
     io, mem,
     pin::Pin,
@@ -7,32 +8,25 @@ use std::{
 };
 
 use actix_rt::{self as rt, net::TcpStream, time::sleep, System};
+use actix_service::ServiceFactory;
 use log::{error, info};
 use tokio::sync::{
     mpsc::{unbounded_channel, UnboundedReceiver},
     oneshot,
 };
 
-use crate::accept::AcceptLoop;
-use crate::join_all;
-use crate::server::{ServerCommand, ServerHandle};
-use crate::service::{InternalServiceFactory, ServiceFactory, StreamNewService};
-use crate::signals::{Signal, Signals};
-use crate::socket::{MioListener, StdSocketAddr, StdTcpListener, ToSocketAddrs};
-use crate::socket::{MioTcpListener, MioTcpSocket};
-use crate::waker_queue::{WakerInterest, WakerQueue};
-use crate::worker::{ServerWorker, ServerWorkerConfig, WorkerHandleAccept, WorkerHandleServer};
-
-#[derive(Debug)]
-#[non_exhaustive]
-pub struct Server;
-
-impl Server {
-    /// Start server building process.
-    pub fn build() -> ServerBuilder {
-        ServerBuilder::default()
-    }
-}
+use crate::{
+    accept::AcceptLoop,
+    join_all,
+    server::{ServerCommand, ServerHandle},
+    service::{InternalServiceFactory, StreamNewService},
+    signals::{Signal, Signals},
+    socket::{
+        MioListener, MioTcpListener, MioTcpSocket, StdSocketAddr, StdTcpListener, ToSocketAddrs,
+    },
+    waker_queue::{WakerInterest, WakerQueue},
+    worker::{ServerWorker, ServerWorkerConfig, WorkerHandleAccept, WorkerHandleServer},
+};
 
 /// Server builder
 pub struct ServerBuilder {
@@ -169,38 +163,48 @@ impl ServerBuilder {
     /// Binds to all network interface addresses that resolve from the `addr` argument.
     /// Eg. using `localhost` might bind to both IPv4 and IPv6 addresses. Bind to multiple distinct
     /// interfaces at the same time by passing a list of socket addresses.
-    pub fn bind<F, U, N: AsRef<str>>(mut self, name: N, addr: U, factory: F) -> io::Result<Self>
+    pub fn bind<F, U, InitErr>(
+        mut self,
+        name: impl AsRef<str>,
+        addr: U,
+        factory: F,
+    ) -> io::Result<Self>
     where
-        F: ServiceFactory<TcpStream>,
+        F: ServiceFactory<TcpStream, Config = (), InitError = InitErr> + Send + Clone + 'static,
+        InitErr: fmt::Debug + Send + 'static,
         U: ToSocketAddrs,
     {
         let sockets = bind_addr(addr, self.backlog)?;
 
         for lst in sockets {
             let token = self.next_token();
+
             self.services.push(StreamNewService::create(
                 name.as_ref().to_string(),
                 token,
                 factory.clone(),
                 lst.local_addr()?,
             ));
+
             self.sockets
                 .push((token, name.as_ref().to_string(), MioListener::Tcp(lst)));
         }
+
         Ok(self)
     }
 
     /// Bind server to existing TCP listener.
     ///
     /// Useful when running as a systemd service and a socket FD can be passed to the process.
-    pub fn listen<F, N: AsRef<str>>(
+    pub fn listen<F, InitErr>(
         mut self,
-        name: N,
+        name: impl AsRef<str>,
         lst: StdTcpListener,
         factory: F,
     ) -> io::Result<Self>
     where
-        F: ServiceFactory<TcpStream>,
+        F: ServiceFactory<TcpStream, Config = (), InitError = InitErr> + Send + Clone + 'static,
+        InitErr: fmt::Debug + Send + 'static,
     {
         lst.set_nonblocking(true)?;
 
@@ -259,7 +263,7 @@ impl ServerBuilder {
                 Signals::start(self.server.clone());
             }
 
-            // start http server actor
+            // start http server
             let server = self.server.clone();
             rt::spawn(self);
             server
@@ -402,11 +406,19 @@ impl ServerBuilder {
 #[cfg(unix)]
 impl ServerBuilder {
     /// 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, InitErr>(
+        self,
+        name: impl AsRef<str>,
+        addr: U,
+        factory: F,
+    ) -> io::Result<Self>
     where
-        F: ServiceFactory<actix_rt::net::UnixStream>,
-        N: AsRef<str>,
+        F: ServiceFactory<actix_rt::net::UnixStream, Config = (), InitError = InitErr>
+            + Send
+            + Clone
+            + 'static,
         U: AsRef<std::path::Path>,
+        InitErr: fmt::Debug + Send + 'static,
     {
         // The path must not exist when we try to bind.
         // Try to remove it to avoid bind error.
@@ -424,14 +436,18 @@ impl ServerBuilder {
     /// Add new unix domain service to the server.
     ///
     /// Useful when running as a systemd service and a socket FD can be passed to the process.
-    pub fn listen_uds<F, N: AsRef<str>>(
+    pub fn listen_uds<F, InitErr>(
         mut self,
-        name: N,
+        name: impl AsRef<str>,
         lst: crate::socket::StdUnixListener,
         factory: F,
     ) -> io::Result<Self>
     where
-        F: ServiceFactory<actix_rt::net::UnixStream>,
+        F: ServiceFactory<actix_rt::net::UnixStream, Config = (), InitError = InitErr>
+            + Send
+            + Clone
+            + 'static,
+        InitErr: fmt::Debug + Send + 'static,
     {
         use std::net::{IpAddr, Ipv4Addr};
 

actix-server/src/lib.rs

@@ -14,9 +14,8 @@ mod test_server;
 mod waker_queue;
 mod worker;
 
-pub use self::builder::{Server, ServerBuilder};
-pub use self::server::{ServerHandle};
-pub use self::service::ServiceFactory;
+pub use self::builder::ServerBuilder;
+pub use self::server::{Server, ServerHandle};
 pub use self::test_server::TestServer;
 
 #[doc(hidden)]

actix-server/src/server.rs

@@ -6,7 +6,18 @@ use std::task::{Context, Poll};
 use tokio::sync::mpsc::UnboundedSender;
 use tokio::sync::oneshot;
 
-use crate::signals::Signal;
+use crate::{signals::Signal, ServerBuilder};
+
+#[derive(Debug)]
+#[non_exhaustive]
+pub struct Server;
+
+impl Server {
+    /// Start server building process.
+    pub fn build() -> ServerBuilder {
+        ServerBuilder::default()
+    }
+}
 
 #[derive(Debug)]
 pub(crate) enum ServerCommand {

actix-server/src/service.rs

@@ -1,20 +1,19 @@
-use std::marker::PhantomData;
-use std::net::SocketAddr;
-use std::task::{Context, Poll};
-
-use actix_service::{Service, ServiceFactory as BaseServiceFactory};
+use std::{
+    fmt,
+    marker::PhantomData,
+    net::SocketAddr,
+    task::{Context, Poll},
+};
+
+use actix_service::{Service, ServiceFactory};
 use actix_utils::future::{ready, Ready};
 use futures_core::future::LocalBoxFuture;
 use log::error;
 
-use crate::socket::{FromStream, MioStream};
-use crate::worker::WorkerCounterGuard;
-
-pub trait ServiceFactory<Stream: FromStream>: Send + Clone + 'static {
-    type Factory: BaseServiceFactory<Stream, Config = ()>;
-
-    fn create(&self) -> Self::Factory;
-}
+use crate::{
+    socket::{FromStream, MioStream},
+    worker::WorkerCounterGuard,
+};
 
 pub(crate) trait InternalServiceFactory: Send {
     fn name(&self, token: usize) -> &str;
@@ -80,17 +79,18 @@ where
     }
 }
 
-pub(crate) struct StreamNewService<F: ServiceFactory<Io>, Io: FromStream> {
+pub(crate) struct StreamNewService<F, Io, InitErr> {
     name: String,
     inner: F,
     token: usize,
     addr: SocketAddr,
-    _t: PhantomData<Io>,
+    _t: PhantomData<(Io, InitErr)>,
 }
 
-impl<F, Io> StreamNewService<F, Io>
+impl<F, Io, InitErr> StreamNewService<F, Io, InitErr>
 where
-    F: ServiceFactory<Io>,
+    F: ServiceFactory<Io, Config = (), InitError = InitErr> + Send + Clone + 'static,
+    InitErr: fmt::Debug + Send + 'static,
     Io: FromStream + Send + 'static,
 {
     pub(crate) fn create(
@@ -109,9 +109,10 @@ where
     }
 }
 
-impl<F, Io> InternalServiceFactory for StreamNewService<F, Io>
+impl<F, Io, InitErr> InternalServiceFactory for StreamNewService<F, Io, InitErr>
 where
-    F: ServiceFactory<Io>,
+    F: ServiceFactory<Io, Config = (), InitError = InitErr> + Send + Clone + 'static,
+    InitErr: fmt::Debug + Send + 'static,
     Io: FromStream + Send + 'static,
 {
     fn name(&self, _: usize) -> &str {
@@ -130,28 +131,18 @@ where
 
     fn create(&self) -> LocalBoxFuture<'static, Result<(usize, BoxedServerService), ()>> {
         let token = self.token;
-        let fut = self.inner.create().new_service(());
+        let fut = self.inner.new_service(());
         Box::pin(async move {
             match fut.await {
                 Ok(inner) => {
                     let service = Box::new(StreamService::new(inner)) as _;
                     Ok((token, service))
                 }
-                Err(_) => Err(()),
+                Err(err) => {
+                    error!("{:?}", err);
+                    Err(())
+                }
             }
         })
     }
 }
-
-impl<F, T, I> ServiceFactory<I> for F
-where
-    F: Fn() -> T + Send + Clone + 'static,
-    T: BaseServiceFactory<I, Config = ()>,
-    I: FromStream,
-{
-    type Factory = T;
-
-    fn create(&self) -> T {
-        (self)()
-    }
-}