use std::{ future::Future, marker::PhantomData, mem, net, pin::Pin, rc::Rc, task::{Context, Poll}, }; use actix_codec::{AsyncRead, AsyncWrite}; use actix_rt::net::TcpStream; use actix_service::{ fn_factory, fn_service, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _, }; use actix_utils::future::ready; use futures_core::{future::LocalBoxFuture, ready}; use log::error; use crate::{ body::{BoxBody, MessageBody}, config::ServiceConfig, error::DispatchError, service::HttpFlow, ConnectCallback, OnConnectData, Request, Response, }; use super::{dispatcher::Dispatcher, handshake_with_timeout, HandshakeWithTimeout}; /// `ServiceFactory` implementation for HTTP/2 transport pub struct H2Service { srv: S, cfg: ServiceConfig, on_connect_ext: Option>>, _phantom: PhantomData<(T, B)>, } impl H2Service where S: ServiceFactory, S::Error: Into> + 'static, S::Response: Into> + 'static, >::Future: 'static, B: MessageBody + 'static, { /// Create new `H2Service` instance with config. pub(crate) fn with_config>( cfg: ServiceConfig, service: F, ) -> Self { H2Service { cfg, on_connect_ext: None, srv: service.into_factory(), _phantom: PhantomData, } } /// Set on connect callback. pub(crate) fn on_connect_ext(mut self, f: Option>>) -> Self { self.on_connect_ext = f; self } } impl H2Service where S: ServiceFactory, S::Future: 'static, S::Error: Into> + 'static, S::Response: Into> + 'static, >::Future: 'static, B: MessageBody + 'static, { /// Create plain TCP based service pub fn tcp( self, ) -> impl ServiceFactory< TcpStream, Config = (), Response = (), Error = DispatchError, InitError = S::InitError, > { fn_factory(|| { ready(Ok::<_, S::InitError>(fn_service(|io: TcpStream| { let peer_addr = io.peer_addr().ok(); ready(Ok::<_, DispatchError>((io, peer_addr))) }))) }) .and_then(self) } } #[cfg(feature = "openssl")] mod openssl { use actix_service::ServiceFactoryExt as _; use actix_tls::accept::{ openssl::{ reexports::{Error as SslError, SslAcceptor}, Acceptor, TlsStream, }, TlsError, }; use super::*; impl H2Service, S, B> where S: ServiceFactory, S::Future: 'static, S::Error: Into> + 'static, S::Response: Into> + 'static, >::Future: 'static, B: MessageBody + 'static, { /// Create OpenSSL based service. pub fn openssl( self, acceptor: SslAcceptor, ) -> impl ServiceFactory< TcpStream, Config = (), Response = (), Error = TlsError, InitError = S::InitError, > { Acceptor::new(acceptor) .map_init_err(|_| { unreachable!("TLS acceptor service factory does not error on init") }) .map_err(TlsError::into_service_error) .map(|io: TlsStream| { let peer_addr = io.get_ref().peer_addr().ok(); (io, peer_addr) }) .and_then(self.map_err(TlsError::Service)) } } } #[cfg(feature = "rustls")] mod rustls { use std::io; use actix_service::ServiceFactoryExt as _; use actix_tls::accept::{ rustls::{reexports::ServerConfig, Acceptor, TlsStream}, TlsError, }; use super::*; impl H2Service, S, B> where S: ServiceFactory, S::Future: 'static, S::Error: Into> + 'static, S::Response: Into> + 'static, >::Future: 'static, B: MessageBody + 'static, { /// Create Rustls based service. pub fn rustls( self, mut config: ServerConfig, ) -> impl ServiceFactory< TcpStream, Config = (), Response = (), Error = TlsError, InitError = S::InitError, > { let mut protos = vec![b"h2".to_vec()]; protos.extend_from_slice(&config.alpn_protocols); config.alpn_protocols = protos; Acceptor::new(config) .map_init_err(|_| { unreachable!("TLS acceptor service factory does not error on init") }) .map_err(TlsError::into_service_error) .map(|io: TlsStream| { let peer_addr = io.get_ref().0.peer_addr().ok(); (io, peer_addr) }) .and_then(self.map_err(TlsError::Service)) } } } impl ServiceFactory<(T, Option)> for H2Service where T: AsyncRead + AsyncWrite + Unpin + 'static, S: ServiceFactory, S::Future: 'static, S::Error: Into> + 'static, S::Response: Into> + 'static, >::Future: 'static, B: MessageBody + 'static, { type Response = (); type Error = DispatchError; type Config = (); type Service = H2ServiceHandler; type InitError = S::InitError; type Future = LocalBoxFuture<'static, Result>; fn new_service(&self, _: ()) -> Self::Future { let service = self.srv.new_service(()); let cfg = self.cfg.clone(); let on_connect_ext = self.on_connect_ext.clone(); Box::pin(async move { let service = service.await?; Ok(H2ServiceHandler::new(cfg, on_connect_ext, service)) }) } } /// `Service` implementation for HTTP/2 transport pub struct H2ServiceHandler where S: Service, { flow: Rc>, cfg: ServiceConfig, on_connect_ext: Option>>, _phantom: PhantomData, } impl H2ServiceHandler where S: Service, S::Error: Into> + 'static, S::Future: 'static, S::Response: Into> + 'static, B: MessageBody + 'static, { fn new( cfg: ServiceConfig, on_connect_ext: Option>>, service: S, ) -> H2ServiceHandler { H2ServiceHandler { flow: HttpFlow::new(service, (), None), cfg, on_connect_ext, _phantom: PhantomData, } } } impl Service<(T, Option)> for H2ServiceHandler where T: AsyncRead + AsyncWrite + Unpin, S: Service, S::Error: Into> + 'static, S::Future: 'static, S::Response: Into> + 'static, B: MessageBody + 'static, { type Response = (); type Error = DispatchError; type Future = H2ServiceHandlerResponse; fn poll_ready(&self, cx: &mut Context<'_>) -> Poll> { self.flow.service.poll_ready(cx).map_err(|err| { let err = err.into(); error!("Service readiness error: {:?}", err); DispatchError::Service(err) }) } fn call(&self, (io, addr): (T, Option)) -> Self::Future { let on_connect_data = OnConnectData::from_io(&io, self.on_connect_ext.as_deref()); H2ServiceHandlerResponse { state: State::Handshake( Some(self.flow.clone()), Some(self.cfg.clone()), addr, on_connect_data, handshake_with_timeout(io, &self.cfg), ), } } } enum State, B: MessageBody> where T: AsyncRead + AsyncWrite + Unpin, S::Future: 'static, { Handshake( Option>>, Option, Option, OnConnectData, HandshakeWithTimeout, ), Established(Dispatcher), } pub struct H2ServiceHandlerResponse where T: AsyncRead + AsyncWrite + Unpin, S: Service, S::Error: Into> + 'static, S::Future: 'static, S::Response: Into> + 'static, B: MessageBody + 'static, { state: State, } impl Future for H2ServiceHandlerResponse where T: AsyncRead + AsyncWrite + Unpin, S: Service, S::Error: Into> + 'static, S::Future: 'static, S::Response: Into> + 'static, B: MessageBody, { type Output = Result<(), DispatchError>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { match self.state { State::Handshake( ref mut srv, ref mut config, ref peer_addr, ref mut conn_data, ref mut handshake, ) => match ready!(Pin::new(handshake).poll(cx)) { Ok((conn, timer)) => { let on_connect_data = mem::take(conn_data); self.state = State::Established(Dispatcher::new( conn, srv.take().unwrap(), config.take().unwrap(), *peer_addr, on_connect_data, timer, )); self.poll(cx) } Err(err) => { log::trace!("H2 handshake error: {}", err); Poll::Ready(Err(err)) } }, State::Established(ref mut disp) => Pin::new(disp).poll(cx), } } }