ntex-service 5.3.0

use super::{Ctx, Service, ServiceFactory, util};

#[derive(Debug, Clone)]
/// Service produced by the `then` combinator.
///
/// This is created by the `ServiceChain::then()` method.
/// Children share the enclosing readiness scope unless explicitly wrapped with
/// [`crate::ready`]. Calls forward directly to each child; a wrapped child checks
/// its own readiness at call time, including after the first child's call awaits.
pub struct Then<A, B> {
    svc1: A,
    svc2: B,
}

impl<A, B> Then<A, B> {
    /// Creates a new `Then` service.
    pub(crate) fn new(svc1: A, svc2: B) -> Then<A, B> {
        Self { svc1, svc2 }
    }
}

impl<A, B, St, Req> Service<St, Req> for Then<A, B>
where
    A: Service<St, Req>,
    B: Service<St, Result<A::Res, A::Error>, Error = A::Error>,
{
    type Res = B::Res;
    type Error = B::Error;

    crate::forward_bind!(St, svc1, svc2);

    #[inline]
    async fn call(&self, req: Req, ctx: Ctx<'_, Self, St>) -> Result<B::Res, B::Error> {
        let result = self.svc1.call(req, ctx.cast()).await;
        self.svc2.call(result, ctx.without_bypass().cast()).await
    }

    #[inline]
    async fn ready(&self, ctx: Ctx<'_, Self, St>) -> Result<(), Self::Error> {
        util::ready(&self.svc1, &self.svc2, ctx).await
    }

    #[inline]
    async fn shutdown(&self, ctx: Ctx<'_, Self, St>) {
        util::shutdown(&self.svc1, &self.svc2, ctx).await;
    }
}

#[derive(Debug, Clone)]
/// Service factory produced by the `then` combinator.
///
/// This is created by the `ServiceChainFactory::then()` method.
pub struct ThenFactory<A, B> {
    svc1: A,
    svc2: B,
}

impl<A, B> ThenFactory<A, B> {
    /// Creates a new `ThenFactory`.
    pub(crate) fn new(svc1: A, svc2: B) -> Self {
        Self { svc1, svc2 }
    }
}

impl<A, B, St, Req> ServiceFactory<St, Req> for ThenFactory<A, B>
where
    A: ServiceFactory<St, Req>,
    B: ServiceFactory<St, Result<A::Res, A::Error>, Error = A::Error, InitError = A::InitError>,
{
    type Res = B::Res;
    type Error = A::Error;

    type Service = Then<A::Service, B::Service>;
    type InitError = A::InitError;

    async fn create(&self, st: &St) -> Result<Self::Service, Self::InitError> {
        Ok(Then::new(
            self.svc1.create(st).await?,
            self.svc2.create(st).await?,
        ))
    }
}

#[cfg(test)]
mod tests {
    use std::{cell::Cell, rc::Rc};

    use crate::{Ctx, Service, factory, fn_factory, service};

    #[derive(Clone)]
    struct Srv1(Rc<Cell<usize>>, Rc<Cell<usize>>);

    impl Service<(), Result<&'static str, &'static str>> for Srv1 {
        type Res = &'static str;
        type Error = ();

        async fn ready(&self, _: Ctx<'_, Self>) -> Result<(), Self::Error> {
            self.0.set(self.0.get() + 1);
            Ok(())
        }

        async fn call(
            &self,
            req: Result<&'static str, &'static str>,
            _: Ctx<'_, Self>,
        ) -> Result<&'static str, ()> {
            match req {
                Ok(msg) => Ok(msg),
                Err(_) => Err(()),
            }
        }

        async fn shutdown(&self, _: Ctx<'_, Self, ()>) {
            self.1.set(self.1.get() + 1);
        }
    }

    #[derive(Clone)]
    struct Srv2(Rc<Cell<usize>>, Rc<Cell<usize>>);

    impl Service<(), Result<&'static str, ()>> for Srv2 {
        type Res = (&'static str, &'static str);
        type Error = ();

        async fn ready(&self, _: Ctx<'_, Self>) -> Result<(), Self::Error> {
            self.0.set(self.0.get() + 1);
            Ok(())
        }

        async fn call(
            &self,
            req: Result<&'static str, ()>,
            _: Ctx<'_, Self>,
        ) -> Result<Self::Res, ()> {
            match req {
                Ok(msg) => Ok((msg, "ok")),
                Err(()) => Ok(("srv2", "err")),
            }
        }

        async fn shutdown(&self, _: Ctx<'_, Self, ()>) {
            self.1.set(self.1.get() + 1);
        }
    }

    #[ntex::test]
    async fn test_ready() {
        let cnt = Rc::new(Cell::new(0));
        let cnt_sht = Rc::new(Cell::new(0));
        let srv = service(Srv1(cnt.clone(), cnt_sht.clone()))
            .then(Srv2(cnt.clone(), cnt_sht.clone()))
            .pipeline(());
        let res = srv.ready().await;
        assert_eq!(res, Ok(()));
        assert_eq!(cnt.get(), 2);

        srv.shutdown().await;
        assert_eq!(cnt_sht.get(), 2);
    }

    #[ntex::test]
    async fn test_call() {
        let cnt = Rc::new(Cell::new(0));
        let srv = service(Srv1(cnt.clone(), Rc::new(Cell::new(0))))
            .then(Srv2(cnt, Rc::new(Cell::new(0))))
            .clone()
            .pipeline(());

        let res = srv.call(Ok("srv1")).await;
        assert!(res.is_ok());
        assert_eq!(res.unwrap(), ("srv1", "ok"));

        let res = srv.call(Err("srv")).await;
        assert!(res.is_ok());
        assert_eq!(res.unwrap(), ("srv2", "err"));
    }

    #[ntex::test]
    async fn test_factory() {
        let cnt = Rc::new(Cell::new(0));
        let checks = cnt.clone();
        let cnt2 = cnt.clone();
        let blank = fn_factory(move |(): &_| {
            let cnt = cnt2.clone();
            async move { Ok::<_, ()>(Srv1(cnt, Rc::new(Cell::new(0)))) }
        });
        let factory = factory(blank)
            .then(fn_factory(move |(): &()| {
                let cnt = cnt.clone();
                async move { Ok(Srv2(cnt.clone(), Rc::new(Cell::new(0)))) }
            }))
            .clone();
        let srv = factory.pipeline(()).await.unwrap();
        let res = srv.call(Ok("srv1")).await;
        assert!(res.is_ok());
        assert_eq!(res.unwrap(), ("srv1", "ok"));

        let res = srv.call(Err("srv")).await;
        assert!(res.is_ok());
        assert_eq!(res.unwrap(), ("srv2", "err"));
        assert_eq!(checks.get(), 2);
        let other = factory.pipeline(()).await.unwrap();
        assert!(other.call(Ok("fresh")).await.is_ok());
        assert_eq!(checks.get(), 4);
    }

    #[ntex::test]
    async fn concurrent_second_stage_entry() {
        use ntex::channel::oneshot::Receiver;

        use crate::util::tests::{Req, Single, State, concurrent_entry};

        let st = Rc::new(State::default());
        let srv = service(crate::fn_service(async |(rx1, rx2): Req| {
            let _ = rx1.await;
            Ok::<_, ()>(rx2)
        }))
        .then(crate::apply_fn(
            crate::ready(Single(st.clone())),
            async |result: Result<Receiver<()>, ()>, ctx| ctx.call(result?).await,
        ))
        .pipeline(());
        concurrent_entry(&srv, &st).await;
    }
}