use std::{
any::Any,
cmp, fmt, io,
marker::PhantomData,
net,
sync::{Arc, Mutex},
};
use actix_http::{body::MessageBody, Extensions, HttpService, KeepAlive, Request, Response};
use actix_server::{Server, ServerBuilder};
use actix_service::{
map_config, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _,
};
#[cfg(feature = "openssl")]
use actix_tls::accept::openssl::reexports::{AlpnError, SslAcceptor, SslAcceptorBuilder};
#[cfg(feature = "rustls")]
use actix_tls::accept::rustls::reexports::ServerConfig as RustlsServerConfig;
use crate::{config::AppConfig, Error};
struct Socket {
scheme: &'static str,
addr: net::SocketAddr,
}
struct Config {
host: Option<String>,
keep_alive: KeepAlive,
client_timeout: u64,
client_shutdown: u64,
}
pub struct HttpServer<F, I, S, B>
where
F: Fn() -> I + Send + Clone + 'static,
I: IntoServiceFactory<S, Request>,
S: ServiceFactory<Request, Config = AppConfig>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
{
pub(super) factory: F,
config: Arc<Mutex<Config>>,
backlog: u32,
sockets: Vec<Socket>,
builder: ServerBuilder,
on_connect_fn: Option<Arc<dyn Fn(&dyn Any, &mut Extensions) + Send + Sync>>,
_phantom: PhantomData<(S, B)>,
}
impl<F, I, S, B> HttpServer<F, I, S, B>
where
F: Fn() -> I + Send + Clone + 'static,
I: IntoServiceFactory<S, Request>,
S: ServiceFactory<Request, Config = AppConfig> + 'static,
S::Error: Into<Error> + 'static,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>> + 'static,
<S::Service as Service<Request>>::Future: 'static,
S::Service: 'static,
B: MessageBody + 'static,
{
pub fn new(factory: F) -> Self {
HttpServer {
factory,
config: Arc::new(Mutex::new(Config {
host: None,
keep_alive: KeepAlive::Timeout(5),
client_timeout: 5000,
client_shutdown: 5000,
})),
backlog: 1024,
sockets: Vec::new(),
builder: ServerBuilder::default(),
on_connect_fn: None,
_phantom: PhantomData,
}
}
pub fn on_connect<CB>(self, f: CB) -> HttpServer<F, I, S, B>
where
CB: Fn(&dyn Any, &mut Extensions) + Send + Sync + 'static,
{
HttpServer {
factory: self.factory,
config: self.config,
backlog: self.backlog,
sockets: self.sockets,
builder: self.builder,
on_connect_fn: Some(Arc::new(f)),
_phantom: PhantomData,
}
}
pub fn workers(mut self, num: usize) -> Self {
self.builder = self.builder.workers(num);
self
}
pub fn backlog(mut self, backlog: u32) -> Self {
self.backlog = backlog;
self.builder = self.builder.backlog(backlog);
self
}
pub fn max_connections(mut self, num: usize) -> Self {
self.builder = self.builder.max_concurrent_connections(num);
self
}
#[allow(unused_variables)]
pub fn max_connection_rate(self, num: usize) -> Self {
#[cfg(any(feature = "rustls", feature = "openssl"))]
actix_tls::accept::max_concurrent_tls_connect(num);
self
}
pub fn worker_max_blocking_threads(mut self, num: usize) -> Self {
self.builder = self.builder.worker_max_blocking_threads(num);
self
}
pub fn keep_alive<T: Into<KeepAlive>>(self, val: T) -> Self {
self.config.lock().unwrap().keep_alive = val.into();
self
}
pub fn client_timeout(self, val: u64) -> Self {
self.config.lock().unwrap().client_timeout = val;
self
}
pub fn client_shutdown(self, val: u64) -> Self {
self.config.lock().unwrap().client_shutdown = val;
self
}
pub fn server_hostname<T: AsRef<str>>(self, val: T) -> Self {
self.config.lock().unwrap().host = Some(val.as_ref().to_owned());
self
}
pub fn system_exit(mut self) -> Self {
self.builder = self.builder.system_exit();
self
}
pub fn disable_signals(mut self) -> Self {
self.builder = self.builder.disable_signals();
self
}
pub fn shutdown_timeout(mut self, sec: u64) -> Self {
self.builder = self.builder.shutdown_timeout(sec);
self
}
pub fn addrs(&self) -> Vec<net::SocketAddr> {
self.sockets.iter().map(|s| s.addr).collect()
}
pub fn addrs_with_scheme(&self) -> Vec<(net::SocketAddr, &str)> {
self.sockets.iter().map(|s| (s.addr, s.scheme)).collect()
}
pub fn listen(mut self, lst: net::TcpListener) -> io::Result<Self> {
let cfg = self.config.clone();
let factory = self.factory.clone();
let addr = lst.local_addr().unwrap();
self.sockets.push(Socket {
addr,
scheme: "http",
});
let on_connect_fn = self.on_connect_fn.clone();
self.builder =
self.builder
.listen(format!("actix-web-service-{}", addr), lst, move || {
let c = cfg.lock().unwrap();
let host = c.host.clone().unwrap_or_else(|| format!("{}", addr));
let mut svc = HttpService::build()
.keep_alive(c.keep_alive)
.client_timeout(c.client_timeout)
.client_disconnect(c.client_shutdown)
.local_addr(addr);
if let Some(handler) = on_connect_fn.clone() {
svc = svc.on_connect_ext(move |io: &_, ext: _| {
(handler)(io as &dyn Any, ext)
})
};
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
svc.finish(map_config(fac, move |_| {
AppConfig::new(false, host.clone(), addr)
}))
.tcp()
})?;
Ok(self)
}
#[cfg(feature = "openssl")]
pub fn listen_openssl(
self,
lst: net::TcpListener,
builder: SslAcceptorBuilder,
) -> io::Result<Self> {
self.listen_ssl_inner(lst, openssl_acceptor(builder)?)
}
#[cfg(feature = "openssl")]
fn listen_ssl_inner(
mut self,
lst: net::TcpListener,
acceptor: SslAcceptor,
) -> io::Result<Self> {
let factory = self.factory.clone();
let cfg = self.config.clone();
let addr = lst.local_addr().unwrap();
self.sockets.push(Socket {
addr,
scheme: "https",
});
let on_connect_fn = self.on_connect_fn.clone();
self.builder =
self.builder
.listen(format!("actix-web-service-{}", addr), lst, move || {
let c = cfg.lock().unwrap();
let host = c.host.clone().unwrap_or_else(|| format!("{}", addr));
let svc = HttpService::build()
.keep_alive(c.keep_alive)
.client_timeout(c.client_timeout)
.client_disconnect(c.client_shutdown)
.local_addr(addr);
let svc = if let Some(handler) = on_connect_fn.clone() {
svc.on_connect_ext(move |io: &_, ext: _| {
(&*handler)(io as &dyn Any, ext)
})
} else {
svc
};
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
svc.finish(map_config(fac, move |_| {
AppConfig::new(true, host.clone(), addr)
}))
.openssl(acceptor.clone())
})?;
Ok(self)
}
#[cfg(feature = "rustls")]
pub fn listen_rustls(
self,
lst: net::TcpListener,
config: RustlsServerConfig,
) -> io::Result<Self> {
self.listen_rustls_inner(lst, config)
}
#[cfg(feature = "rustls")]
fn listen_rustls_inner(
mut self,
lst: net::TcpListener,
config: RustlsServerConfig,
) -> io::Result<Self> {
let factory = self.factory.clone();
let cfg = self.config.clone();
let addr = lst.local_addr().unwrap();
self.sockets.push(Socket {
addr,
scheme: "https",
});
let on_connect_fn = self.on_connect_fn.clone();
self.builder =
self.builder
.listen(format!("actix-web-service-{}", addr), lst, move || {
let c = cfg.lock().unwrap();
let host = c.host.clone().unwrap_or_else(|| format!("{}", addr));
let svc = HttpService::build()
.keep_alive(c.keep_alive)
.client_timeout(c.client_timeout)
.client_disconnect(c.client_shutdown);
let svc = if let Some(handler) = on_connect_fn.clone() {
svc.on_connect_ext(move |io: &_, ext: _| (handler)(io as &dyn Any, ext))
} else {
svc
};
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
svc.finish(map_config(fac, move |_| {
AppConfig::new(true, host.clone(), addr)
}))
.rustls(config.clone())
})?;
Ok(self)
}
pub fn bind<A: net::ToSocketAddrs>(mut self, addr: A) -> io::Result<Self> {
let sockets = self.bind2(addr)?;
for lst in sockets {
self = self.listen(lst)?;
}
Ok(self)
}
fn bind2<A: net::ToSocketAddrs>(&self, addr: A) -> io::Result<Vec<net::TcpListener>> {
let mut err = None;
let mut success = false;
let mut sockets = Vec::new();
for addr in addr.to_socket_addrs()? {
match create_tcp_listener(addr, self.backlog) {
Ok(lst) => {
success = true;
sockets.push(lst);
}
Err(e) => err = Some(e),
}
}
if success {
Ok(sockets)
} else if let Some(e) = err.take() {
Err(e)
} else {
Err(io::Error::new(
io::ErrorKind::Other,
"Can not bind to address.",
))
}
}
#[cfg(feature = "openssl")]
pub fn bind_openssl<A>(mut self, addr: A, builder: SslAcceptorBuilder) -> io::Result<Self>
where
A: net::ToSocketAddrs,
{
let sockets = self.bind2(addr)?;
let acceptor = openssl_acceptor(builder)?;
for lst in sockets {
self = self.listen_ssl_inner(lst, acceptor.clone())?;
}
Ok(self)
}
#[cfg(feature = "rustls")]
pub fn bind_rustls<A: net::ToSocketAddrs>(
mut self,
addr: A,
config: RustlsServerConfig,
) -> io::Result<Self> {
let sockets = self.bind2(addr)?;
for lst in sockets {
self = self.listen_rustls_inner(lst, config.clone())?;
}
Ok(self)
}
#[cfg(unix)]
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::{fn_service, ServiceFactoryExt as _};
let cfg = self.config.clone();
let factory = self.factory.clone();
let socket_addr =
net::SocketAddr::new(net::IpAddr::V4(net::Ipv4Addr::new(127, 0, 0, 1)), 8080);
self.sockets.push(Socket {
scheme: "http",
addr: socket_addr,
});
let addr = lst.local_addr()?;
let name = format!("actix-web-service-{:?}", addr);
let on_connect_fn = self.on_connect_fn.clone();
self.builder = self.builder.listen_uds(name, lst, move || {
let c = cfg.lock().unwrap();
let config = AppConfig::new(
false,
c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
socket_addr,
);
fn_service(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) }).and_then({
let mut svc = HttpService::build()
.keep_alive(c.keep_alive)
.client_timeout(c.client_timeout)
.client_disconnect(c.client_shutdown);
if let Some(handler) = on_connect_fn.clone() {
svc = svc
.on_connect_ext(move |io: &_, ext: _| (&*handler)(io as &dyn Any, ext));
}
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
svc.finish(map_config(fac, move |_| config.clone()))
})
})?;
Ok(self)
}
#[cfg(unix)]
pub fn bind_uds<A>(mut self, addr: A) -> io::Result<Self>
where
A: AsRef<std::path::Path>,
{
use actix_http::Protocol;
use actix_rt::net::UnixStream;
use actix_service::{fn_service, ServiceFactoryExt as _};
let cfg = self.config.clone();
let factory = self.factory.clone();
let socket_addr =
net::SocketAddr::new(net::IpAddr::V4(net::Ipv4Addr::new(127, 0, 0, 1)), 8080);
self.sockets.push(Socket {
scheme: "http",
addr: socket_addr,
});
self.builder = self.builder.bind_uds(
format!("actix-web-service-{:?}", addr.as_ref()),
addr,
move || {
let c = cfg.lock().unwrap();
let config = AppConfig::new(
false,
c.host.clone().unwrap_or_else(|| format!("{}", socket_addr)),
socket_addr,
);
let fac = factory()
.into_factory()
.map_err(|err| err.into().error_response());
fn_service(|io: UnixStream| async { Ok((io, Protocol::Http1, None)) }).and_then(
HttpService::build()
.keep_alive(c.keep_alive)
.client_timeout(c.client_timeout)
.client_disconnect(c.client_shutdown)
.finish(map_config(fac, move |_| config.clone())),
)
},
)?;
Ok(self)
}
}
impl<F, I, S, B> HttpServer<F, I, S, B>
where
F: Fn() -> I + Send + Clone + 'static,
I: IntoServiceFactory<S, Request>,
S: ServiceFactory<Request, Config = AppConfig>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
S::Service: 'static,
B: MessageBody,
{
pub fn run(self) -> Server {
self.builder.run()
}
}
fn create_tcp_listener(addr: net::SocketAddr, backlog: u32) -> io::Result<net::TcpListener> {
use socket2::{Domain, Protocol, Socket, Type};
let domain = Domain::for_address(addr);
let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
socket.set_reuse_address(true)?;
socket.bind(&addr.into())?;
let backlog = cmp::min(backlog, i32::MAX as u32) as i32;
socket.listen(backlog)?;
Ok(net::TcpListener::from(socket))
}
#[cfg(feature = "openssl")]
fn openssl_acceptor(mut builder: SslAcceptorBuilder) -> io::Result<SslAcceptor> {
builder.set_alpn_select_callback(|_, protocols| {
const H2: &[u8] = b"\x02h2";
const H11: &[u8] = b"\x08http/1.1";
if protocols.windows(3).any(|window| window == H2) {
Ok(b"h2")
} else if protocols.windows(9).any(|window| window == H11) {
Ok(b"http/1.1")
} else {
Err(AlpnError::NOACK)
}
});
builder.set_alpn_protos(b"\x08http/1.1\x02h2")?;
Ok(builder.build())
}