mirror of
https://github.com/fafhrd91/actix-net
synced 2025-06-26 19:47:43 +02:00
move rustls and nativetls acceptor services to actix-tls
This commit is contained in:
Nikolay Kim
@ -1,42 +0,0 @@
|
||||
//! SSL Services
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use crate::counter::Counter;
|
||||
|
||||
#[cfg(feature = "openssl")]
|
||||
mod openssl;
|
||||
#[cfg(feature = "openssl")]
|
||||
pub use self::openssl::OpensslAcceptor;
|
||||
|
||||
#[cfg(feature = "nativetls")]
|
||||
mod nativetls;
|
||||
#[cfg(feature = "nativetls")]
|
||||
pub use self::nativetls::NativeTlsAcceptor;
|
||||
|
||||
#[cfg(feature = "rustls")]
|
||||
mod rustls;
|
||||
#[cfg(feature = "rustls")]
|
||||
pub use self::rustls::RustlsAcceptor;
|
||||
|
||||
/// Sets the maximum per-worker concurrent ssl connection establish process.
|
||||
///
|
||||
/// All listeners will stop accepting connections when this limit is
|
||||
/// reached. It can be used to limit the global SSL CPU usage.
|
||||
///
|
||||
/// By default max connections is set to a 256.
|
||||
pub fn max_concurrent_ssl_connect(num: usize) {
|
||||
MAX_CONN.store(num, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
pub(crate) static MAX_CONN: AtomicUsize = AtomicUsize::new(256);
|
||||
|
||||
thread_local! {
|
||||
static MAX_CONN_COUNTER: Counter = Counter::new(MAX_CONN.load(Ordering::Relaxed));
|
||||
}
|
||||
|
||||
/// Ssl error combinded with service error.
|
||||
#[derive(Debug)]
|
||||
pub enum SslError<E1, E2> {
|
||||
Ssl(E1),
|
||||
Service(E2),
|
||||
}
|
||||
@ -1,122 +0,0 @@
|
||||
use std::convert::Infallible;
|
||||
use std::marker::PhantomData;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
use actix_codec::{AsyncRead, AsyncWrite};
|
||||
use actix_service::{Service, ServiceFactory};
|
||||
use futures::future::{self, FutureExt as _, LocalBoxFuture, TryFutureExt as _};
|
||||
use native_tls::Error;
|
||||
use tokio_tls::{TlsAcceptor, TlsStream};
|
||||
|
||||
use crate::counter::Counter;
|
||||
use crate::ssl::MAX_CONN_COUNTER;
|
||||
use crate::{Io, ServerConfig};
|
||||
|
||||
/// Support `SSL` connections via native-tls package
|
||||
///
|
||||
/// `tls` feature enables `NativeTlsAcceptor` type
|
||||
pub struct NativeTlsAcceptor<T, P = ()> {
|
||||
acceptor: TlsAcceptor,
|
||||
io: PhantomData<(T, P)>,
|
||||
}
|
||||
|
||||
impl<T, P> NativeTlsAcceptor<T, P>
|
||||
where
|
||||
T: AsyncRead + AsyncWrite + Unpin,
|
||||
{
|
||||
/// Create `NativeTlsAcceptor` instance
|
||||
#[inline]
|
||||
pub fn new(acceptor: TlsAcceptor) -> Self {
|
||||
NativeTlsAcceptor {
|
||||
acceptor,
|
||||
io: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P> Clone for NativeTlsAcceptor<T, P> {
|
||||
#[inline]
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
acceptor: self.acceptor.clone(),
|
||||
io: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P> ServiceFactory for NativeTlsAcceptor<T, P>
|
||||
where
|
||||
T: AsyncRead + AsyncWrite + Unpin + 'static,
|
||||
P: 'static,
|
||||
{
|
||||
type Request = Io<T, P>;
|
||||
type Response = Io<TlsStream<T>, P>;
|
||||
type Error = Error;
|
||||
|
||||
type Config = ServerConfig;
|
||||
type Service = NativeTlsAcceptorService<T, P>;
|
||||
type InitError = Infallible;
|
||||
type Future = future::Ready<Result<Self::Service, Self::InitError>>;
|
||||
|
||||
fn new_service(&self, cfg: &ServerConfig) -> Self::Future {
|
||||
cfg.set_secure();
|
||||
|
||||
MAX_CONN_COUNTER.with(|conns| {
|
||||
future::ok(NativeTlsAcceptorService {
|
||||
acceptor: self.acceptor.clone(),
|
||||
conns: conns.clone(),
|
||||
io: PhantomData,
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub struct NativeTlsAcceptorService<T, P> {
|
||||
acceptor: TlsAcceptor,
|
||||
io: PhantomData<(T, P)>,
|
||||
conns: Counter,
|
||||
}
|
||||
|
||||
impl<T, P> Clone for NativeTlsAcceptorService<T, P> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
acceptor: self.acceptor.clone(),
|
||||
io: PhantomData,
|
||||
conns: self.conns.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P> Service for NativeTlsAcceptorService<T, P>
|
||||
where
|
||||
T: AsyncRead + AsyncWrite + Unpin + 'static,
|
||||
P: 'static,
|
||||
{
|
||||
type Request = Io<T, P>;
|
||||
type Response = Io<TlsStream<T>, P>;
|
||||
type Error = Error;
|
||||
type Future = LocalBoxFuture<'static, Result<Io<TlsStream<T>, P>, Error>>;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
if self.conns.available(cx) {
|
||||
Poll::Ready(Ok(()))
|
||||
} else {
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
let guard = self.conns.get();
|
||||
let this = self.clone();
|
||||
let (io, params, proto) = req.into_parts();
|
||||
async move { this.acceptor.accept(io).await }
|
||||
.map_ok(move |stream| Io::from_parts(stream, params, proto))
|
||||
.map_ok(move |io| {
|
||||
// Required to preserve `CounterGuard` until `Self::Future`
|
||||
// is completely resolved.
|
||||
let _ = guard;
|
||||
io
|
||||
})
|
||||
.boxed_local()
|
||||
}
|
||||
}
|
||||
@ -1,124 +0,0 @@
|
||||
use std::future::Future;
|
||||
use std::io;
|
||||
use std::marker::PhantomData;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
use actix_codec::{AsyncRead, AsyncWrite};
|
||||
use actix_service::{Service, ServiceFactory};
|
||||
use futures::future::{ok, Ready};
|
||||
use rust_tls::ServerConfig;
|
||||
use tokio_rustls::{server::TlsStream, Accept, TlsAcceptor};
|
||||
|
||||
use crate::counter::{Counter, CounterGuard};
|
||||
use crate::ssl::MAX_CONN_COUNTER;
|
||||
use crate::{Io, Protocol, ServerConfig as SrvConfig};
|
||||
|
||||
/// Support `SSL` connections via rustls package
|
||||
///
|
||||
/// `rust-tls` feature enables `RustlsAcceptor` type
|
||||
pub struct RustlsAcceptor<T, P = ()> {
|
||||
config: Arc<ServerConfig>,
|
||||
io: PhantomData<(T, P)>,
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite, P> RustlsAcceptor<T, P> {
|
||||
/// Create `RustlsAcceptor` new service
|
||||
pub fn new(config: ServerConfig) -> Self {
|
||||
RustlsAcceptor {
|
||||
config: Arc::new(config),
|
||||
io: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P> Clone for RustlsAcceptor<T, P> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
config: self.config.clone(),
|
||||
io: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite + Unpin, P> ServiceFactory for RustlsAcceptor<T, P> {
|
||||
type Request = Io<T, P>;
|
||||
type Response = Io<TlsStream<T>, P>;
|
||||
type Error = io::Error;
|
||||
|
||||
type Config = SrvConfig;
|
||||
type Service = RustlsAcceptorService<T, P>;
|
||||
type InitError = ();
|
||||
type Future = Ready<Result<Self::Service, Self::InitError>>;
|
||||
|
||||
fn new_service(&self, cfg: &SrvConfig) -> Self::Future {
|
||||
cfg.set_secure();
|
||||
|
||||
MAX_CONN_COUNTER.with(|conns| {
|
||||
ok(RustlsAcceptorService {
|
||||
acceptor: self.config.clone().into(),
|
||||
conns: conns.clone(),
|
||||
io: PhantomData,
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub struct RustlsAcceptorService<T, P> {
|
||||
acceptor: TlsAcceptor,
|
||||
io: PhantomData<(T, P)>,
|
||||
conns: Counter,
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite + Unpin, P> Service for RustlsAcceptorService<T, P> {
|
||||
type Request = Io<T, P>;
|
||||
type Response = Io<TlsStream<T>, P>;
|
||||
type Error = io::Error;
|
||||
type Future = RustlsAcceptorServiceFut<T, P>;
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
|
||||
if self.conns.available(cx) {
|
||||
Poll::Ready(Ok(()))
|
||||
} else {
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
let (io, params, _) = req.into_parts();
|
||||
RustlsAcceptorServiceFut {
|
||||
_guard: self.conns.get(),
|
||||
fut: self.acceptor.accept(io),
|
||||
params: Some(params),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct RustlsAcceptorServiceFut<T, P>
|
||||
where
|
||||
T: AsyncRead + AsyncWrite + Unpin,
|
||||
{
|
||||
fut: Accept<T>,
|
||||
params: Option<P>,
|
||||
_guard: CounterGuard,
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite + Unpin, P> Unpin for RustlsAcceptorServiceFut<T, P> {}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite + Unpin, P> Future for RustlsAcceptorServiceFut<T, P> {
|
||||
type Output = Result<Io<TlsStream<T>, P>, io::Error>;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
|
||||
let this = self.get_mut();
|
||||
|
||||
let res = futures::ready!(Pin::new(&mut this.fut).poll(cx));
|
||||
match res {
|
||||
Ok(io) => {
|
||||
let params = this.params.take().unwrap();
|
||||
Poll::Ready(Ok(Io::from_parts(io, params, Protocol::Unknown)))
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user