use std::{cell::Cell, cell::RefCell, error, fmt, marker, rc::Rc, task::Context};
use crate::io::{Filter, Io, IoRef, types};
use crate::service::{IntoServiceFactory, Service, ServiceCtx, ServiceFactory};
use crate::{SharedCfg, channel::oneshot, util::HashSet, util::join};
use super::body::MessageBody;
use super::config::DispatcherConfig;
use super::error::{DispatchError, H2Error, ResponseError};
use super::request::Request;
use super::response::Response;
use super::{h1, h2};
#[derive(derive_more::Debug)]
#[debug("HttpService")]
pub struct HttpService<
F,
S,
B,
C1 = h1::DefaultControlService,
C2 = h2::DefaultControlService,
> {
srv: S,
h1_control: C1,
h2_control: Rc<C2>,
_t: marker::PhantomData<(F, B)>,
}
impl<F, S, B> HttpService<F, S, B>
where
F: Filter,
S: ServiceFactory<Request, SharedCfg> + 'static,
S::Error: ResponseError,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
{
pub fn new<U>(service: U) -> Self
where
U: IntoServiceFactory<S, Request, SharedCfg>,
{
HttpService {
srv: service.into_factory(),
h1_control: h1::DefaultControlService,
h2_control: Rc::new(h2::DefaultControlService),
_t: marker::PhantomData,
}
}
}
impl<F, S, B> HttpService<F, S, B>
where
F: Filter,
S: ServiceFactory<Request, SharedCfg> + 'static,
S::Error: ResponseError,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
{
pub fn h1<U: IntoServiceFactory<S, Request, SharedCfg>>(
service: U,
) -> h1::H1Service<F, S, B, h1::DefaultControlService> {
h1::H1Service::new(service)
}
pub fn h2<U: IntoServiceFactory<S, Request, SharedCfg>>(
service: U,
) -> h2::H2Service<F, S, B, h2::DefaultControlService> {
h2::H2Service::new(service)
}
}
impl<F, S, B, C1, C2> HttpService<F, S, B, C1, C2>
where
F: Filter,
S: ServiceFactory<Request, SharedCfg> + 'static,
S::Error: ResponseError,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
C1: ServiceFactory<h1::Control<F, S::Error>, SharedCfg, Response = h1::ControlAck<F>>,
C1::Error: error::Error,
C1::InitError: fmt::Debug,
C2: ServiceFactory<h2::Control<H2Error>, SharedCfg, Response = h2::ControlAck>,
C2::Error: error::Error,
C2::InitError: fmt::Debug,
{
pub fn h1_control<CT, U>(self, control: U) -> HttpService<F, S, B, CT, C2>
where
U: IntoServiceFactory<CT, h1::Control<F, S::Error>, SharedCfg>,
CT: ServiceFactory<
h1::Control<F, S::Error>,
SharedCfg,
Response = h1::ControlAck<F>,
>,
CT::Error: error::Error,
CT::InitError: fmt::Debug,
{
HttpService {
h1_control: control.into_factory(),
h2_control: self.h2_control,
srv: self.srv,
_t: marker::PhantomData,
}
}
pub fn h2_control<CT, U>(self, control: U) -> HttpService<F, S, B, C1, CT>
where
U: IntoServiceFactory<CT, h2::Control<H2Error>, SharedCfg>,
CT: ServiceFactory<h2::Control<H2Error>, SharedCfg, Response = h2::ControlAck>,
CT::Error: error::Error,
CT::InitError: fmt::Debug,
{
HttpService {
h1_control: self.h1_control,
h2_control: Rc::new(control.into_factory()),
srv: self.srv,
_t: marker::PhantomData,
}
}
}
#[cfg(feature = "openssl")]
#[allow(clippy::wildcard_imports)]
mod openssl {
use ntex_tls::openssl::{SslAcceptor, SslFilter};
use tls_openssl::ssl;
use super::*;
use crate::{io::Layer, server::SslError};
impl<F, S, B, C1, C2> HttpService<Layer<SslFilter, F>, S, B, C1, C2>
where
F: Filter,
S: ServiceFactory<Request, SharedCfg> + 'static,
S::Error: ResponseError,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
C1: ServiceFactory<
h1::Control<Layer<SslFilter, F>, S::Error>,
SharedCfg,
Response = h1::ControlAck<Layer<SslFilter, F>>,
> + 'static,
C1::Error: error::Error,
C1::InitError: fmt::Debug,
C2: ServiceFactory<h2::Control<H2Error>, SharedCfg, Response = h2::ControlAck>
+ 'static,
C2::Error: error::Error,
C2::InitError: fmt::Debug,
{
pub fn openssl(
self,
acceptor: ssl::SslAcceptor,
) -> impl ServiceFactory<
Io<F>,
SharedCfg,
Response = (),
Error = SslError<DispatchError>,
InitError = (),
> {
SslAcceptor::new(acceptor)
.map_err(SslError::Ssl)
.map_init_err(|()| unreachable!())
.and_then(self.map_err(SslError::Service))
}
}
}
#[cfg(feature = "rustls")]
#[allow(clippy::wildcard_imports)]
mod rustls {
use ntex_tls::rustls::{TlsAcceptor, TlsServerFilter};
use tls_rustls::ServerConfig;
use super::*;
use crate::{io::Layer, server::SslError};
impl<F, S, B, C1, C2> HttpService<Layer<TlsServerFilter, F>, S, B, C1, C2>
where
F: Filter,
S: ServiceFactory<Request, SharedCfg> + 'static,
S::Error: ResponseError,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
C1: ServiceFactory<
h1::Control<Layer<TlsServerFilter, F>, S::Error>,
SharedCfg,
Response = h1::ControlAck<Layer<TlsServerFilter, F>>,
> + 'static,
C1::Error: error::Error,
C1::InitError: fmt::Debug,
C2: ServiceFactory<h2::Control<H2Error>, SharedCfg, Response = h2::ControlAck>
+ 'static,
C2::Error: error::Error,
C2::InitError: fmt::Debug,
{
pub fn rustls(
self,
mut config: ServerConfig,
) -> impl ServiceFactory<
Io<F>,
SharedCfg,
Response = (),
Error = SslError<DispatchError>,
InitError = (),
> {
let protos = vec!["h2".to_string().into(), "http/1.1".to_string().into()];
config.alpn_protocols = protos;
TlsAcceptor::from(config)
.map_err(|e| SslError::Ssl(Box::new(e)))
.map_init_err(|()| unreachable!())
.and_then(self.map_err(SslError::Service))
}
}
}
impl<F, S, B, C1, C2> ServiceFactory<Io<F>, SharedCfg> for HttpService<F, S, B, C1, C2>
where
F: Filter,
S: ServiceFactory<Request, SharedCfg> + 'static,
S::Error: ResponseError,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
C1: ServiceFactory<h1::Control<F, S::Error>, SharedCfg, Response = h1::ControlAck<F>>
+ 'static,
C1::Error: error::Error,
C1::InitError: fmt::Debug,
C2: ServiceFactory<h2::Control<H2Error>, SharedCfg, Response = h2::ControlAck>
+ 'static,
C2::Error: error::Error,
C2::InitError: fmt::Debug,
{
type Response = ();
type Error = DispatchError;
type InitError = ();
type Service = HttpServiceHandler<F, S::Service, B, C1::Service, C2>;
async fn create(&self, cfg: SharedCfg) -> Result<Self::Service, Self::InitError> {
let service = self
.srv
.create(cfg.clone())
.await
.map_err(|e| log::error!("Cannot construct publish service: {e:?}"))?;
let control = self
.h1_control
.create(cfg.clone())
.await
.map_err(|e| log::error!("Cannot construct control service: {e:?}"))?;
let (tx, rx) = oneshot::channel();
let config = DispatcherConfig::new(cfg.get(), service, control);
Ok(HttpServiceHandler {
cfg,
config: Rc::new(config),
h2_control: self.h2_control.clone(),
inflight: RefCell::new(HashSet::default()),
rx: Cell::new(Some(rx)),
tx: Cell::new(Some(tx)),
_t: marker::PhantomData,
})
}
}
#[derive(derive_more::Debug)]
#[debug("HttpServiceHandler")]
pub struct HttpServiceHandler<F, S, B, C1, C2> {
cfg: SharedCfg,
config: Rc<DispatcherConfig<S, C1>>,
h2_control: Rc<C2>,
inflight: RefCell<HashSet<IoRef>>,
rx: Cell<Option<oneshot::Receiver<()>>>,
tx: Cell<Option<oneshot::Sender<()>>>,
_t: marker::PhantomData<(F, B)>,
}
impl<F, S, B, C1, C2> Service<Io<F>> for HttpServiceHandler<F, S, B, C1, C2>
where
F: Filter,
S: Service<Request> + 'static,
S::Error: ResponseError,
S::Response: Into<Response<B>>,
B: MessageBody,
C1: Service<h1::Control<F, S::Error>, Response = h1::ControlAck<F>> + 'static,
C1::Error: error::Error,
C2: ServiceFactory<h2::Control<H2Error>, SharedCfg, Response = h2::ControlAck>
+ 'static,
C2::Error: error::Error,
C2::InitError: fmt::Debug,
{
type Response = ();
type Error = DispatchError;
async fn ready(&self, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
let cfg = self.config.as_ref();
let (ready1, ready2) = join(cfg.control.ready(), cfg.service.ready()).await;
ready1.map_err(|e| {
log::error!("Http control service readiness error: {e:?}");
DispatchError::Control(Rc::new(e))
})?;
ready2.map_err(|e| {
log::error!("Http service readiness error: {e:?}");
DispatchError::Service(Rc::new(e))
})
}
#[inline]
fn poll(&self, cx: &mut Context<'_>) -> Result<(), Self::Error> {
let cfg = self.config.as_ref();
cfg.control
.poll(cx)
.map_err(|e| DispatchError::Control(Rc::new(e)))?;
cfg.service
.poll(cx)
.map_err(|e| DispatchError::Service(Rc::new(e)))
}
async fn shutdown(&self) {
self.config.shutdown();
let inflight = {
let inflight = self.inflight.borrow();
for io in inflight.iter() {
io.notify_dispatcher();
}
inflight.len()
};
if inflight != 0 {
log::trace!("Shutting down service, in-flight connections: {inflight}");
if let Some(rx) = self.rx.take() {
let _ = rx.await;
}
log::trace!("Shutting down is complected");
}
join(
self.config.control.shutdown(),
self.config.service.shutdown(),
)
.await;
}
async fn call(
&self,
io: Io<F>,
_: ServiceCtx<'_, Self>,
) -> Result<Self::Response, Self::Error> {
let id = self.config.next_id();
let ioref = io.get_ref();
let result = if io.query::<types::HttpProtocol>().get()
== Some(types::HttpProtocol::Http2)
{
let control = self
.h2_control
.create(self.cfg.clone())
.await
.map_err(|e| {
DispatchError::Control(crate::util::str_rc_error(format!(
"Cannot construct control service: {e:?}"
)))
})?;
let inflight = {
let mut inflight = self.inflight.borrow_mut();
inflight.insert(io.get_ref());
inflight.len()
};
log::trace!(
"New http2 connection {id}, peer address {:?}, in-flight: {inflight}",
io.query::<types::PeerAddr>().get(),
);
h2::handle(id, io.into(), control, self.config.clone()).await
} else {
let inflight = {
let mut inflight = self.inflight.borrow_mut();
inflight.insert(io.get_ref());
inflight.len()
};
log::trace!(
"New http1 connection {id}, peer address {:?}, in-flight: {inflight}",
io.query::<types::PeerAddr>().get(),
);
h1::handle_io(id, io, self.config.clone()).await
};
{
let mut inflight = self.inflight.borrow_mut();
inflight.remove(&ioref);
if inflight.is_empty()
&& let Some(tx) = self.tx.take()
{
let _ = tx.send(());
}
}
result
}
}