1#![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)]
62pub trait Service<St, Req> {
122 type Res;
124
125 type Error;
127
128 async fn call(&self, req: Req, ctx: Ctx<'_, Self, St>) -> Result<Self::Res, Self::Error>;
136
137 #[inline]
138 fn bind(&self, ctx: BindCtx<'_>) {}
156
157 #[inline]
158 async fn ready(&self, ctx: Ctx<'_, Self, St>) -> Result<(), Self::Error> {
184 Ok(())
185 }
186
187 #[inline]
188 async fn shutdown(&self, ctx: Ctx<'_, Self, St>) {}
192
193 #[inline]
194 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 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 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 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
251pub trait ServiceFactory<St, Req> {
262 type Res;
264
265 type Error;
267
268 type Service: Service<St, Req, Res = Self::Res, Error = Self::Error>;
270
271 type InitError;
273
274 async fn create(&self, st: &St) -> Result<Self::Service, Self::InitError>;
276
277 #[inline]
278 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 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 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 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 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 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
443pub trait ServiceCaller<Req, Res, Err> {
445 async fn call_service(&self, req: Req) -> Result<Res, Err>;
447}
448
449pub trait IntoService<S, St, Req>
451where
452 S: Service<St, Req>,
453{
454 fn into_service(self) -> S;
456}
457
458pub trait IntoServiceFactory<Sf, St, Req>
460where
461 Sf: ServiceFactory<St, Req>,
462{
463 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
487pub 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 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 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 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 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}