Skip to main content

ntex_service/
lib.rs

1//! Asynchronous services, factories, middleware, and execution pipelines.
2//!
3//! The [`Service`] trait is the crate's central abstraction. A service
4//! asynchronously transforms a request into a response, while
5//! [`ServiceFactory`] constructs services and [`Pipeline`] manages readiness,
6//! calls, and shutdown.
7#![deny(clippy::pedantic)]
8#![allow(
9    clippy::cast_possible_truncation,
10    clippy::missing_fields_in_debug,
11    clippy::missing_errors_doc,
12    clippy::missing_panics_doc,
13    clippy::must_use_candidate,
14    clippy::type_complexity,
15    clippy::unused_async,
16    clippy::unused_async_trait_impl
17)]
18use std::rc::Rc;
19
20mod and_then;
21mod apply;
22mod bind;
23pub mod boxed;
24pub mod cfg;
25mod chain;
26mod ctx;
27mod fn_ready;
28mod fn_service;
29mod fn_shutdown;
30mod macros;
31mod map;
32mod map_err;
33mod map_init_err;
34mod map_state;
35mod middleware;
36mod readiness;
37mod ready;
38mod service_binding;
39pub mod state;
40mod then;
41mod util;
42
43pub mod pipeline;
44mod pl_factory;
45mod pl_inner;
46mod pl_state;
47
48pub use crate::apply::{apply_fn, apply_fn_factory};
49pub use crate::bind::BindCtx;
50pub use crate::chain::{ServiceChain, ServiceChainFactory, factory, service};
51pub use crate::ctx::Ctx;
52pub use crate::fn_service::{fn_factory, fn_service, fn_service_st};
53pub use crate::map_state::{map_state, map_state_factory};
54pub use crate::middleware::{Identity, Middleware, Stack, apply, fn_layer};
55pub use crate::pipeline::Pipeline;
56pub use crate::readiness::{ReadyCallback, ReadyNotifier};
57pub use crate::ready::{ready, ready_factory};
58pub use crate::service_binding::ServiceBinding;
59pub use crate::state::{RequestState, State};
60
61#[allow(unused_variables)]
62/// An asynchronous operation from a request to a response.
63///
64/// A service receives requests and asynchronously produces responses.
65/// Conceptually, it is similar to:
66///
67/// ```rust,ignore
68/// async fn(Request) -> Result<Response, Error>
69/// ```
70///
71/// The request and pipeline-state types are generic parameters. The response
72/// and error types are associated types, allowing one service type to implement
73/// `Service` for multiple request types.
74///
75/// Methods take `&self`, so implementations that mutate internal state must use
76/// interior mutability such as `Cell`, `RefCell`, or a synchronization
77/// primitive when appropriate.
78///
79/// The same abstraction can represent client- and server-side operations.
80/// Services focus on transformation, making them straightforward to test and
81/// compose.
82///
83/// A service call requires a [`Ctx`] and therefore runs through a [`Pipeline`]
84/// or from another service. The pipeline coordinates readiness across a
85/// composed service chain before dispatching a request.
86///
87/// ```rust
88/// # use std::convert::Infallible;
89/// #
90/// # use ntex_service::{Service, Ctx};
91///
92/// struct MyService;
93///
94/// impl Service<(), u8> for MyService {
95///     type Res = u64;
96///     type Error = Infallible;
97///
98///     async fn call(&self, req: u8, ctx: Ctx<'_, Self>) -> Result<Self::Res, Self::Error> {
99///         Ok(req as u64)
100///     }
101/// }
102/// ```
103///
104/// Simple services do not need a manual trait implementation. The example
105/// above can be expressed with [`fn_service`]:
106///
107/// ```rust
108/// # use std::convert::Infallible;
109/// # use ntex_service::{Pipeline, fn_service};
110/// #
111/// # async fn run() -> Result<(), Infallible> {
112/// let service = fn_service(|req: u8| async move {
113///     Ok::<_, Infallible>(u64::from(req))
114/// });
115/// let pipeline = Pipeline::new((), service);
116///
117/// assert_eq!(pipeline.call(10).await?, 10);
118/// # Ok(())
119/// # }
120/// ```
121pub trait Service<St, Req> {
122    /// Response produced by the service.
123    type Res;
124
125    /// Error produced while checking readiness or processing a request.
126    type Error;
127
128    /// Processes a request and asynchronously returns the response.
129    ///
130    /// The enclosing pipeline checks readiness before invoking this method.
131    /// Implementations should not call their own `ready` method. A composed
132    /// service can use [`Ctx::call`] to call an inner service using the closest
133    /// upstream readiness cache. A service wrapped with [`ready()`] can be
134    /// called directly to check its own cache instead.
135    async fn call(&self, req: Req, ctx: Ctx<'_, Self, St>) -> Result<Self::Res, Self::Error>;
136
137    #[inline]
138    /// Initializes the service within a [`Pipeline`].
139    ///
140    /// Called during pipeline construction or [`Ctx::create`], before runtime
141    /// operations on the new service.
142    ///
143    /// The default is suitable for leaves without notification sources.
144    /// Composites must bind every existing child, and services with changing
145    /// readiness must subscribe to all their sources here. Lazily created
146    /// children must be initialized through [`Ctx::create`], which binds them
147    /// explicitly before use.
148    /// Register reusable [`ReadyCallback`] handles with [`BindCtx::on_notify`]
149    /// to observe notifications reaching the service's scope.
150    ///
151    /// Binding is synchronous. Shared services can be bound to multiple parents;
152    /// implementations must support repeated binding.
153    /// [`forward_bind!`] and [`forward_pl_bind!`] handle simple forwarding.
154    /// Ordinary readiness checks, calls, and shutdown do not add dependencies.
155    fn bind(&self, ctx: BindCtx<'_>) {}
156
157    #[inline]
158    /// Waits until the service is ready to process a request.
159    ///
160    /// If the service is at capacity, the returned future remains pending until
161    /// capacity becomes available.
162    ///
163    /// Pipeline readiness is coordinated across all services in a composed
164    /// chain. A request is dispatched only when the chain is ready.
165    ///
166    /// A [`Pipeline`] or [`ServiceBinding`] caches success until a subscribed
167    /// [`ReadyNotifier`] invalidates it; calls do not consume that success.
168    /// Subscribe in [`Self::bind`] and notify on every readiness-changing
169    /// transition, including consuming or releasing capacity and external
170    /// failures, before another caller can use stale cached readiness.
171    /// A task wake schedules a pending check but does not invalidate a cached
172    /// success. Services must not rely on this method being polled on every
173    /// call to drive background progress.
174    ///
175    /// Readiness checks and their wake sources run on the pipeline's thread.
176    /// This local-only requirement does not apply to task wakers supplied to
177    /// [`Self::call`] or [`Self::shutdown`].
178    ///
179    /// Readiness may inspect [`Ctx::st`], but it describes the shared gate, not
180    /// a reservation for one request or state value. State-taking pipeline
181    /// handles share the same cache; independent gates need separate pipelines
182    /// or service bindings.
183    async fn ready(&self, ctx: Ctx<'_, Self, St>) -> Result<(), Self::Error> {
184        Ok(())
185    }
186
187    #[inline]
188    /// Shuts down the service.
189    ///
190    /// Returns when the service has been properly shut down.
191    async fn shutdown(&self, ctx: Ctx<'_, Self, St>) {}
192
193    #[inline]
194    /// Maps this service's output to a different type, returning a new service.
195    ///
196    /// This is similar to `Option::map` or `Iterator::map`, changing the
197    /// output type of the underlying service.
198    ///
199    /// This function consumes the original service and returns a wrapped version,
200    /// following the pattern of standard library `map` methods.
201    fn map<F, Res>(self, f: F) -> ServiceChain<dev::Map<F, Self, Res>, St, Req>
202    where
203        Self: Sized,
204        F: Fn(Self::Res) -> Res,
205    {
206        service(dev::Map::new(f, self))
207    }
208
209    #[inline]
210    /// Maps this service's error to a different type, returning a new service.
211    ///
212    /// This is similar to `Result::map_err`, changing the error type of the
213    /// underlying service. It is useful, for example, to ensure multiple
214    /// services have the same error type.
215    ///
216    /// This function consumes the original service and returns a wrapped version.
217    fn map_err<F, E>(self, f: F) -> ServiceChain<dev::MapErr<F, Self, E>, St, Req>
218    where
219        Self: Sized,
220        F: Fn(Self::Error) -> E,
221    {
222        service(dev::MapErr::new(f, self))
223    }
224
225    #[inline]
226    /// Calls another service after this service completes successfully.
227    ///
228    /// The first service's response becomes the second service's request. If
229    /// the first service returns an error, the second service is not called.
230    fn and_then<Next, F>(self, f: F) -> ServiceChain<dev::AndThen<Self, Next>, St, Req>
231    where
232        Self: Sized,
233        Next: Service<St, Self::Res, Error = Self::Error>,
234        F: IntoService<Next, St, Self::Res>,
235    {
236        service(dev::AndThen::new(self, f.into_service()))
237    }
238
239    #[inline]
240    /// Wraps this service and its state in a [`Pipeline`].
241    fn pipeline(self, st: St) -> Pipeline<Req, Self::Res, Self::Error>
242    where
243        Self: Sized + 'static,
244        St: 'static,
245        Req: 'static,
246    {
247        Pipeline::new(st, self)
248    }
249}
250
251/// A factory for asynchronously creating [`Service`] values.
252///
253/// This is useful when new `Service`s must be produced dynamically. For example,
254/// a TCP server listener accepts new connections, constructs a new `Service` for
255/// each connection using the `ServiceFactory` trait, and uses that service to
256/// handle inbound requests.
257///
258/// `St` is the state type shared by the factory and its services.
259///
260/// Simple factories can often use [`fn_factory`] to reduce boilerplate.
261pub trait ServiceFactory<St, Req> {
262    /// Response produced by the created services.
263    type Res;
264
265    /// Error produced by the created services.
266    type Error;
267
268    /// The type of `Service` produced by this factory.
269    type Service: Service<St, Req, Res = Self::Res, Error = Self::Error>;
270
271    /// Error that can occur while constructing a service.
272    type InitError;
273
274    /// Asynchronously creates a service using the supplied state.
275    async fn create(&self, st: &St) -> Result<Self::Service, Self::InitError>;
276
277    #[inline]
278    /// Creates a service and wraps it with its state in a [`Pipeline`].
279    async fn pipeline(
280        &self,
281        st: St,
282    ) -> Result<Pipeline<Req, Self::Res, Self::Error>, Self::InitError>
283    where
284        Self: 'static,
285        St: 'static,
286        Req: 'static,
287    {
288        let svc = self.create(&st).await?;
289        Ok(Pipeline::new(st, svc))
290    }
291
292    #[inline]
293    /// Returns a factory whose services map responses to a different type.
294    fn map<F, Res>(self, f: F) -> ServiceChainFactory<dev::MapFactory<F, Self, Res>, St, Req>
295    where
296        Self: Sized,
297        F: Fn(Self::Res) -> Res + Clone,
298    {
299        factory(dev::MapFactory::new(f, self))
300    }
301
302    #[inline]
303    /// Returns a factory whose services map errors to a different type.
304    fn map_err<F, E>(self, f: F) -> ServiceChainFactory<dev::MapErrFactory<F, Self, E>, St, Req>
305    where
306        Self: Sized,
307        F: Fn(Self::Error) -> E + Clone,
308    {
309        factory(dev::MapErrFactory::new(f, self))
310    }
311
312    #[inline]
313    /// Maps this factory's initialization error to a different error,
314    /// returning a new service factory.
315    fn map_init_err<F, E>(self, f: F) -> ServiceChainFactory<dev::MapInitErr<F, Self, E>, St, Req>
316    where
317        Self: Sized,
318        F: Fn(Self::InitError) -> E + Clone,
319    {
320        factory(dev::MapInitErr::new(f, self))
321    }
322
323    /// Chains another factory after this factory's services.
324    ///
325    /// Each response from the first service becomes a request to the second
326    /// service. The second service is not called when the first returns an
327    /// error.
328    fn and_then<U, F>(self, f: F) -> ServiceChainFactory<dev::AndThenFactory<Self, U>, St, Req>
329    where
330        Self: Sized,
331        U: ServiceFactory<St, Self::Res, Error = Self::Error, InitError = Self::InitError>,
332        F: IntoServiceFactory<U, St, Self::Res>,
333    {
334        factory(dev::AndThenFactory::new(self, f.into_factory()))
335    }
336
337    /// Creates a boxed service factory.
338    fn boxed(self) -> boxed::BoxServiceFactory<St, Req, Self::Res, Self::Error, Self::InitError>
339    where
340        St: 'static,
341        Req: 'static,
342        Self: Sized + 'static,
343    {
344        boxed::factory(self)
345    }
346}
347
348impl<S, St, Req> Service<St, Req> for &S
349where
350    S: Service<St, Req>,
351{
352    type Res = S::Res;
353    type Error = S::Error;
354
355    fn bind(&self, ctx: BindCtx<'_>) {
356        ctx.bind::<_, St, Req>(&**self);
357    }
358
359    #[inline]
360    async fn ready(&self, ctx: Ctx<'_, Self, St>) -> Result<(), S::Error> {
361        Service::ready(&**self, ctx.cast()).await
362    }
363
364    #[inline]
365    async fn call(&self, req: Req, ctx: Ctx<'_, Self, St>) -> Result<S::Res, S::Error> {
366        Service::call(&**self, req, ctx.cast()).await
367    }
368
369    #[inline]
370    async fn shutdown(&self, ctx: Ctx<'_, Self, St>) {
371        Service::shutdown(&**self, ctx.cast()).await;
372    }
373}
374
375impl<S, St, Req> Service<St, Req> for Box<S>
376where
377    S: Service<St, Req>,
378{
379    type Res = S::Res;
380    type Error = S::Error;
381
382    fn bind(&self, ctx: BindCtx<'_>) {
383        ctx.bind::<_, St, Req>(&**self);
384    }
385
386    #[inline]
387    async fn ready(&self, ctx: Ctx<'_, Self, St>) -> Result<(), S::Error> {
388        Service::ready(&**self, ctx.cast()).await
389    }
390
391    #[inline]
392    async fn call(&self, req: Req, ctx: Ctx<'_, Self, St>) -> Result<S::Res, S::Error> {
393        Service::call(&**self, req, ctx.cast()).await
394    }
395
396    #[inline]
397    async fn shutdown(&self, ctx: Ctx<'_, Self, St>) {
398        Service::shutdown(&**self, ctx.cast()).await;
399    }
400}
401
402impl<S, St, Req> Service<St, Req> for Rc<S>
403where
404    S: Service<St, Req>,
405{
406    type Res = S::Res;
407    type Error = S::Error;
408
409    fn bind(&self, ctx: BindCtx<'_>) {
410        ctx.bind::<_, St, Req>(&**self);
411    }
412
413    #[inline]
414    async fn ready(&self, ctx: Ctx<'_, Self, St>) -> Result<(), S::Error> {
415        Service::ready(&**self, ctx.cast()).await
416    }
417
418    #[inline]
419    async fn call(&self, req: Req, ctx: Ctx<'_, Self, St>) -> Result<S::Res, S::Error> {
420        Service::call(&**self, req, ctx.cast()).await
421    }
422
423    #[inline]
424    async fn shutdown(&self, ctx: Ctx<'_, Self, St>) {
425        Service::shutdown(&**self, ctx.cast()).await;
426    }
427}
428
429impl<Sf, St, Req> ServiceFactory<St, Req> for Rc<Sf>
430where
431    Sf: ServiceFactory<St, Req>,
432{
433    type Res = Sf::Res;
434    type Error = Sf::Error;
435    type Service = Sf::Service;
436    type InitError = Sf::InitError;
437
438    async fn create(&self, st: &St) -> Result<Self::Service, Self::InitError> {
439        self.as_ref().create(st).await
440    }
441}
442
443/// A common interface for values that can call a service.
444pub trait ServiceCaller<Req, Res, Err> {
445    /// Waits for readiness, then calls the service.
446    async fn call_service(&self, req: Req) -> Result<Res, Err>;
447}
448
449/// Conversion into a [`Service`].
450pub trait IntoService<S, St, Req>
451where
452    S: Service<St, Req>,
453{
454    /// Converts this value into a service.
455    fn into_service(self) -> S;
456}
457
458/// Conversion into a [`ServiceFactory`].
459pub trait IntoServiceFactory<Sf, St, Req>
460where
461    Sf: ServiceFactory<St, Req>,
462{
463    /// Converts this value into a service factory.
464    fn into_factory(self) -> Sf;
465}
466
467impl<S, St, Req> IntoService<S, St, Req> for S
468where
469    S: Service<St, Req>,
470{
471    #[inline]
472    fn into_service(self) -> S {
473        self
474    }
475}
476
477impl<Sf, St, Req> IntoServiceFactory<Sf, St, Req> for Sf
478where
479    Sf: ServiceFactory<St, Req>,
480{
481    #[inline]
482    fn into_factory(self) -> Sf {
483        self
484    }
485}
486
487/// Combinator and helper types used by the public API.
488pub mod dev {
489    pub use crate::and_then::{AndThen, AndThenFactory};
490    pub use crate::apply::{Apply, ApplyCtx, ApplyFactory};
491    pub use crate::chain::{ServiceChain, ServiceChainFactory};
492    pub use crate::fn_ready::FnReadiness;
493    pub use crate::fn_service::{FnFactory, FnService, FnServiceSt, FnServiceStFactory};
494    pub use crate::fn_shutdown::FnShutdown;
495    pub use crate::map::{Map, MapFactory};
496    pub use crate::map_err::{MapErr, MapErrFactory};
497    pub use crate::map_init_err::MapInitErr;
498    pub use crate::map_state::{MapState, MapStateFactory};
499    pub use crate::middleware::{ApplyMiddleware, FnMiddleware};
500    pub use crate::ready::{Ready, ReadyFactory};
501    pub use crate::then::{Then, ThenFactory};
502}
503
504#[cfg(test)]
505mod tests {
506    use std::{cell::Cell, task::Poll};
507
508    use ntex::util::lazy;
509
510    use super::*;
511    use crate::dev::FnServiceSt;
512    use crate::pipeline::PipelineFactory;
513
514    #[derive(Clone, Debug, Default)]
515    struct Srv(Rc<Cell<usize>>);
516
517    impl Service<usize, usize> for Srv {
518        type Res = usize;
519        type Error = &'static str;
520
521        async fn ready(&self, _: Ctx<'_, Self, usize>) -> Result<(), Self::Error> {
522            self.0.set(self.0.get() + 1);
523            Ok(())
524        }
525
526        async fn call(&self, req: usize, ctx: Ctx<'_, Self, usize>) -> Result<usize, Self::Error> {
527            if req == 0 { Err("zero") } else { Ok(req + *ctx) }
528        }
529
530        async fn shutdown(&self, _: Ctx<'_, Self, usize>) {
531            self.0.set(self.0.get() + 100);
532        }
533    }
534
535    struct Fwd {
536        inner: Srv,
537        pl: Pipeline<usize, usize, &'static str>,
538    }
539
540    impl Service<usize, usize> for Fwd {
541        type Res = usize;
542        type Error = usize;
543
544        crate::forward_bind!(usize, inner);
545        crate::forward_ready!(usize, inner, str::len);
546
547        async fn call(&self, req: usize, ctx: Ctx<'_, Self, usize>) -> Result<usize, usize> {
548            let res = ctx.call(&self.inner, req).await.map_err(str::len)?;
549            self.pl.call(res).await.map_err(str::len)
550        }
551
552        crate::forward_shutdown!(usize, inner);
553    }
554
555    struct FwdPl {
556        pl: Pipeline<usize, usize, &'static str>,
557    }
558
559    impl Service<usize, usize> for FwdPl {
560        type Res = usize;
561        type Error = &'static str;
562
563        crate::forward_pl_bind!(usize, pl);
564        crate::forward_pl_ready!(usize, pl);
565        crate::forward_pl_shutdown!(usize, pl);
566
567        async fn call(&self, req: usize, _: Ctx<'_, Self, usize>) -> Result<usize, &'static str> {
568            self.pl.call(req).await
569        }
570    }
571
572    struct FwdPlErr {
573        pl: Pipeline<usize, usize, &'static str>,
574    }
575
576    impl Service<usize, usize> for FwdPlErr {
577        type Res = usize;
578        type Error = usize;
579
580        crate::forward_pl_bind!(usize, pl);
581        crate::forward_pl_ready!(usize, pl, str::len);
582
583        async fn call(&self, req: usize, _: Ctx<'_, Self, usize>) -> Result<usize, usize> {
584            self.pl.call(req).await.map_err(str::len)
585        }
586    }
587
588    #[ntex::test]
589    async fn service_wrappers() {
590        let cnt = Rc::new(Cell::new(0));
591
592        let srv: &'static Srv = Box::leak(Box::new(Srv(cnt.clone())));
593        let pl = Pipeline::new(1, srv);
594        assert_eq!(pl.call(1).await, Ok(2));
595        assert_eq!(pl.call(0).await, Err("zero"));
596        pl.shutdown().await;
597        assert_eq!(cnt.get(), 101);
598
599        let pl = Pipeline::new(2, Box::new(Srv(cnt.clone())));
600        assert_eq!(pl.call(1).await, Ok(3));
601        pl.shutdown().await;
602        assert_eq!(cnt.get(), 202);
603
604        let pl = Pipeline::new(3, Rc::new(Srv(cnt.clone())));
605        assert_eq!(pl.call(1).await, Ok(4));
606        pl.shutdown().await;
607        assert_eq!(cnt.get(), 303);
608
609        let pl = Pipeline::new(
610            1,
611            Fwd {
612                inner: Srv(cnt.clone()),
613                pl: Pipeline::new(10, Srv(cnt.clone())),
614            },
615        );
616        assert_eq!(pl.call(1).await, Ok(12));
617        assert_eq!(pl.call(0).await, Err(4));
618        pl.shutdown().await;
619        assert_eq!(cnt.get(), 405);
620
621        let pl = Pipeline::new(
622            1,
623            FwdPl {
624                pl: Pipeline::new(10, Srv(cnt.clone())),
625            },
626        );
627        assert_eq!(pl.call(1).await, Ok(11));
628        pl.shutdown().await;
629        // inner pipeline readiness is checked once, by `forward_pl_ready!`
630        assert_eq!(cnt.get(), 506);
631
632        let pl = Pipeline::new(
633            1,
634            FwdPlErr {
635                pl: Pipeline::new(10, Srv(cnt.clone())),
636            },
637        );
638        assert_eq!(pl.call(0).await, Err(4));
639        pl.shutdown().await;
640        assert_eq!(cnt.get(), 507);
641    }
642
643    #[ntex::test]
644    async fn pipeline_calls() {
645        let cnt = Rc::new(Cell::new(0));
646        let pl = Srv(cnt.clone()).pipeline(1);
647
648        assert_eq!(pl.call_static(1).await, Ok(2));
649        assert_eq!(cnt.get(), 1);
650
651        // all entry points reuse the same successful readiness check
652        assert_eq!(pl.ready().await, Ok(()));
653        assert_eq!(cnt.get(), 1);
654        assert_eq!(pl.call(2).await, Ok(3));
655        assert_eq!(cnt.get(), 1);
656        assert_eq!(ServiceCaller::call_service(&pl, 3).await, Ok(4));
657        assert_eq!(cnt.get(), 1);
658
659        assert_eq!(lazy(|cx| pl.poll_ready(cx)).await, Poll::Ready(Ok(())));
660        assert_eq!(cnt.get(), 1);
661        assert_eq!(pl.call_static(3).await, Ok(4));
662        assert_eq!(pl.call_static(3).await, Ok(4));
663        assert_eq!(cnt.get(), 1);
664
665        let b = pl.bind();
666        assert!(format!("{b:?}").contains("PipelineBinding"));
667        assert_eq!(b.call_static(4).await, Ok(5));
668        assert_eq!(cnt.get(), 1);
669        assert_eq!(b.ready().await, Ok(()));
670        assert_eq!(cnt.get(), 1);
671        assert_eq!(pl.call(4).await, Ok(5));
672        assert_eq!(b.call(4).await, Ok(5));
673        assert_eq!(cnt.get(), 1);
674
675        // owned calls also share readiness regardless of polling order
676        assert_eq!(pl.ready().await, Ok(()));
677        let fut1 = pl.call_static(1);
678        let fut2 = pl.call_static(2);
679        assert_eq!(fut2.await, Ok(3));
680        assert_eq!(cnt.get(), 1);
681        assert_eq!(fut1.await, Ok(2));
682        assert_eq!(cnt.get(), 1);
683
684        // shutdown invalidates cached readiness
685        assert_eq!(pl.ready().await, Ok(()));
686        assert_eq!(cnt.get(), 1);
687        pl.shutdown().await;
688        assert_eq!(pl.call(1).await, Ok(2));
689        assert_eq!(cnt.get(), 102);
690
691        let svc = apply_fn(
692            Srv(cnt.clone()),
693            async |req: usize, svc: &dev::ApplyCtx<'_, Srv, usize, usize>| {
694                svc.call_service(req * 2).await
695            },
696        );
697        let pl = Pipeline::new(1, svc);
698        assert_eq!(pl.call(2).await, Ok(5));
699        assert_eq!(cnt.get(), 103);
700    }
701
702    #[ntex::test]
703    async fn fn_conversions() {
704        let pl = Pipeline::new(5, async |st: &usize, req: usize| Ok::<_, ()>(st + req));
705        assert_eq!(pl.call(1).await, Ok(6));
706
707        let _: FnServiceSt<_, usize, usize, usize, ()> =
708            IntoService::into_service(async |st: &usize, req: usize| Ok::<_, ()>(st + req));
709
710        let f = factory(async |st: &usize| Ok::<_, ()>(Srv(Rc::new(Cell::new(*st)))));
711        let pl = f.pipeline(1).await.unwrap();
712        assert_eq!(pl.call(1).await, Ok(2));
713    }
714
715    #[ntex::test]
716    async fn factory_combinators() {
717        let cnt = Rc::new(Cell::new(0));
718        let c = cnt.clone();
719        let f = fn_factory(async move |st: &usize| {
720            if *st == 0 {
721                Err(())
722            } else {
723                Ok::<_, ()>(Srv(c.clone()))
724            }
725        });
726
727        let rc = Rc::new(f.clone());
728        let pl = rc.pipeline(1).await.unwrap();
729        assert_eq!(pl.call(1).await, Ok(2));
730        assert!(rc.pipeline(0).await.is_err());
731
732        let f2 = f.clone().map_init_err(|()| "init");
733        assert_eq!(f2.create(&0).await.err(), Some("init"));
734
735        let f2 = f.clone().and_then(f.clone());
736        let pl = f2.pipeline(1).await.unwrap();
737        assert_eq!(pl.call(1).await, Ok(3));
738
739        let f2 = factory(f.clone()).map(|r| r * 10).map_err(|_| ());
740        let pl = f2.pipeline(1).await.unwrap();
741        assert_eq!(pl.call(1).await, Ok(20));
742        assert_eq!(pl.call(0).await, Err(()));
743
744        let pf = PipelineFactory::new(f.clone());
745        let pf2 = pf.clone();
746        assert!(format!("{pf2:?}").contains("PipelineFactory"));
747        let pl = pf2.create(2).await.unwrap();
748        assert_eq!(pl.call(1).await, Ok(3));
749        assert!(pf.create(0).await.is_err());
750    }
751
752    #[ntex::test]
753    async fn boxed_and_debug() {
754        let cnt = Rc::new(Cell::new(0));
755
756        let svc = boxed::service(Srv(cnt.clone()));
757        let pl = Pipeline::new(1, svc.clone());
758        assert_eq!(pl.call(1).await, Ok(2));
759
760        let f = boxed::factory(fn_factory(async |_: &usize| Ok::<_, ()>(Srv::default())));
761        let pl = f.clone().pipeline(1).await.unwrap();
762        assert_eq!(pl.call(1).await, Ok(2));
763
764        let s = format!("{:?}", service(Srv::default()).map(|r| r).map_err(|e| e));
765        assert!(s.contains("Map") && s.contains("MapErr"));
766    }
767
768    #[test]
769    fn request_state() {
770        let st = State {
771            req: 1,
772            state: "st",
773        };
774        assert_eq!(st.unpack(), ("st", 1));
775        assert_eq!(("st", 2).unpack(), ("st", 2));
776    }
777}