ntex 4.0.0-beta.17

use crate::error::IntoFailure;
use crate::http::error::{DispatchError, ResponseError};
use crate::http::{Request, Response, config::DispatcherConfig};
use crate::io::{Filter, Io, types};
use crate::service::pipeline::{Pipeline, PipelineFactory};
use crate::service::{Ctx, IntoServiceFactory, RequestState, Service, ServiceFactory};

use super::control::{Control, ControlAck, ControlResult};
use super::dispatcher::Dispatcher;

/// Server transport service that dispatches HTTP/1 requests.
///
/// Construct this service with [`HttpService::h1`](crate::http::HttpService::h1).
/// Request handling is delegated to the configured application service, while
/// connection lifecycle events get their default action unless a control
/// service is provided with [`control`](Self::control).
#[derive(derive_more::Debug)]
#[debug("H1Service")]
pub struct H1Service<F, Req: RequestState<Io<F>>, Err> {
    sf: crate::http::HttpPipeline<Req::State, Err>,
    /// Without a control service the default action is applied to every event.
    ctl: Option<crate::http::Ctl1Pipeline<Req::State, F, Err>>,
    config: DispatcherConfig,
}

impl<F, Req, Err> H1Service<F, Req, Err>
where
    F: Filter,
    Req: RequestState<Io<F>>,
    Req::State: Clone,
    Err: ResponseError + 'static,
{
    /// Create new `H1Service` instance.
    pub(crate) fn new<Sf>(sf: impl IntoServiceFactory<Sf, Req::State, Request>) -> Self
    where
        Sf: ServiceFactory<Req::State, Request, Error = Err> + 'static,
        Sf::Res: Into<Response>,
        Sf::InitError: IntoFailure,
    {
        H1Service {
            sf: PipelineFactory::new(
                sf.into_factory()
                    .map(Into::into)
                    .map_init_err(|e| DispatchError::Control(e.fail())),
            ),
            ctl: None,
            config: DispatcherConfig::default(),
        }
    }
}

impl<F, Req, Err> H1Service<F, Req, Err>
where
    F: Filter,
    Req: RequestState<Io<F>>,
    Req::State: Clone,
    Err: ResponseError + 'static,
{
    #[must_use]
    /// Provides the HTTP/1 control service.
    ///
    /// The control service receives connection, request, expectation, upgrade,
    /// and disconnect events. Service and protocol errors are reported by the
    /// disconnect event. Returning [`Control::ack`] applies the
    /// default action for each event.
    pub fn control<I, Sf>(self, ctl: I) -> Self
    where
        I: IntoServiceFactory<Sf, Req::State, Control<F, Err>>,
        Sf: ServiceFactory<Req::State, Control<F, Err>, Res = ControlAck<F>> + 'static,
        Sf::Error: IntoFailure,
        Sf::InitError: IntoFailure,
    {
        H1Service {
            sf: self.sf,
            ctl: Some(PipelineFactory::new(
                ctl.into_factory()
                    .map_err(|e| DispatchError::Service(e.fail()))
                    .map_init_err(|e| DispatchError::Control(e.fail())),
            )),
            config: self.config,
        }
    }
}

impl<St, F, Req, Err> Service<St, Req> for H1Service<F, Req, Err>
where
    F: Filter,
    Req: RequestState<Io<F>>,
    Req::State: Clone,
    Err: ResponseError + 'static,
{
    type Res = ();
    type Error = DispatchError;

    async fn call(&self, req: Req, _: Ctx<'_, Self, St>) -> Result<(), Self::Error> {
        let (st, io) = req.unpack();

        let svc = self.sf.create(st.clone()).await?;
        let ctl = match &self.ctl {
            Some(ctl) => Some(ctl.create(st).await?),
            None => None,
        };

        let id = self.config.next_id();
        let ioref = io.get_ref();
        let (_guard, inflight) = self.config.insert_io(&ioref);

        log::trace!(
            "{}: New http1 connection {id}, peer address {:?}, inflight: {}",
            io.tag(),
            io.query::<types::PeerAddr>().get(),
            inflight
        );

        handle_io(id, io, svc, ctl, self.config.clone()).await
    }

    async fn ready(&self, _: Ctx<'_, Self, St>) -> Result<(), Self::Error> {
        Ok(())
    }

    async fn shutdown(&self, _: crate::Ctx<'_, Self, St>) {
        // check inflight connections
        let inflight = self.config.shutdown();
        if inflight != 0 {
            log::trace!("Shutting down service, in-flight connections: {inflight}");

            self.config.wait_shutdown().await;
            log::trace!("Shutting down is complected");
        }
    }
}

pub(crate) async fn handle_io<F, Err>(
    id: usize,
    io: Io<F>,
    svc: Pipeline<Request, Response, Err>,
    ctl: Option<Pipeline<Control<F, Err>, ControlAck<F>, DispatchError>>,
    config: DispatcherConfig,
) -> Result<(), DispatchError>
where
    F: Filter,
    Err: ResponseError + 'static,
{
    // Notify control service
    let io = if let Some(ctl) = &ctl {
        let ack = ctl.call_nowait(Control::connect(id, io)).await?;
        let ControlResult::Connect(io) = ack.result else {
            unreachable!();
        };
        io
    } else {
        io
    };

    Dispatcher::new(id, io, svc, ctl, config).await
}