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_server_config::{Io, IoStream, ServerConfig as SrvConfig}; use actix_service::{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 super::dispatcher::Dispatcher; /// `ServiceFactory` implementation for HTTP2 transport pub struct H2Service { srv: S, cfg: ServiceConfig, on_connect: Option Box>>, _t: PhantomData<(T, P, B)>, } impl H2Service where S: ServiceFactory, S::Error: Into + Unpin + 'static, S::Response: Into> + Unpin + 'static, S::Future: Unpin, ::Future: Unpin + 'static, B: MessageBody + 'static, P: Unpin, { /// Create new `HttpService` instance. pub fn new>(service: F) -> Self { let cfg = ServiceConfig::new(KeepAlive::Timeout(5), 5000, 0); H2Service { cfg, on_connect: None, srv: service.into_factory(), _t: PhantomData, } } /// Create new `HttpService` instance with config. pub fn with_config>( 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 Box>>, ) -> Self { self.on_connect = f; self } } impl ServiceFactory for H2Service where T: IoStream, S: ServiceFactory, S::Error: Into + Unpin + 'static, S::Response: Into> + Unpin + 'static, S::Future: Unpin, ::Future: Unpin + 'static, B: MessageBody + 'static, P: Unpin, { type Config = SrvConfig; type Request = Io; type Response = (); type Error = DispatchError; type InitError = S::InitError; type Service = H2ServiceHandler; type Future = H2ServiceResponse; fn new_service(&self, cfg: &SrvConfig) -> Self::Future { H2ServiceResponse { fut: self.srv.new_service(cfg), cfg: Some(self.cfg.clone()), on_connect: self.on_connect.clone(), _t: PhantomData, } } } #[doc(hidden)] pub struct H2ServiceResponse { fut: S::Future, cfg: Option, on_connect: Option Box>>, _t: PhantomData<(T, P, B)>, } impl Future for H2ServiceResponse where T: IoStream, S: ServiceFactory, S::Error: Into + Unpin + 'static, S::Response: Into> + Unpin + 'static, S::Future: Unpin, ::Future: Unpin + 'static, B: MessageBody + 'static, P: Unpin, { type Output = Result, S::InitError>; fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll { let this = self.get_mut(); Poll::Ready(ready!(Pin::new(&mut this.fut).poll(cx)).map(|service| { H2ServiceHandler::new( this.cfg.take().unwrap(), this.on_connect.clone(), service, ) })) } } /// `Service` implementation for http/2 transport pub struct H2ServiceHandler { srv: CloneableService, cfg: ServiceConfig, on_connect: Option Box>>, _t: PhantomData<(T, P, B)>, } impl H2ServiceHandler where S: Service, S::Error: Into + Unpin + 'static, S::Future: Unpin + 'static, S::Response: Into> + Unpin + 'static, B: MessageBody + 'static, P: Unpin, { fn new( cfg: ServiceConfig, on_connect: Option Box>>, srv: S, ) -> H2ServiceHandler { H2ServiceHandler { cfg, on_connect, srv: CloneableService::new(srv), _t: PhantomData, } } } impl Service for H2ServiceHandler where T: IoStream, S: Service, S::Error: Into + Unpin + 'static, S::Future: Unpin + 'static, S::Response: Into> + Unpin + 'static, B: MessageBody + 'static, P: Unpin, { type Request = Io; type Response = (); type Error = DispatchError; type Future = H2ServiceHandlerResponse; fn poll_ready(&mut self, cx: &mut Context) -> Poll> { self.srv.poll_ready(cx).map_err(|e| { let e = e.into(); error!("Service readiness error: {:?}", e); DispatchError::Service(e) }) } fn call(&mut self, req: Self::Request) -> Self::Future { let io = req.into_parts().0; let peer_addr = io.peer_addr(); 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()), peer_addr, on_connect, server::handshake(io), ), } } } enum State, B: MessageBody> where S::Future: 'static, { Incoming(Dispatcher), Handshake( Option>, Option, Option, Option>, Handshake, ), } pub struct H2ServiceHandlerResponse where T: IoStream, S: Service, S::Error: Into + Unpin + 'static, S::Future: Unpin + 'static, S::Response: Into> + Unpin + 'static, B: MessageBody + 'static, { state: State, } impl Future for H2ServiceHandlerResponse where T: IoStream, S: Service, S::Error: Into + Unpin + 'static, S::Future: Unpin + 'static, S::Response: Into> + Unpin + 'static, B: MessageBody, { type Output = Result<(), DispatchError>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { 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, }, } } }