mirror of
https://github.com/fafhrd91/actix-web
synced 2024-11-25 00:43:00 +01:00
391 lines
11 KiB
Rust
391 lines
11 KiB
Rust
use std::fmt::Debug;
|
|
use std::future::Future;
|
|
use std::marker::PhantomData;
|
|
use std::pin::Pin;
|
|
use std::task::{Context, Poll};
|
|
use std::{io, net, rc};
|
|
|
|
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
|
use actix_rt::net::TcpStream;
|
|
use actix_service::{
|
|
factory_fn, pipeline_factory, service_fn2, IntoServiceFactory, Service,
|
|
ServiceFactory,
|
|
};
|
|
use bytes::Bytes;
|
|
use futures::future::{ok, Ready};
|
|
use futures::{ready, Stream};
|
|
use h2::server::{self, Connection, Handshake};
|
|
use h2::RecvStream;
|
|
use log::error;
|
|
|
|
use crate::body::MessageBody;
|
|
use crate::cloneable::CloneableService;
|
|
use crate::config::{KeepAlive, ServiceConfig};
|
|
use crate::error::{DispatchError, Error, ParseError, ResponseError};
|
|
use crate::helpers::DataFactory;
|
|
use crate::payload::Payload;
|
|
use crate::request::Request;
|
|
use crate::response::Response;
|
|
use crate::Protocol;
|
|
|
|
use super::dispatcher::Dispatcher;
|
|
|
|
/// `ServiceFactory` implementation for HTTP2 transport
|
|
pub struct H2Service<T, S, B> {
|
|
srv: S,
|
|
cfg: ServiceConfig,
|
|
on_connect: Option<rc::Rc<dyn Fn(&T) -> Box<dyn DataFactory>>>,
|
|
_t: PhantomData<(T, B)>,
|
|
}
|
|
|
|
impl<T, S, B> H2Service<T, S, B>
|
|
where
|
|
S: ServiceFactory<Config = (), Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
<S::Service as Service>::Future: 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
/// Create new `HttpService` instance with config.
|
|
pub(crate) fn with_config<F: IntoServiceFactory<S>>(
|
|
cfg: ServiceConfig,
|
|
service: F,
|
|
) -> Self {
|
|
H2Service {
|
|
cfg,
|
|
on_connect: None,
|
|
srv: service.into_factory(),
|
|
_t: PhantomData,
|
|
}
|
|
}
|
|
|
|
/// Set on connect callback.
|
|
pub(crate) fn on_connect(
|
|
mut self,
|
|
f: Option<rc::Rc<dyn Fn(&T) -> Box<dyn DataFactory>>>,
|
|
) -> Self {
|
|
self.on_connect = f;
|
|
self
|
|
}
|
|
}
|
|
|
|
impl<S, B> H2Service<TcpStream, S, B>
|
|
where
|
|
S: ServiceFactory<Config = (), Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
<S::Service as Service>::Future: 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
/// Create simple tcp based service
|
|
pub fn tcp(
|
|
self,
|
|
) -> impl ServiceFactory<
|
|
Config = (),
|
|
Request = TcpStream,
|
|
Response = (),
|
|
Error = DispatchError,
|
|
InitError = S::InitError,
|
|
> {
|
|
pipeline_factory(factory_fn(|| {
|
|
async {
|
|
Ok::<_, S::InitError>(service_fn2(|io: TcpStream| {
|
|
let peer_addr = io.peer_addr().ok();
|
|
ok::<_, DispatchError>((io, peer_addr))
|
|
}))
|
|
}
|
|
}))
|
|
.and_then(self)
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "openssl")]
|
|
mod openssl {
|
|
use actix_service::{factory_fn, service_fn2};
|
|
use actix_tls::openssl::{Acceptor, SslAcceptor, SslStream};
|
|
use actix_tls::{openssl::HandshakeError, SslError};
|
|
|
|
use super::*;
|
|
|
|
impl<S, B> H2Service<SslStream<TcpStream>, S, B>
|
|
where
|
|
S: ServiceFactory<Config = (), Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
<S::Service as Service>::Future: 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
/// Create ssl based service
|
|
pub fn openssl(
|
|
self,
|
|
acceptor: SslAcceptor,
|
|
) -> impl ServiceFactory<
|
|
Config = (),
|
|
Request = TcpStream,
|
|
Response = (),
|
|
Error = SslError<HandshakeError<TcpStream>, DispatchError>,
|
|
InitError = S::InitError,
|
|
> {
|
|
pipeline_factory(
|
|
Acceptor::new(acceptor)
|
|
.map_err(SslError::Ssl)
|
|
.map_init_err(|_| panic!()),
|
|
)
|
|
.and_then(factory_fn(|| {
|
|
ok::<_, S::InitError>(service_fn2(|io: SslStream<TcpStream>| {
|
|
let peer_addr = io.get_ref().peer_addr().ok();
|
|
ok((io, peer_addr))
|
|
}))
|
|
}))
|
|
.and_then(self.map_err(SslError::Service))
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "rustls")]
|
|
mod rustls {
|
|
use super::*;
|
|
use actix_tls::rustls::{Acceptor, ServerConfig, Session, TlsStream};
|
|
use actix_tls::SslError;
|
|
use std::{fmt, io};
|
|
|
|
impl<S, B> H2Service<TlsStream<TcpStream>, S, B>
|
|
where
|
|
S: ServiceFactory<Config = (), Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
<S::Service as Service>::Future: 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
/// Create openssl based service
|
|
pub fn rustls(
|
|
self,
|
|
mut config: ServerConfig,
|
|
) -> impl ServiceFactory<
|
|
Config = (),
|
|
Request = TcpStream,
|
|
Response = (),
|
|
Error = SslError<io::Error, DispatchError>,
|
|
InitError = S::InitError,
|
|
> {
|
|
let protos = vec!["h2".to_string().into()];
|
|
config.set_protocols(&protos);
|
|
|
|
pipeline_factory(
|
|
Acceptor::new(config)
|
|
.map_err(SslError::Ssl)
|
|
.map_init_err(|_| panic!()),
|
|
)
|
|
.and_then(factory_fn(|| {
|
|
ok::<_, S::InitError>(service_fn2(|io: TlsStream<TcpStream>| {
|
|
let peer_addr = io.get_ref().0.peer_addr().ok();
|
|
ok((io, peer_addr))
|
|
}))
|
|
}))
|
|
.and_then(self.map_err(SslError::Service))
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T, S, B> ServiceFactory for H2Service<T, S, B>
|
|
where
|
|
T: AsyncRead + AsyncWrite + Unpin,
|
|
S: ServiceFactory<Config = (), Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
<S::Service as Service>::Future: 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
type Config = ();
|
|
type Request = (T, Option<net::SocketAddr>);
|
|
type Response = ();
|
|
type Error = DispatchError;
|
|
type InitError = S::InitError;
|
|
type Service = H2ServiceHandler<T, S::Service, B>;
|
|
type Future = H2ServiceResponse<T, S, B>;
|
|
|
|
fn new_service(&self, _: ()) -> Self::Future {
|
|
H2ServiceResponse {
|
|
fut: self.srv.new_service(()),
|
|
cfg: Some(self.cfg.clone()),
|
|
on_connect: self.on_connect.clone(),
|
|
_t: PhantomData,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[doc(hidden)]
|
|
#[pin_project::pin_project]
|
|
pub struct H2ServiceResponse<T, S: ServiceFactory, B> {
|
|
#[pin]
|
|
fut: S::Future,
|
|
cfg: Option<ServiceConfig>,
|
|
on_connect: Option<rc::Rc<dyn Fn(&T) -> Box<dyn DataFactory>>>,
|
|
_t: PhantomData<(T, B)>,
|
|
}
|
|
|
|
impl<T, S, B> Future for H2ServiceResponse<T, S, B>
|
|
where
|
|
T: AsyncRead + AsyncWrite + Unpin,
|
|
S: ServiceFactory<Config = (), Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
<S::Service as Service>::Future: 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
type Output = Result<H2ServiceHandler<T, S::Service, B>, S::InitError>;
|
|
|
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
|
let this = self.as_mut().project();
|
|
|
|
Poll::Ready(ready!(this.fut.poll(cx)).map(|service| {
|
|
let this = self.as_mut().project();
|
|
H2ServiceHandler::new(
|
|
this.cfg.take().unwrap(),
|
|
this.on_connect.clone(),
|
|
service,
|
|
)
|
|
}))
|
|
}
|
|
}
|
|
|
|
/// `Service` implementation for http/2 transport
|
|
pub struct H2ServiceHandler<T, S, B> {
|
|
srv: CloneableService<S>,
|
|
cfg: ServiceConfig,
|
|
on_connect: Option<rc::Rc<dyn Fn(&T) -> Box<dyn DataFactory>>>,
|
|
_t: PhantomData<(T, B)>,
|
|
}
|
|
|
|
impl<T, S, B> H2ServiceHandler<T, S, B>
|
|
where
|
|
S: Service<Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Future: 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
fn new(
|
|
cfg: ServiceConfig,
|
|
on_connect: Option<rc::Rc<dyn Fn(&T) -> Box<dyn DataFactory>>>,
|
|
srv: S,
|
|
) -> H2ServiceHandler<T, S, B> {
|
|
H2ServiceHandler {
|
|
cfg,
|
|
on_connect,
|
|
srv: CloneableService::new(srv),
|
|
_t: PhantomData,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T, S, B> Service for H2ServiceHandler<T, S, B>
|
|
where
|
|
T: AsyncRead + AsyncWrite + Unpin,
|
|
S: Service<Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Future: 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
type Request = (T, Option<net::SocketAddr>);
|
|
type Response = ();
|
|
type Error = DispatchError;
|
|
type Future = H2ServiceHandlerResponse<T, S, B>;
|
|
|
|
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
|
self.srv.poll_ready(cx).map_err(|e| {
|
|
let e = e.into();
|
|
error!("Service readiness error: {:?}", e);
|
|
DispatchError::Service(e)
|
|
})
|
|
}
|
|
|
|
fn call(&mut self, (io, addr): Self::Request) -> Self::Future {
|
|
let on_connect = if let Some(ref on_connect) = self.on_connect {
|
|
Some(on_connect(&io))
|
|
} else {
|
|
None
|
|
};
|
|
|
|
H2ServiceHandlerResponse {
|
|
state: State::Handshake(
|
|
Some(self.srv.clone()),
|
|
Some(self.cfg.clone()),
|
|
addr,
|
|
on_connect,
|
|
server::handshake(io),
|
|
),
|
|
}
|
|
}
|
|
}
|
|
|
|
enum State<T, S: Service<Request = Request>, B: MessageBody>
|
|
where
|
|
T: AsyncRead + AsyncWrite + Unpin,
|
|
S::Future: 'static,
|
|
{
|
|
Incoming(Dispatcher<T, S, B>),
|
|
Handshake(
|
|
Option<CloneableService<S>>,
|
|
Option<ServiceConfig>,
|
|
Option<net::SocketAddr>,
|
|
Option<Box<dyn DataFactory>>,
|
|
Handshake<T, Bytes>,
|
|
),
|
|
}
|
|
|
|
pub struct H2ServiceHandlerResponse<T, S, B>
|
|
where
|
|
T: AsyncRead + AsyncWrite + Unpin,
|
|
S: Service<Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Future: 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
B: MessageBody + 'static,
|
|
{
|
|
state: State<T, S, B>,
|
|
}
|
|
|
|
impl<T, S, B> Future for H2ServiceHandlerResponse<T, S, B>
|
|
where
|
|
T: AsyncRead + AsyncWrite + Unpin,
|
|
S: Service<Request = Request>,
|
|
S::Error: Into<Error> + 'static,
|
|
S::Future: 'static,
|
|
S::Response: Into<Response<B>> + 'static,
|
|
B: MessageBody,
|
|
{
|
|
type Output = Result<(), DispatchError>;
|
|
|
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
|
match self.state {
|
|
State::Incoming(ref mut disp) => Pin::new(disp).poll(cx),
|
|
State::Handshake(
|
|
ref mut srv,
|
|
ref mut config,
|
|
ref peer_addr,
|
|
ref mut on_connect,
|
|
ref mut handshake,
|
|
) => match Pin::new(handshake).poll(cx) {
|
|
Poll::Ready(Ok(conn)) => {
|
|
self.state = State::Incoming(Dispatcher::new(
|
|
srv.take().unwrap(),
|
|
conn,
|
|
on_connect.take(),
|
|
config.take().unwrap(),
|
|
None,
|
|
*peer_addr,
|
|
));
|
|
self.poll(cx)
|
|
}
|
|
Poll::Ready(Err(err)) => {
|
|
trace!("H2 handshake error: {}", err);
|
|
Poll::Ready(Err(err.into()))
|
|
}
|
|
Poll::Pending => Poll::Pending,
|
|
},
|
|
}
|
|
}
|
|
}
|