1
0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-24 16:02:59 +01:00
actix-extras/src/server.rs

610 lines
20 KiB
Rust
Raw Normal View History

use std::{io, net, thread};
2017-10-07 06:48:14 +02:00
use std::rc::Rc;
use std::sync::Arc;
use std::marker::PhantomData;
2017-10-07 06:48:14 +02:00
use actix::dev::*;
use futures::Stream;
use futures::sync::mpsc;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_core::net::TcpStream;
use num_cpus;
use socket2::{Socket, Domain, Type};
2017-10-07 06:48:14 +02:00
#[cfg(feature="tls")]
use futures::{future, Future};
#[cfg(feature="tls")]
use native_tls::TlsAcceptor;
#[cfg(feature="tls")]
use tokio_tls::{TlsStream, TlsAcceptorExt};
2017-11-04 21:24:57 +01:00
#[cfg(feature="alpn")]
use futures::{future, Future};
#[cfg(feature="alpn")]
use openssl::ssl::{SslMethod, SslAcceptor, SslAcceptorBuilder};
#[cfg(feature="alpn")]
use openssl::pkcs12::ParsedPkcs12;
#[cfg(feature="alpn")]
use tokio_openssl::{SslStream, SslAcceptorExt};
2017-12-06 20:00:39 +01:00
use channel::{HttpChannel, HttpHandler, IntoHttpHandler};
2017-10-22 03:54:24 +02:00
2017-12-08 07:54:44 +01:00
/// Various server settings
2017-12-08 18:24:05 +01:00
#[derive(Debug, Clone)]
pub struct ServerSettings {
addr: Option<net::SocketAddr>,
2017-12-08 07:54:44 +01:00
secure: bool,
host: String,
}
impl Default for ServerSettings {
fn default() -> Self {
ServerSettings {
addr: None,
secure: false,
host: "localhost:8080".to_owned(),
}
}
2017-12-08 07:54:44 +01:00
}
2017-12-08 18:24:05 +01:00
impl ServerSettings {
2017-12-08 07:54:44 +01:00
/// Crate server settings instance
fn new(addr: Option<net::SocketAddr>, secure: bool) -> Self {
2017-12-08 18:24:05 +01:00
let host = if let Some(ref addr) = addr {
format!("{}", addr)
2017-12-08 18:24:05 +01:00
} else {
"unknown".to_owned()
2017-12-08 18:24:05 +01:00
};
ServerSettings {
addr: addr,
secure: secure,
host: host,
}
2017-12-08 07:54:44 +01:00
}
/// Returns the socket address of the local half of this TCP connection
pub fn local_addr(&self) -> Option<net::SocketAddr> {
2017-12-08 18:24:05 +01:00
self.addr
2017-12-08 07:54:44 +01:00
}
/// Returns true if connection is secure(https)
pub fn secure(&self) -> bool {
2017-12-08 18:24:05 +01:00
self.secure
2017-12-08 07:54:44 +01:00
}
2017-12-08 18:24:05 +01:00
/// Returns host header value
pub fn host(&self) -> &str {
&self.host
2017-12-08 07:54:44 +01:00
}
}
2017-10-08 23:56:51 +02:00
/// An HTTP Server
///
/// `T` - async stream, anything that implements `AsyncRead` + `AsyncWrite`.
///
/// `A` - peer address
2017-10-22 03:54:24 +02:00
///
/// `H` - request handler
pub struct HttpServer<T, A, H, U>
where H: 'static
{
2017-12-08 18:24:05 +01:00
h: Rc<Vec<H>>,
io: PhantomData<T>,
addr: PhantomData<A>,
threads: usize,
factory: Arc<Fn() -> U + Send + Sync>,
workers: Vec<SyncAddress<Worker<H>>>,
2017-10-07 06:48:14 +02:00
}
impl<T: 'static, A: 'static, H, U: 'static> Actor for HttpServer<T, A, H, U> {
2017-10-07 06:48:14 +02:00
type Context = Context<Self>;
}
impl<T, A, H, U, V> HttpServer<T, A, H, U>
where H: HttpHandler,
U: IntoIterator<Item=V> + 'static,
V: IntoHttpHandler<Handler=H>,
2017-10-22 03:54:24 +02:00
{
/// Create new http server with vec of http handlers
pub fn new<F>(factory: F) -> Self
where F: Sync + Send + 'static + Fn() -> U,
2017-12-06 20:00:39 +01:00
{
HttpServer{ h: Rc::new(Vec::new()),
2017-10-22 03:54:24 +02:00
io: PhantomData,
addr: PhantomData,
threads: num_cpus::get(),
factory: Arc::new(factory),
workers: Vec::new(),
}
}
/// Set number of workers to start.
///
/// By default http server uses number of available logical cpu as threads count.
pub fn threads(mut self, num: usize) -> Self {
self.threads = num;
self
2017-10-07 06:48:14 +02:00
}
}
impl<T, A, H, U, V> HttpServer<T, A, H, U>
where T: AsyncRead + AsyncWrite + 'static,
2017-10-22 03:54:24 +02:00
A: 'static,
H: HttpHandler,
U: IntoIterator<Item=V> + 'static,
V: IntoHttpHandler<Handler=H>,
{
/// Start listening for incomming connections from a stream.
///
/// This method uses only one thread for handling incoming connections.
2017-12-08 07:54:44 +01:00
pub fn serve_incoming<S, Addr>(mut self, stream: S, secure: bool) -> io::Result<Addr>
where Self: ActorAddress<Self, Addr>,
S: Stream<Item=(T, A), Error=io::Error> + 'static
{
2017-12-08 18:24:05 +01:00
// set server settings
let addr: net::SocketAddr = "127.0.0.1:8080".parse().unwrap();
2017-12-08 18:24:05 +01:00
let settings = ServerSettings::new(Some(addr), secure);
let mut apps: Vec<_> = (*self.factory)().into_iter().map(|h| h.into_handler()).collect();
for app in &mut apps {
app.server_settings(settings.clone());
2017-12-08 18:24:05 +01:00
}
self.h = Rc::new(apps);
2017-12-08 07:54:44 +01:00
2017-12-08 18:24:05 +01:00
// start server
Ok(HttpServer::create(move |ctx| {
ctx.add_stream(stream.map(
2017-12-08 18:24:05 +01:00
move |(t, _)| IoStream{io: t, peer: None, http2: false}));
self
}))
}
2017-10-07 06:48:14 +02:00
fn bind<S: net::ToSocketAddrs>(&self, addr: S)
-> io::Result<Vec<(net::SocketAddr, Socket)>>
2017-10-07 06:48:14 +02:00
{
2017-10-15 23:53:03 +02:00
let mut err = None;
let mut sockets = Vec::new();
2017-10-16 10:19:23 +02:00
if let Ok(iter) = addr.to_socket_addrs() {
2017-10-15 23:53:03 +02:00
for addr in iter {
match addr {
net::SocketAddr::V4(a) => {
let socket = Socket::new(Domain::ipv4(), Type::stream(), None)?;
match socket.bind(&a.into()) {
Ok(_) => {
socket.listen(1024)
.expect("failed to set socket backlog");
socket.set_reuse_address(true)
.expect("failed to set socket reuse address");
sockets.push((addr, socket));
},
Err(e) => err = Some(e),
}
}
net::SocketAddr::V6(a) => {
let socket = Socket::new(Domain::ipv6(), Type::stream(), None)?;
match socket.bind(&a.into()) {
Ok(_) => {
socket.listen(1024)
.expect("failed to set socket backlog");
socket.set_reuse_address(true)
.expect("failed to set socket reuse address");
sockets.push((addr, socket))
}
Err(e) => err = Some(e),
}
}
2017-10-15 23:53:03 +02:00
}
}
}
if sockets.is_empty() {
2017-10-15 23:53:03 +02:00
if let Some(e) = err.take() {
Err(e)
} else {
Err(io::Error::new(io::ErrorKind::Other, "Can not bind to address."))
}
} else {
Ok(sockets)
2017-10-15 23:53:03 +02:00
}
2017-10-07 06:48:14 +02:00
}
fn start_workers(&mut self, settings: &ServerSettings)
-> Vec<mpsc::UnboundedSender<IoStream<net::TcpStream>>>
{
// start workers
let mut workers = Vec::new();
for _ in 0..self.threads {
let s = settings.clone();
let (tx, rx) = mpsc::unbounded::<IoStream<net::TcpStream>>();
let factory = Arc::clone(&self.factory);
let addr = Arbiter::start(move |ctx: &mut Context<_>| {
let mut apps: Vec<_> = (*factory)()
.into_iter().map(|h| h.into_handler()).collect();
for app in &mut apps {
app.server_settings(s.clone());
}
ctx.add_stream(rx);
Worker{h: Rc::new(apps)}
});
workers.push(tx);
self.workers.push(addr);
}
info!("Starting {} http workers", self.threads);
workers
}
}
impl<H: HttpHandler, U, V> HttpServer<TcpStream, net::SocketAddr, H, U>
where U: IntoIterator<Item=V> + 'static,
V: IntoHttpHandler<Handler=H>,
{
/// Start listening for incomming connections.
///
/// This methods converts address to list of `SocketAddr`
/// then binds to all available addresses.
/// It also starts number of http handler workers in seperate threads.
/// For each address this method starts separate thread which does `accept()` in a loop.
2017-12-08 07:54:44 +01:00
pub fn serve<S, Addr>(mut self, addr: S) -> io::Result<Addr>
where Self: ActorAddress<Self, Addr>,
S: net::ToSocketAddrs,
{
let addrs = self.bind(addr)?;
2017-12-08 18:24:05 +01:00
let settings = ServerSettings::new(Some(addrs[0].0), false);
let workers = self.start_workers(&settings);
// start acceptors threads
for (addr, sock) in addrs {
let wrks = workers.clone();
let _ = thread::Builder::new().name(format!("Accept on {}", addr)).spawn(move || {
let mut next = 0;
loop {
match sock.accept() {
Ok((socket, addr)) => {
let addr = if let Some(addr) = addr.as_inet() {
net::SocketAddr::V4(addr)
} else {
net::SocketAddr::V6(addr.as_inet6().unwrap())
};
let msg = IoStream{
io: socket.into_tcp_stream(), peer: Some(addr), http2: false};
2017-12-13 20:10:03 +01:00
println!("next: {}", next);
wrks[next].unbounded_send(msg).expect("worker thread died");
next = (next + 1) % wrks.len();
}
Err(err) => error!("Error accepting connection: {:?}", err),
}
}
});
info!("Starting http server on {}", addr);
2017-12-08 18:24:05 +01:00
}
// start http server actor
Ok(HttpServer::create(|_| {self}))
}
}
#[cfg(feature="tls")]
impl<H: HttpHandler, U, V> HttpServer<TlsStream<TcpStream>, net::SocketAddr, H, U>
where U: IntoIterator<Item=V> + 'static,
V: IntoHttpHandler<Handler=H>,
{
/// Start listening for incomming tls connections.
///
/// This methods converts address to list of `SocketAddr`
/// then binds to all available addresses.
pub fn serve_tls<S, Addr>(self, addr: S, pkcs12: ::Pkcs12) -> io::Result<Addr>
where Self: ActorAddress<Self, Addr>,
S: net::ToSocketAddrs,
{
let addrs = self.bind(addr)?;
let settings = ServerSettings::new(Some(addrs[0].0), false);
let acceptor = match TlsAcceptor::builder(pkcs12) {
Ok(builder) => {
match builder.build() {
Ok(acceptor) => acceptor,
Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err))
}
}
Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err))
};
// start workers
let mut workers = Vec::new();
for _ in 0..self.threads {
let s = settings.clone();
let (tx, rx) = mpsc::unbounded::<IoStream<net::TcpStream>>();
let acc = acceptor.clone();
let factory = Arc::clone(&self.factory);
let _addr = Arbiter::start(move |ctx: &mut Context<_>| {
let mut apps: Vec<_> = (*factory)()
.into_iter().map(|h| h.into_handler()).collect();
for app in &mut apps {
app.server_settings(s.clone());
}
ctx.add_stream(rx);
TlsWorker{h: Rc::new(apps), acceptor: acc}
});
workers.push(tx);
// self.workers.push(addr);
}
info!("Starting {} http workers", self.threads);
// start acceptors threads
for (addr, sock) in addrs {
let wrks = workers.clone();
let _ = thread::Builder::new().name(format!("Accept on {}", addr)).spawn(move || {
let mut next = 0;
loop {
match sock.accept() {
Ok((socket, addr)) => {
let addr = if let Some(addr) = addr.as_inet() {
net::SocketAddr::V4(addr)
} else {
net::SocketAddr::V6(addr.as_inet6().unwrap())
};
let msg = IoStream{
io: socket.into_tcp_stream(), peer: Some(addr), http2: false};
wrks[next].unbounded_send(msg).expect("worker thread died");
next = (next + 1) % wrks.len();
}
Err(err) => error!("Error accepting connection: {:?}", err),
}
}
});
info!("Starting tls http server on {}", addr);
}
// start http server actor
Ok(HttpServer::create(|_| {self}))
}
}
#[cfg(feature="alpn")]
impl<H: HttpHandler, U, V> HttpServer<SslStream<TcpStream>, net::SocketAddr, H, U>
where U: IntoIterator<Item=V> + 'static,
V: IntoHttpHandler<Handler=H>,
{
/// Start listening for incomming tls connections.
///
/// This methods converts address to list of `SocketAddr`
/// then binds to all available addresses.
pub fn serve_tls<S, Addr>(self, addr: S, identity: &ParsedPkcs12) -> io::Result<Addr>
where Self: ActorAddress<Self, Addr>,
S: net::ToSocketAddrs,
{
let addrs = self.bind(addr)?;
let settings = ServerSettings::new(Some(addrs[0].0), false);
2017-12-08 07:54:44 +01:00
let acceptor = match SslAcceptorBuilder::mozilla_intermediate(
SslMethod::tls(), &identity.pkey, &identity.cert, &identity.chain)
{
Ok(mut builder) => {
match builder.set_alpn_protocols(&[b"h2", b"http/1.1"]) {
Ok(_) => builder.build(),
Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err)),
}
},
Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err))
};
// start workers
let mut workers = Vec::new();
for _ in 0..self.threads {
let s = settings.clone();
let (tx, rx) = mpsc::unbounded::<IoStream<net::TcpStream>>();
let acc = acceptor.clone();
let factory = Arc::clone(&self.factory);
let _addr = Arbiter::start(move |ctx: &mut Context<_>| {
let mut apps: Vec<_> = (*factory)()
.into_iter().map(|h| h.into_handler()).collect();
for app in &mut apps {
app.server_settings(s.clone());
}
ctx.add_stream(rx);
AlpnWorker{h: Rc::new(apps), acceptor: acc}
});
workers.push(tx);
// self.workers.push(addr);
}
info!("Starting {} http workers", self.threads);
// start acceptors threads
for (addr, sock) in addrs {
let wrks = workers.clone();
let _ = thread::Builder::new().name(format!("Accept on {}", addr)).spawn(move || {
let mut next = 0;
loop {
match sock.accept() {
Ok((socket, addr)) => {
let addr = if let Some(addr) = addr.as_inet() {
net::SocketAddr::V4(addr)
} else {
net::SocketAddr::V6(addr.as_inet6().unwrap())
};
let msg = IoStream{
io: socket.into_tcp_stream(), peer: Some(addr), http2: false};
wrks[next].unbounded_send(msg).expect("worker thread died");
next = (next + 1) % wrks.len();
}
Err(err) => error!("Error accepting connection: {:?}", err),
}
}
});
info!("Starting tls http server on {}", addr);
}
// start http server actor
Ok(HttpServer::create(|_| {self}))
}
}
2017-12-08 18:24:05 +01:00
struct IoStream<T> {
io: T,
peer: Option<net::SocketAddr>,
http2: bool,
}
2017-10-22 00:21:16 +02:00
2017-12-08 18:24:05 +01:00
impl<T> ResponseType for IoStream<T>
{
2017-10-07 06:48:14 +02:00
type Item = ();
type Error = ();
}
impl<T, A, H, U> StreamHandler<IoStream<T>, io::Error> for HttpServer<T, A, H, U>
2017-10-22 03:54:24 +02:00
where T: AsyncRead + AsyncWrite + 'static,
H: HttpHandler + 'static,
U: 'static,
A: 'static {}
2017-10-07 06:48:14 +02:00
impl<T, A, H, U> Handler<IoStream<T>, io::Error> for HttpServer<T, A, H, U>
where T: AsyncRead + AsyncWrite + 'static,
2017-10-22 03:54:24 +02:00
H: HttpHandler + 'static,
U: 'static,
A: 'static,
{
fn error(&mut self, err: io::Error, _: &mut Context<Self>) {
2017-11-04 17:07:44 +01:00
debug!("Error handling request: {}", err)
}
2017-12-08 18:24:05 +01:00
fn handle(&mut self, msg: IoStream<T>, _: &mut Context<Self>)
-> Response<Self, IoStream<T>>
2017-10-07 06:48:14 +02:00
{
Arbiter::handle().spawn(
2017-12-08 18:24:05 +01:00
HttpChannel::new(Rc::clone(&self.h), msg.io, msg.peer, msg.http2));
2017-10-07 09:22:09 +02:00
Self::empty()
2017-10-07 06:48:14 +02:00
}
}
/// Http workers
///
/// Worker accepts Socket objects via unbounded channel and start requests processing.
struct Worker<H> {
h: Rc<Vec<H>>,
}
impl<H: 'static> Actor for Worker<H> {
type Context = Context<Self>;
}
impl<H> StreamHandler<IoStream<net::TcpStream>> for Worker<H>
where H: HttpHandler + 'static {}
impl<H> Handler<IoStream<net::TcpStream>> for Worker<H>
where H: HttpHandler + 'static,
{
fn handle(&mut self, msg: IoStream<net::TcpStream>, _: &mut Context<Self>)
-> Response<Self, IoStream<net::TcpStream>>
{
let io = TcpStream::from_stream(msg.io, Arbiter::handle())
.expect("failed to associate TCP stream");
Arbiter::handle().spawn(
HttpChannel::new(Rc::clone(&self.h), io, msg.peer, msg.http2));
Self::empty()
}
}
/// Tls http workers
///
/// Worker accepts Socket objects via unbounded channel and start requests processing.
#[cfg(feature="tls")]
struct TlsWorker<H> {
h: Rc<Vec<H>>,
acceptor: TlsAcceptor,
}
#[cfg(feature="tls")]
impl<H: 'static> Actor for TlsWorker<H> {
type Context = Context<Self>;
}
#[cfg(feature="tls")]
impl<H> StreamHandler<IoStream<net::TcpStream>> for TlsWorker<H>
where H: HttpHandler + 'static {}
#[cfg(feature="tls")]
impl<H> Handler<IoStream<net::TcpStream>> for TlsWorker<H>
where H: HttpHandler + 'static,
{
fn handle(&mut self, msg: IoStream<net::TcpStream>, _: &mut Context<Self>)
-> Response<Self, IoStream<net::TcpStream>>
{
let IoStream { io, peer, http2 } = msg;
let io = TcpStream::from_stream(io, Arbiter::handle())
.expect("failed to associate TCP stream");
let h = Rc::clone(&self.h);
Arbiter::handle().spawn(
TlsAcceptorExt::accept_async(&self.acceptor, io).then(move |res| {
match res {
Ok(io) => Arbiter::handle().spawn(
HttpChannel::new(h, io, peer, http2)),
Err(err) =>
trace!("Error during handling tls connection: {}", err),
};
future::result(Ok(()))
})
);
Self::empty()
}
}
/// Tls http workers with alpn support
///
/// Worker accepts Socket objects via unbounded channel and start requests processing.
#[cfg(feature="alpn")]
struct AlpnWorker<H> {
h: Rc<Vec<H>>,
acceptor: SslAcceptor,
}
#[cfg(feature="alpn")]
impl<H: 'static> Actor for AlpnWorker<H> {
type Context = Context<Self>;
}
#[cfg(feature="alpn")]
impl<H> StreamHandler<IoStream<net::TcpStream>> for AlpnWorker<H>
where H: HttpHandler + 'static {}
#[cfg(feature="alpn")]
impl<H> Handler<IoStream<net::TcpStream>> for AlpnWorker<H>
where H: HttpHandler + 'static,
{
fn handle(&mut self, msg: IoStream<net::TcpStream>, _: &mut Context<Self>)
-> Response<Self, IoStream<net::TcpStream>>
{
let IoStream { io, peer, .. } = msg;
let io = TcpStream::from_stream(io, Arbiter::handle())
.expect("failed to associate TCP stream");
let h = Rc::clone(&self.h);
Arbiter::handle().spawn(
SslAcceptorExt::accept_async(&self.acceptor, io).then(move |res| {
match res {
Ok(io) => {
let http2 = if let Some(p) = io.get_ref().ssl().selected_alpn_protocol()
{
p.len() == 2 && &p == b"h2"
} else {
false
};
Arbiter::handle().spawn(
HttpChannel::new(h, io, peer, http2));
},
Err(err) =>
trace!("Error during handling tls connection: {}", err),
};
future::result(Ok(()))
})
);
Self::empty()
}
}