use std::marker::PhantomData; use futures::{future::ok, future::FutureResult, Async, Future, Poll}; use openssl::ssl::{Error, SslAcceptor, SslConnector}; use tokio_io::{AsyncRead, AsyncWrite}; use tokio_openssl::{AcceptAsync, ConnectAsync, SslAcceptorExt, SslConnectorExt, SslStream}; use super::MAX_CONN_COUNTER; use counter::{Counter, CounterGuard}; use resolver::RequestHost; use service::{NewService, Service}; /// Support `SSL` connections via openssl package /// /// `ssl` feature enables `OpensslAcceptor` type pub struct OpensslAcceptor { acceptor: SslAcceptor, io: PhantomData, } impl OpensslAcceptor { /// Create default `OpensslAcceptor` pub fn new(acceptor: SslAcceptor) -> Self { OpensslAcceptor { acceptor, io: PhantomData, } } } impl Clone for OpensslAcceptor { fn clone(&self) -> Self { Self { acceptor: self.acceptor.clone(), io: PhantomData, } } } impl NewService for OpensslAcceptor { type Response = SslStream; type Error = Error; type Service = OpensslAcceptorService; type InitError = (); type Future = FutureResult; fn new_service(&self) -> Self::Future { MAX_CONN_COUNTER.with(|conns| { ok(OpensslAcceptorService { acceptor: self.acceptor.clone(), conns: conns.clone(), io: PhantomData, }) }) } } pub struct OpensslAcceptorService { acceptor: SslAcceptor, io: PhantomData, conns: Counter, } impl Service for OpensslAcceptorService { type Response = SslStream; type Error = Error; type Future = OpensslAcceptorServiceFut; fn poll_ready(&mut self) -> Poll<(), Self::Error> { if self.conns.available() { Ok(Async::Ready(())) } else { Ok(Async::NotReady) } } fn call(&mut self, req: T) -> Self::Future { OpensslAcceptorServiceFut { _guard: self.conns.get(), fut: SslAcceptorExt::accept_async(&self.acceptor, req), } } } pub struct OpensslAcceptorServiceFut where T: AsyncRead + AsyncWrite, { fut: AcceptAsync, _guard: CounterGuard, } impl Future for OpensslAcceptorServiceFut { type Item = SslStream; type Error = Error; fn poll(&mut self) -> Poll { self.fut.poll() } } /// Openssl connector factory pub struct OpensslConnector { connector: SslConnector, _t: PhantomData<(R, T, E)>, } impl OpensslConnector { pub fn new(connector: SslConnector) -> Self { OpensslConnector { connector, _t: PhantomData, } } } impl OpensslConnector { pub fn service( connector: SslConnector, ) -> impl Service<(R, T), Response = (R, SslStream), Error = Error> { OpensslConnectorService { connector: connector, _t: PhantomData, } } } impl Clone for OpensslConnector { fn clone(&self) -> Self { Self { connector: self.connector.clone(), _t: PhantomData, } } } impl NewService<(R, T)> for OpensslConnector { type Response = (R, SslStream); type Error = Error; type Service = OpensslConnectorService; type InitError = E; type Future = FutureResult; fn new_service(&self) -> Self::Future { ok(OpensslConnectorService { connector: self.connector.clone(), _t: PhantomData, }) } } pub struct OpensslConnectorService { connector: SslConnector, _t: PhantomData<(R, T)>, } impl Service<(R, T)> for OpensslConnectorService { type Response = (R, SslStream); type Error = Error; type Future = ConnectAsyncExt; fn poll_ready(&mut self) -> Poll<(), Self::Error> { Ok(Async::Ready(())) } fn call(&mut self, (req, stream): (R, T)) -> Self::Future { ConnectAsyncExt { fut: SslConnectorExt::connect_async(&self.connector, req.host(), stream), req: Some(req), } } } pub struct ConnectAsyncExt { req: Option, fut: ConnectAsync, } impl Future for ConnectAsyncExt where R: RequestHost, T: AsyncRead + AsyncWrite, { type Item = (R, SslStream); type Error = Error; fn poll(&mut self) -> Poll { match self.fut.poll()? { Async::Ready(stream) => Ok(Async::Ready((self.req.take().unwrap(), stream))), Async::NotReady => Ok(Async::NotReady), } } }