actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-28 01:32:57 +01:00
Code Issues Releases Wiki Activity

set server settings to HttpHandler

Browse Source
This commit is contained in:
Nikolay Kim 2017-12-08 09:24:05 -08:00
parent 2192d14eff
commit 1293619096
5 changed files with 92 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/channel.rs
View File

@ -1,3 +1,4 @@
use std::rc::Rc;
use std::net::SocketAddr; use std::net::SocketAddr;
use actix::dev::*; use actix::dev::*;
@ -12,9 +13,13 @@ use httprequest::HttpRequest;
use server::ServerSettings; use server::ServerSettings;
/// Low level http request handler /// Low level http request handler
#[allow(unused_variables)]
pub trait HttpHandler: 'static { pub trait HttpHandler: 'static {
/// Handle request /// Handle request
fn handle(&self, req: HttpRequest) -> Result<Pipeline, HttpRequest>; fn handle(&self, req: HttpRequest) -> Result<Pipeline, HttpRequest>;
/// Set server settings
fn server_settings(&mut self, settings: ServerSettings) {}
} }
/// Conversion helper trait /// Conversion helper trait
@ -51,17 +56,16 @@ pub struct HttpChannel<T, H>
impl<T, H> HttpChannel<T, H> impl<T, H> HttpChannel<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: HttpHandler + 'static where T: AsyncRead + AsyncWrite + 'static, H: HttpHandler + 'static
{ {
pub fn new(settings: ServerSettings<H>, pub fn new(h: Rc<Vec<H>>, io: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H>
stream: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H>
{ {
if http2 { if http2 {
HttpChannel { HttpChannel {
proto: Some(HttpProtocol::H2( proto: Some(HttpProtocol::H2(
h2::Http2::new(settings, stream, peer, Bytes::new()))) } h2::Http2::new(h, io, peer, Bytes::new()))) }
} else { } else {
HttpChannel { HttpChannel {
proto: Some(HttpProtocol::H1( proto: Some(HttpProtocol::H1(
h1::Http1::new(settings, stream, peer))) } h1::Http1::new(h, io, peer))) }
} }
} }
} }
@ -97,8 +101,7 @@ impl<T, H> Future for HttpChannel<T, H>
return Err(()), return Err(()),
} }
} }
Some(HttpProtocol::H2(ref mut h2)) => Some(HttpProtocol::H2(ref mut h2)) => return h2.poll(),
return h2.poll(),
None => unreachable!(), None => unreachable!(),
} }
@ -106,9 +109,9 @@ impl<T, H> Future for HttpChannel<T, H>
let proto = self.proto.take().unwrap(); let proto = self.proto.take().unwrap();
match proto { match proto {
HttpProtocol::H1(h1) => { HttpProtocol::H1(h1) => {
let (settings, stream, addr, buf) = h1.into_inner(); let (h, io, addr, buf) = h1.into_inner();
self.proto = Some( self.proto = Some(
HttpProtocol::H2(h2::Http2::new(settings, stream, addr, buf))); HttpProtocol::H2(h2::Http2::new(h, io, addr, buf)));
self.poll() self.poll()
} }
_ => unreachable!() _ => unreachable!()

22
src/h1.rs
View File

@ -1,4 +1,5 @@
use std::{self, io, ptr}; use std::{self, io, ptr};
use std::rc::Rc;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::time::Duration; use std::time::Duration;
use std::collections::VecDeque; use std::collections::VecDeque;
@ -20,7 +21,6 @@ use httpcodes::HTTPNotFound;
use httprequest::HttpRequest; use httprequest::HttpRequest;
use error::{ParseError, PayloadError, ResponseError}; use error::{ParseError, PayloadError, ResponseError};
use payload::{Payload, PayloadWriter, DEFAULT_BUFFER_SIZE}; use payload::{Payload, PayloadWriter, DEFAULT_BUFFER_SIZE};
use server::ServerSettings;
const KEEPALIVE_PERIOD: u64 = 15; // seconds const KEEPALIVE_PERIOD: u64 = 15; // seconds
const INIT_BUFFER_SIZE: usize = 8192; const INIT_BUFFER_SIZE: usize = 8192;
@ -31,7 +31,6 @@ const HTTP2_PREFACE: [u8; 14] = *b"PRI * HTTP/2.0";
bitflags! { bitflags! {
struct Flags: u8 { struct Flags: u8 {
const SECURE = 0b0000_0001;
const ERROR = 0b0000_0010; const ERROR = 0b0000_0010;
const KEEPALIVE = 0b0000_0100; const KEEPALIVE = 0b0000_0100;
const H2 = 0b0000_1000; const H2 = 0b0000_1000;
@ -60,7 +59,7 @@ enum Item {
pub(crate) struct Http1<T: AsyncWrite + 'static, H: 'static> { pub(crate) struct Http1<T: AsyncWrite + 'static, H: 'static> {
flags: Flags, flags: Flags,
settings: ServerSettings<H>, handlers: Rc<Vec<H>>,
addr: Option<SocketAddr>, addr: Option<SocketAddr>,
stream: H1Writer<T>, stream: H1Writer<T>,
reader: Reader, reader: Reader,
@ -78,14 +77,9 @@ impl<T, H> Http1<T, H>
where T: AsyncRead + AsyncWrite + 'static, where T: AsyncRead + AsyncWrite + 'static,
H: HttpHandler + 'static H: HttpHandler + 'static
{ {
pub fn new(settings: ServerSettings<H>, stream: T, addr: Option<SocketAddr>) -> Self { pub fn new(h: Rc<Vec<H>>, stream: T, addr: Option<SocketAddr>) -> Self {
let flags = if settings.secure() { Http1{ flags: Flags::KEEPALIVE,
Flags::SECURE | Flags::KEEPALIVE handlers: h,
} else {
Flags::KEEPALIVE
};
Http1{ flags: flags,
settings: settings,
addr: addr, addr: addr,
stream: H1Writer::new(stream), stream: H1Writer::new(stream),
reader: Reader::new(), reader: Reader::new(),
@ -94,8 +88,8 @@ impl<T, H> Http1<T, H>
keepalive_timer: None } keepalive_timer: None }
} }
pub fn into_inner(mut self) -> (ServerSettings<H>, T, Option<SocketAddr>, Bytes) { pub fn into_inner(mut self) -> (Rc<Vec<H>>, T, Option<SocketAddr>, Bytes) {
(self.settings, self.stream.unwrap(), self.addr, self.read_buf.freeze()) (self.handlers, self.stream.unwrap(), self.addr, self.read_buf.freeze())
} }
pub fn poll(&mut self) -> Poll<Http1Result, ()> { pub fn poll(&mut self) -> Poll<Http1Result, ()> {
@ -204,7 +198,7 @@ impl<T, H> Http1<T, H>
// start request processing // start request processing
let mut pipe = None; let mut pipe = None;
for h in self.settings.handlers() { for h in self.handlers.iter() {
req = match h.handle(req) { req = match h.handle(req) {
Ok(t) => { Ok(t) => {
pipe = Some(t); pipe = Some(t);

17
src/h2.rs
View File

@ -1,4 +1,5 @@
use std::{io, cmp, mem}; use std::{io, cmp, mem};
use std::rc::Rc;
use std::io::{Read, Write}; use std::io::{Read, Write};
use std::time::Duration; use std::time::Duration;
use std::net::SocketAddr; use std::net::SocketAddr;
@ -21,13 +22,11 @@ use encoding::PayloadType;
use httpcodes::HTTPNotFound; use httpcodes::HTTPNotFound;
use httprequest::HttpRequest; use httprequest::HttpRequest;
use payload::{Payload, PayloadWriter}; use payload::{Payload, PayloadWriter};
use server::ServerSettings;
const KEEPALIVE_PERIOD: u64 = 15; // seconds const KEEPALIVE_PERIOD: u64 = 15; // seconds
bitflags! { bitflags! {
struct Flags: u8 { struct Flags: u8 {
const SECURE = 0b0000_0001;
const DISCONNECTED = 0b0000_0010; const DISCONNECTED = 0b0000_0010;
} }
} }
@ -37,7 +36,7 @@ pub(crate) struct Http2<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: 'static where T: AsyncRead + AsyncWrite + 'static, H: 'static
{ {
flags: Flags, flags: Flags,
settings: ServerSettings<H>, handlers: Rc<Vec<H>>,
addr: Option<SocketAddr>, addr: Option<SocketAddr>,
state: State<IoWrapper<T>>, state: State<IoWrapper<T>>,
tasks: VecDeque<Entry>, tasks: VecDeque<Entry>,
@ -54,10 +53,10 @@ impl<T, H> Http2<T, H>
where T: AsyncRead + AsyncWrite + 'static, where T: AsyncRead + AsyncWrite + 'static,
H: HttpHandler + 'static H: HttpHandler + 'static
{ {
pub fn new(settings: ServerSettings<H>, stream: T, addr: Option<SocketAddr>, buf: Bytes) -> Self pub fn new(h: Rc<Vec<H>>, stream: T, addr: Option<SocketAddr>, buf: Bytes) -> Self
{ {
Http2{ flags: if settings.secure() { Flags::SECURE } else { Flags::empty() }, Http2{ flags: Flags::empty(),
settings: settings, handlers: h,
addr: addr, addr: addr,
tasks: VecDeque::new(), tasks: VecDeque::new(),
state: State::Handshake( state: State::Handshake(
@ -152,7 +151,7 @@ impl<T, H> Http2<T, H>
self.keepalive_timer.take(); self.keepalive_timer.take();
self.tasks.push_back( self.tasks.push_back(
Entry::new(parts, body, resp, self.addr, &self.settings)); Entry::new(parts, body, resp, self.addr, &self.handlers));
} }
Ok(Async::NotReady) => { Ok(Async::NotReady) => {
// start keep-alive timer // start keep-alive timer
@ -231,7 +230,7 @@ impl Entry {
recv: RecvStream, recv: RecvStream,
resp: Respond<Bytes>, resp: Respond<Bytes>,
addr: Option<SocketAddr>, addr: Option<SocketAddr>,
settings: &ServerSettings<H>) -> Entry handlers: &Rc<Vec<H>>) -> Entry
where H: HttpHandler + 'static where H: HttpHandler + 'static
{ {
// Payload and Content-Encoding // Payload and Content-Encoding
@ -248,7 +247,7 @@ impl Entry {
// start request processing // start request processing
let mut task = None; let mut task = None;
for h in settings.handlers() { for h in handlers.iter() {
req = match h.handle(req) { req = match h.handle(req) {
Ok(t) => { Ok(t) => {
task = Some(t); task = Some(t);

3
src/httprequest.rs
View File

@ -142,7 +142,8 @@ impl<S> HttpRequest<S> {
&self.0.headers &self.0.headers
} }
#[cfg(test)] #[doc(hidden)]
#[inline]
pub fn headers_mut(&mut self) -> &mut HeaderMap { pub fn headers_mut(&mut self) -> &mut HeaderMap {
&mut self.as_mut().headers &mut self.as_mut().headers
} }

128
src/server.rs
View File

@ -5,6 +5,8 @@ use std::marker::PhantomData;
use actix::dev::*; use actix::dev::*;
use futures::Stream; use futures::Stream;
use http::HttpTryFrom;
use http::header::HeaderValue;
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use tokio_core::net::{TcpListener, TcpStream}; use tokio_core::net::{TcpListener, TcpStream};
@ -27,49 +29,41 @@ use tokio_openssl::{SslStream, SslAcceptorExt};
use channel::{HttpChannel, HttpHandler, IntoHttpHandler}; use channel::{HttpChannel, HttpHandler, IntoHttpHandler};
/// Various server settings /// Various server settings
pub struct ServerSettings<H> (Rc<InnerServerSettings<H>>); #[derive(Debug, Clone)]
pub struct ServerSettings {
struct InnerServerSettings<H> {
h: Vec<H>,
addr: Option<SocketAddr>, addr: Option<SocketAddr>,
secure: bool, secure: bool,
sethost: bool, host: Option<HeaderValue>,
} }
impl<H> Clone for ServerSettings<H> { impl ServerSettings {
fn clone(&self) -> Self {
ServerSettings(Rc::clone(&self.0))
}
}
impl<H> ServerSettings<H> {
/// Crate server settings instance /// Crate server settings instance
fn new(h: Vec<H>, addr: Option<SocketAddr>, secure: bool, sethost: bool) -> Self { fn new(addr: Option<SocketAddr>, secure: bool) -> Self {
ServerSettings( let host = if let Some(ref addr) = addr {
Rc::new(InnerServerSettings { HeaderValue::try_from(format!("{}", addr).as_str()).ok()
h: h, } else {
None
};
ServerSettings {
addr: addr, addr: addr,
secure: secure, secure: secure,
sethost: sethost })) host: host,
}
} }
/// Returns list of http handlers
pub fn handlers(&self) -> &Vec<H> {
&self.0.h
}
/// Returns the socket address of the local half of this TCP connection /// Returns the socket address of the local half of this TCP connection
pub fn local_addr(&self) -> Option<SocketAddr> { pub fn local_addr(&self) -> Option<SocketAddr> {
self.0.addr self.addr
} }
/// Returns true if connection is secure(https) /// Returns true if connection is secure(https)
pub fn secure(&self) -> bool { pub fn secure(&self) -> bool {
self.0.secure self.secure
} }
/// Should http channel set *HOST* header /// Returns host header value
pub fn set_host_header(&self) -> bool { pub fn host(&self) -> Option<&HeaderValue> {
self.0.sethost self.host.as_ref()
} }
} }
@ -81,10 +75,9 @@ impl<H> ServerSettings<H> {
/// ///
/// `H` - request handler /// `H` - request handler
pub struct HttpServer<T, A, H> { pub struct HttpServer<T, A, H> {
h: Option<Vec<H>>, h: Rc<Vec<H>>,
io: PhantomData<T>, io: PhantomData<T>,
addr: PhantomData<A>, addr: PhantomData<A>,
sethost: bool,
} }
impl<T: 'static, A: 'static, H: 'static> Actor for HttpServer<T, A, H> { impl<T: 'static, A: 'static, H: 'static> Actor for HttpServer<T, A, H> {
@ -99,16 +92,9 @@ impl<T, A, H> HttpServer<T, A, H> where H: HttpHandler
{ {
let apps: Vec<_> = handler.into_iter().map(|h| h.into_handler()).collect(); let apps: Vec<_> = handler.into_iter().map(|h| h.into_handler()).collect();
HttpServer {h: Some(apps), HttpServer{ h: Rc::new(apps),
io: PhantomData, io: PhantomData,
addr: PhantomData, addr: PhantomData }
sethost: false}
}
/// Set *HOST* header if not set
pub fn set_host_header(mut self) -> Self {
self.sethost = true;
self
} }
} }
@ -122,14 +108,17 @@ impl<T, A, H> HttpServer<T, A, H>
where Self: ActorAddress<Self, Addr>, where Self: ActorAddress<Self, Addr>,
S: Stream<Item=(T, A), Error=io::Error> + 'static S: Stream<Item=(T, A), Error=io::Error> + 'static
{ {
// set server settings
let addr: SocketAddr = "127.0.0.1:8080".parse().unwrap(); let addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
let settings = ServerSettings::new( let settings = ServerSettings::new(Some(addr), secure);
self.h.take().unwrap(), Some(addr), secure, self.sethost); for h in Rc::get_mut(&mut self.h).unwrap().iter_mut() {
h.server_settings(settings.clone());
}
// start server
Ok(HttpServer::create(move |ctx| { Ok(HttpServer::create(move |ctx| {
ctx.add_stream(stream.map( ctx.add_stream(stream.map(
move |(t, _)| IoStream{settings: settings.clone(), move |(t, _)| IoStream{io: t, peer: None, http2: false}));
io: t, peer: None, http2: false}));
self self
})) }))
} }
@ -170,16 +159,20 @@ impl<H: HttpHandler> HttpServer<TcpStream, net::SocketAddr, H> {
S: net::ToSocketAddrs, S: net::ToSocketAddrs,
{ {
let addrs = self.bind(addr)?; let addrs = self.bind(addr)?;
let settings = ServerSettings::new(
self.h.take().unwrap(), Some(addrs[0].0), false, self.sethost);
// set server settings
let settings = ServerSettings::new(Some(addrs[0].0), false);
for h in Rc::get_mut(&mut self.h).unwrap().iter_mut() {
h.server_settings(settings.clone());
}
// start server
Ok(HttpServer::create(move |ctx| { Ok(HttpServer::create(move |ctx| {
for (addr, tcp) in addrs { for (addr, tcp) in addrs {
info!("Starting http server on {}", addr); info!("Starting http server on {}", addr);
let s = settings.clone();
ctx.add_stream(tcp.incoming().map( ctx.add_stream(tcp.incoming().map(
move |(t, a)| IoStream{settings: s.clone(), move |(t, a)| IoStream{io: t, peer: Some(a), http2: false}));
io: t, peer: Some(a), http2: false}));
} }
self self
})) }))
@ -198,8 +191,12 @@ impl<H: HttpHandler> HttpServer<TlsStream<TcpStream>, net::SocketAddr, H> {
S: net::ToSocketAddrs, S: net::ToSocketAddrs,
{ {
let addrs = self.bind(addr)?; let addrs = self.bind(addr)?;
let settings = ServerSettings::new(
self.h.take().unwrap(), Some(addrs[0].0.clone()), true, self.sethost); // set server settings
let settings = ServerSettings::new(Some(addrs[0].0), true);
for h in Rc::get_mut(&mut self.h).unwrap().iter_mut() {
h.server_settings(settings.clone());
}
let acceptor = match TlsAcceptor::builder(pkcs12) { let acceptor = match TlsAcceptor::builder(pkcs12) {
Ok(builder) => { Ok(builder) => {
@ -211,18 +208,15 @@ impl<H: HttpHandler> HttpServer<TlsStream<TcpStream>, net::SocketAddr, H> {
Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err)) Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err))
}; };
// start server
Ok(HttpServer::create(move |ctx| { Ok(HttpServer::create(move |ctx| {
for (srv, tcp) in addrs { for (srv, tcp) in addrs {
info!("Starting tls http server on {}", srv); info!("Starting tls http server on {}", srv);
let st = settings.clone();
let acc = acceptor.clone(); let acc = acceptor.clone();
ctx.add_stream(tcp.incoming().and_then(move |(stream, addr)| { ctx.add_stream(tcp.incoming().and_then(move |(stream, addr)| {
let st2 = st.clone();
TlsAcceptorExt::accept_async(acc.as_ref(), stream) TlsAcceptorExt::accept_async(acc.as_ref(), stream)
.map(move |t| .map(move |t| IoStream{io: t, peer: Some(addr), http2: false})
IoStream{settings: st2.clone(),
io: t, peer: Some(addr), http2: false})
.map_err(|err| { .map_err(|err| {
trace!("Error during handling tls connection: {}", err); trace!("Error during handling tls connection: {}", err);
io::Error::new(io::ErrorKind::Other, err) io::Error::new(io::ErrorKind::Other, err)
@ -246,8 +240,12 @@ impl<H: HttpHandler> HttpServer<SslStream<TcpStream>, net::SocketAddr, H> {
S: net::ToSocketAddrs, S: net::ToSocketAddrs,
{ {
let addrs = self.bind(addr)?; let addrs = self.bind(addr)?;
let settings = ServerSettings::new(
self.h.take().unwrap(), Some(addrs[0].0.clone()), true, self.sethost); // set server settings
let settings = ServerSettings::new(Some(addrs[0].0), true);
for h in Rc::get_mut(&mut self.h).unwrap().iter_mut() {
h.server_settings(settings.clone());
}
let acceptor = match SslAcceptorBuilder::mozilla_intermediate( let acceptor = match SslAcceptorBuilder::mozilla_intermediate(
SslMethod::tls(), &identity.pkey, &identity.cert, &identity.chain) SslMethod::tls(), &identity.pkey, &identity.cert, &identity.chain)
@ -265,10 +263,8 @@ impl<H: HttpHandler> HttpServer<SslStream<TcpStream>, net::SocketAddr, H> {
for (srv, tcp) in addrs { for (srv, tcp) in addrs {
info!("Starting tls http server on {}", srv); info!("Starting tls http server on {}", srv);
let st = settings.clone();
let acc = acceptor.clone(); let acc = acceptor.clone();
ctx.add_stream(tcp.incoming().and_then(move |(stream, addr)| { ctx.add_stream(tcp.incoming().and_then(move |(stream, addr)| {
let st2 = st.clone();
SslAcceptorExt::accept_async(&acc, stream) SslAcceptorExt::accept_async(&acc, stream)
.map(move |stream| { .map(move |stream| {
let http2 = if let Some(p) = let http2 = if let Some(p) =
@ -278,8 +274,7 @@ impl<H: HttpHandler> HttpServer<SslStream<TcpStream>, net::SocketAddr, H> {
} else { } else {
false false
}; };
IoStream{settings: st2.clone(), IoStream{io: stream, peer: Some(addr), http2: http2}
io: stream, peer: Some(addr), http2: http2}
}) })
.map_err(|err| { .map_err(|err| {
trace!("Error during handling tls connection: {}", err); trace!("Error during handling tls connection: {}", err);
@ -292,26 +287,25 @@ impl<H: HttpHandler> HttpServer<SslStream<TcpStream>, net::SocketAddr, H> {
} }
} }
struct IoStream<T, H> { struct IoStream<T> {
io: T, io: T,
peer: Option<SocketAddr>, peer: Option<SocketAddr>,
http2: bool, http2: bool,
settings: ServerSettings<H>,
} }
impl<T, H> ResponseType for IoStream<T, H> impl<T> ResponseType for IoStream<T>
where T: AsyncRead + AsyncWrite + 'static where T: AsyncRead + AsyncWrite + 'static
{ {
type Item = (); type Item = ();
type Error = (); type Error = ();
} }
impl<T, A, H> StreamHandler<IoStream<T, H>, io::Error> for HttpServer<T, A, H> impl<T, A, H> StreamHandler<IoStream<T>, io::Error> for HttpServer<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, where T: AsyncRead + AsyncWrite + 'static,
H: HttpHandler + 'static, H: HttpHandler + 'static,
A: 'static {} A: 'static {}
impl<T, A, H> Handler<IoStream<T, H>, io::Error> for HttpServer<T, A, H> impl<T, A, H> Handler<IoStream<T>, io::Error> for HttpServer<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, where T: AsyncRead + AsyncWrite + 'static,
H: HttpHandler + 'static, H: HttpHandler + 'static,
A: 'static, A: 'static,
@ -320,11 +314,11 @@ impl<T, A, H> Handler<IoStream<T, H>, io::Error> for HttpServer<T, A, H>
debug!("Error handling request: {}", err) debug!("Error handling request: {}", err)
} }
fn handle(&mut self, msg: IoStream<T, H>, _: &mut Context<Self>) fn handle(&mut self, msg: IoStream<T>, _: &mut Context<Self>)
-> Response<Self, IoStream<T, H>> -> Response<Self, IoStream<T>>
{ {
Arbiter::handle().spawn( Arbiter::handle().spawn(
HttpChannel::new(msg.settings, msg.io, msg.peer, msg.http2)); HttpChannel::new(Rc::clone(&self.h), msg.io, msg.peer, msg.http2));
Self::empty() Self::empty()
} }
} }