mirrors/ntex
1
0
Fork 0
mirror of https://github.com/ntex-rs/ntex.git synced 2025-04-03 21:07:39 +03:00
Code Issues Projects Releases Packages Wiki Activity

Use GAT (#153)

Browse source 
* Rename Transform to Middleware

* Drop FnService's shutdown helper

* refactor Service trait to use GAT

* Migrate ntex to new service

* move Stack to service

* use BoxFuture

* simplify poll_shitdown method
This commit is contained in:
Nikolay Kim 2022-12-27 00:58:38 +06:00 • committed by GitHub
parent de9738c9c0
commit 537d8dc18d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
81 changed files with 2069 additions and 2495 deletions
Download patch file Download diff file Expand all files Collapse all files

10
ntex-service/CHANGES.md
View file

@ -1,5 +1,15 @@
# Changes
## [0.4.0-alpha.0] - 2022-12-xx
* Rename Transform to Middleware
* Drop FnService's shutdown helper
* Simplify Service::poll_shutdown() method
* Add forward_poll_ready and forward_poll_shutdown macros
## [0.3.3] - 2022-07-08
* Revert cleanups

6
ntex-service/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "ntex-service"
version = "0.3.3"
version = "0.4.0-alpha.0"
authors = ["ntex contributors <team@ntex.rs>"]
description = "ntex service"
keywords = ["network", "framework", "async", "futures"]
@ -19,5 +19,5 @@ path = "src/lib.rs"
pin-project-lite = "0.2.6"
[dev-dependencies]
ntex = { version = "0.5", features = ["tokio"] }
ntex-util = "0.1.5"
ntex = { version = "0.6.0-alpha.0", features = ["tokio"] }
ntex-util = "0.2.0-alpha.0"

187
ntex-service/src/and_then.rs
View file

@ -1,6 +1,4 @@
use std::{
future::Future, marker::PhantomData, pin::Pin, rc::Rc, task::Context, task::Poll,
};
use std::{future::Future, pin::Pin, task::Context, task::Poll};
use super::{Service, ServiceFactory};
@ -8,49 +6,51 @@ use super::{Service, ServiceFactory};
/// of another service which completes successfully.
///
/// This is created by the `ServiceExt::and_then` method.
pub struct AndThen<A, B, Req>(Rc<(A, B)>, PhantomData<fn(Req)>);
pub struct AndThen<A, B> {
svc1: A,
svc2: B,
}
impl<A, B, Req> AndThen<A, B, Req> {
impl<A, B> AndThen<A, B> {
/// Create new `AndThen` combinator
pub(crate) fn new(a: A, b: B) -> Self
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
Self(Rc::new((a, b)), PhantomData)
pub(crate) fn new(svc1: A, svc2: B) -> Self {
Self { svc1, svc2 }
}
}
impl<A, B, Req> Clone for AndThen<A, B, Req> {
impl<A, B> Clone for AndThen<A, B>
where
A: Clone,
B: Clone,
{
fn clone(&self) -> Self {
AndThen(self.0.clone(), PhantomData)
AndThen {
svc1: self.svc1.clone(),
svc2: self.svc2.clone(),
}
}
}
impl<A, B, Req> Service<Req> for AndThen<A, B, Req>
impl<A, B, Req> Service<Req> for AndThen<A, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
type Response = B::Response;
type Error = A::Error;
type Future = AndThenServiceResponse<A, B, Req>;
type Future<'f> = AndThenServiceResponse<'f, A, B, Req> where Self: 'f, Req: 'f;
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let srv = self.0.as_ref();
let not_ready = !srv.0.poll_ready(cx)?.is_ready();
if !srv.1.poll_ready(cx)?.is_ready() || not_ready {
let not_ready = !self.svc1.poll_ready(cx)?.is_ready();
if !self.svc2.poll_ready(cx)?.is_ready() || not_ready {
Poll::Pending
} else {
Poll::Ready(Ok(()))
}
}
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
let srv = self.0.as_ref();
if srv.0.poll_shutdown(cx, is_error).is_ready()
&& srv.1.poll_shutdown(cx, is_error).is_ready()
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
if self.svc1.poll_shutdown(cx).is_ready() && self.svc2.poll_shutdown(cx).is_ready()
{
Poll::Ready(())
} else {
@ -59,41 +59,45 @@ where
}
#[inline]
fn call(&self, req: Req) -> Self::Future {
fn call(&self, req: Req) -> Self::Future<'_> {
AndThenServiceResponse {
slf: self,
state: State::A {
fut: self.0.as_ref().0.call(req),
b: Some(self.0.clone()),
fut: self.svc1.call(req),
},
}
}
}
pin_project_lite::pin_project! {
pub struct AndThenServiceResponse<A, B, Req>
pub struct AndThenServiceResponse<'f, A, B, Req>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
slf: &'f AndThen<A, B>,
#[pin]
state: State<A, B, Req>,
state: State<'f, A, B, Req>,
}
}
pin_project_lite::pin_project! {
#[project = StateProject]
enum State<A, B, Req>
enum State<'f, A, B, Req>
where
A: Service<Req>,
A: 'f,
Req: 'f,
B: Service<A::Response, Error = A::Error>,
B: 'f,
{
A { #[pin] fut: A::Future, b: Option<Rc<(A, B)>> },
B { #[pin] fut: B::Future },
A { #[pin] fut: A::Future<'f> },
B { #[pin] fut: B::Future<'f> },
Empty,
}
}
impl<A, B, Req> Future for AndThenServiceResponse<A, B, Req>
impl<'f, A, B, Req> Future for AndThenServiceResponse<'f, A, B, Req>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
@ -104,11 +108,9 @@ where
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProject::A { fut, b } => match fut.poll(cx)? {
StateProject::A { fut } => match fut.poll(cx)? {
Poll::Ready(res) => {
let b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A
let fut = b.as_ref().1.call(res);
let fut = this.slf.svc2.call(res);
this.state.set(State::B { fut });
self.poll(cx)
}
@ -126,27 +128,19 @@ where
}
/// `.and_then()` service factory combinator
pub struct AndThenFactory<A, B, Req, Cfg> {
inner: Rc<(A, B)>,
_t: PhantomData<fn(Req, Cfg)>,
pub struct AndThenFactory<A, B> {
svc1: A,
svc2: B,
}
impl<A, B, Req, Cfg> AndThenFactory<A, B, Req, Cfg>
where
A: ServiceFactory<Req, Cfg>,
B: ServiceFactory<A::Response, Cfg, Error = A::Error, InitError = A::InitError>,
Cfg: Clone,
{
impl<A, B> AndThenFactory<A, B> {
/// Create new `AndThenFactory` combinator
pub fn new(a: A, b: B) -> Self {
Self {
inner: Rc::new((a, b)),
_t: PhantomData,
}
pub fn new(svc1: A, svc2: B) -> Self {
Self { svc1, svc2 }
}
}
impl<A, B, Req, Cfg> ServiceFactory<Req, Cfg> for AndThenFactory<A, B, Req, Cfg>
impl<A, B, Req, Cfg> ServiceFactory<Req, Cfg> for AndThenFactory<A, B>
where
A: ServiceFactory<Req, Cfg>,
B: ServiceFactory<A::Response, Cfg, Error = A::Error, InitError = A::InitError>,
@ -155,83 +149,77 @@ where
type Response = B::Response;
type Error = A::Error;
type Service = AndThen<A::Service, B::Service, Req>;
type Service = AndThen<A::Service, B::Service>;
type InitError = A::InitError;
type Future = AndThenServiceFactoryResponse<A, B, Req, Cfg>;
type Future<'f> = AndThenFactoryResponse<'f, A, B, Req, Cfg> where Self: 'f, Cfg: 'f;
fn new_service(&self, cfg: Cfg) -> Self::Future {
let inner = &*self.inner;
AndThenServiceFactoryResponse::new(
inner.0.new_service(cfg.clone()),
inner.1.new_service(cfg),
)
#[inline]
fn create(&self, cfg: Cfg) -> Self::Future<'_> {
AndThenFactoryResponse {
fut1: self.svc1.create(cfg.clone()),
fut2: self.svc2.create(cfg),
svc1: None,
svc2: None,
}
}
}
impl<A, B, Req, Cfg> Clone for AndThenFactory<A, B, Req, Cfg> {
impl<A, B> Clone for AndThenFactory<A, B>
where
A: Clone,
B: Clone,
{
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
_t: PhantomData,
svc1: self.svc1.clone(),
svc2: self.svc2.clone(),
}
}
}
pin_project_lite::pin_project! {
pub struct AndThenServiceFactoryResponse<A, B, Req, Cfg>
pub struct AndThenFactoryResponse<'f, A, B, Req, Cfg>
where
A: ServiceFactory<Req, Cfg>,
A: 'f,
B: ServiceFactory<A::Response, Cfg>,
B: 'f,
Cfg: 'f
{
#[pin]
fut_a: A::Future,
fut1: A::Future<'f>,
#[pin]
fut_b: B::Future,
fut2: B::Future<'f>,
a: Option<A::Service>,
b: Option<B::Service>,
svc1: Option<A::Service>,
svc2: Option<B::Service>,
}
}
impl<A, B, Req, Cfg> AndThenServiceFactoryResponse<A, B, Req, Cfg>
where
A: ServiceFactory<Req, Cfg>,
B: ServiceFactory<A::Response, Cfg>,
{
fn new(fut_a: A::Future, fut_b: B::Future) -> Self {
AndThenServiceFactoryResponse {
fut_a,
fut_b,
a: None,
b: None,
}
}
}
impl<A, B, Req, Cfg> Future for AndThenServiceFactoryResponse<A, B, Req, Cfg>
impl<'f, A, B, Req, Cfg> Future for AndThenFactoryResponse<'f, A, B, Req, Cfg>
where
A: ServiceFactory<Req, Cfg>,
B: ServiceFactory<A::Response, Cfg, Error = A::Error, InitError = A::InitError>,
{
type Output = Result<AndThen<A::Service, B::Service, Req>, A::InitError>;
type Output = Result<AndThen<A::Service, B::Service>, A::InitError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
if this.a.is_none() {
if let Poll::Ready(service) = this.fut_a.poll(cx)? {
*this.a = Some(service);
if this.svc1.is_none() {
if let Poll::Ready(service) = this.fut1.poll(cx)? {
*this.svc1 = Some(service);
}
}
if this.b.is_none() {
if let Poll::Ready(service) = this.fut_b.poll(cx)? {
*this.b = Some(service);
if this.svc2.is_none() {
if let Poll::Ready(service) = this.fut2.poll(cx)? {
*this.svc2 = Some(service);
}
}
if this.a.is_some() && this.b.is_some() {
if this.svc1.is_some() && this.svc2.is_some() {
Poll::Ready(Ok(AndThen::new(
this.a.take().unwrap(),
this.b.take().unwrap(),
this.svc1.take().unwrap(),
this.svc2.take().unwrap(),
)))
} else {
Poll::Pending
@ -246,19 +234,20 @@ mod tests {
use crate::{fn_factory, pipeline, pipeline_factory, Service, ServiceFactory};
use ntex_util::future::{lazy, Ready};
#[derive(Clone)]
struct Srv1(Rc<Cell<usize>>);
impl Service<&'static str> for Srv1 {
type Response = &'static str;
type Error = ();
type Future = Ready<Self::Response, ()>;
type Future<'f> = Ready<Self::Response, ()>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.set(self.0.get() + 1);
Poll::Ready(Ok(()))
}
fn call(&self, req: &'static str) -> Self::Future {
fn call(&self, req: &'static str) -> Self::Future<'_> {
Ready::Ok(req)
}
}
@ -269,14 +258,14 @@ mod tests {
impl Service<&'static str> for Srv2 {
type Response = (&'static str, &'static str);
type Error = ();
type Future = Ready<Self::Response, ()>;
type Future<'f> = Ready<Self::Response, ()>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.set(self.0.get() + 1);
Poll::Ready(Ok(()))
}
fn call(&self, req: &'static str) -> Self::Future {
fn call(&self, req: &'static str) -> Self::Future<'_> {
Ready::Ok((req, "srv2"))
}
}
@ -290,7 +279,7 @@ mod tests {
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert_eq!(res, Poll::Ready(Ok(())));
assert_eq!(cnt.get(), 2);
let res = lazy(|cx| srv.poll_shutdown(cx, true)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
}
@ -313,7 +302,7 @@ mod tests {
.and_then(move || Ready::from(Ok(Srv2(cnt.clone()))))
.clone();
let srv = new_srv.new_service(()).await.unwrap();
let srv = new_srv.create(&()).await.unwrap();
let res = srv.call("srv1").await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), ("srv1", "srv2"));

134
ntex-service/src/apply.rs
View file

@ -1,4 +1,6 @@
use std::{future::Future, marker::PhantomData, pin::Pin, task::Context, task::Poll};
use std::{
future::Future, marker::PhantomData, pin::Pin, rc::Rc, task::Context, task::Poll,
};
use super::{IntoService, IntoServiceFactory, Service, ServiceFactory};
@ -9,7 +11,7 @@ pub fn apply_fn<T, Req, F, R, In, Out, Err, U>(
) -> Apply<T, Req, F, R, In, Out, Err>
where
T: Service<Req, Error = Err>,
F: Fn(In, &T) -> R,
F: Fn(In, Rc<T>) -> R,
R: Future<Output = Result<Out, Err>>,
U: IntoService<T, Req>,
{
@ -20,14 +22,14 @@ where
pub fn apply_fn_factory<T, Req, Cfg, F, R, In, Out, Err, U>(
service: U,
f: F,
) -> ApplyServiceFactory<T, Req, Cfg, F, R, In, Out, Err>
) -> ApplyFactory<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R + Clone,
F: Fn(In, Rc<T::Service>) -> R + Clone,
R: Future<Output = Result<Out, Err>>,
U: IntoServiceFactory<T, Req, Cfg>,
{
ApplyServiceFactory::new(service.into_factory(), f)
ApplyFactory::new(service.into_factory(), f)
}
/// `Apply` service combinator
@ -35,7 +37,7 @@ pub struct Apply<T, Req, F, R, In, Out, Err>
where
T: Service<Req, Error = Err>,
{
service: T,
service: Rc<T>,
f: F,
r: PhantomData<fn(Req) -> (In, Out, R)>,
}
@ -43,14 +45,14 @@ where
impl<T, Req, F, R, In, Out, Err> Apply<T, Req, F, R, In, Out, Err>
where
T: Service<Req, Error = Err>,
F: Fn(In, &T) -> R,
F: Fn(In, Rc<T>) -> R,
R: Future<Output = Result<Out, Err>>,
{
/// Create new `Apply` combinator
fn new(service: T, f: F) -> Self {
Self {
service,
f,
service: Rc::new(service),
r: PhantomData,
}
}
@ -59,7 +61,7 @@ where
impl<T, Req, F, R, In, Out, Err> Clone for Apply<T, Req, F, R, In, Out, Err>
where
T: Service<Req, Error = Err> + Clone,
F: Fn(In, &T) -> R + Clone,
F: Fn(In, Rc<T>) -> R + Clone,
R: Future<Output = Result<Out, Err>>,
{
fn clone(&self) -> Self {
@ -74,34 +76,27 @@ where
impl<T, Req, F, R, In, Out, Err> Service<In> for Apply<T, Req, F, R, In, Out, Err>
where
T: Service<Req, Error = Err>,
F: Fn(In, &T) -> R,
F: Fn(In, Rc<T>) -> R,
R: Future<Output = Result<Out, Err>>,
{
type Response = Out;
type Error = Err;
type Future = R;
type Future<'f> = R where Self: 'f, In: 'f, R: 'f;
crate::forward_poll_ready!(service);
crate::forward_poll_shutdown!(service);
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.service.poll_ready(cx)
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
self.service.poll_shutdown(cx, is_error)
}
#[inline]
fn call(&self, req: In) -> Self::Future {
(self.f)(req, &self.service)
fn call(&self, req: In) -> Self::Future<'_> {
(self.f)(req, self.service.clone())
}
}
/// `apply()` service factory
pub struct ApplyServiceFactory<T, Req, Cfg, F, R, In, Out, Err>
pub struct ApplyFactory<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R + Clone,
F: Fn(In, Rc<T::Service>) -> R + Clone,
R: Future<Output = Result<Out, Err>>,
{
service: T,
@ -109,10 +104,10 @@ where
r: PhantomData<fn(Req, Cfg) -> (R, In, Out)>,
}
impl<T, Req, Cfg, F, R, In, Out, Err> ApplyServiceFactory<T, Req, Cfg, F, R, In, Out, Err>
impl<T, Req, Cfg, F, R, In, Out, Err> ApplyFactory<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R + Clone,
F: Fn(In, Rc<T::Service>) -> R + Clone,
R: Future<Output = Result<Out, Err>>,
{
/// Create new `ApplyNewService` new service instance
@ -126,10 +121,10 @@ where
}
impl<T, Req, Cfg, F, R, In, Out, Err> Clone
for ApplyServiceFactory<T, Req, Cfg, F, R, In, Out, Err>
for ApplyFactory<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err> + Clone,
F: Fn(In, &T::Service) -> R + Clone,
F: Fn(In, Rc<T::Service>) -> R + Clone,
R: Future<Output = Result<Out, Err>>,
{
fn clone(&self) -> Self {
@ -142,10 +137,10 @@ where
}
impl<T, Req, Cfg, F, R, In, Out, Err> ServiceFactory<In, Cfg>
for ApplyServiceFactory<T, Req, Cfg, F, R, In, Out, Err>
for ApplyFactory<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R + Clone,
F: Fn(In, Rc<T::Service>) -> R + Clone,
R: Future<Output = Result<Out, Err>>,
{
type Response = Out;
@ -153,48 +148,39 @@ where
type Service = Apply<T::Service, Req, F, R, In, Out, Err>;
type InitError = T::InitError;
type Future = ApplyServiceFactoryResponse<T, Req, Cfg, F, R, In, Out, Err>;
type Future<'f> = ApplyFactoryResponse<'f, T, Req, Cfg, F, R, In, Out, Err> where Self: 'f, Cfg: 'f;
fn new_service(&self, cfg: Cfg) -> Self::Future {
ApplyServiceFactoryResponse::new(self.service.new_service(cfg), self.f.clone())
}
}
pin_project_lite::pin_project! {
pub struct ApplyServiceFactoryResponse<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R,
R: Future<Output = Result<Out, Err>>,
{
#[pin]
fut: T::Future,
f: Option<F>,
r: PhantomData<(In, Out)>,
}
}
impl<T, Req, Cfg, F, R, In, Out, Err>
ApplyServiceFactoryResponse<T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R,
R: Future<Output = Result<Out, Err>>,
{
fn new(fut: T::Future, f: F) -> Self {
Self {
f: Some(f),
fut,
r: PhantomData,
#[inline]
fn create(&self, cfg: Cfg) -> Self::Future<'_> {
ApplyFactoryResponse {
fut: self.service.create(cfg),
f: Some(self.f.clone()),
_t: PhantomData,
}
}
}
impl<T, Req, Cfg, F, R, In, Out, Err> Future
for ApplyServiceFactoryResponse<T, Req, Cfg, F, R, In, Out, Err>
pin_project_lite::pin_project! {
pub struct ApplyFactoryResponse<'f, T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
T: 'f,
F: Fn(In, Rc<T::Service>) -> R,
R: Future<Output = Result<Out, Err>>,
Cfg: 'f,
{
#[pin]
fut: T::Future<'f>,
f: Option<F>,
_t: PhantomData<(In, Out)>,
}
}
impl<'f, T, Req, Cfg, F, R, In, Out, Err> Future
for ApplyFactoryResponse<'f, T, Req, Cfg, F, R, In, Out, Err>
where
T: ServiceFactory<Req, Cfg, Error = Err>,
F: Fn(In, &T::Service) -> R,
F: Fn(In, Rc<T::Service>) -> R,
R: Future<Output = Result<Out, Err>>,
{
type Output = Result<Apply<T::Service, Req, F, R, In, Out, Err>, T::InitError>;
@ -213,7 +199,7 @@ where
#[cfg(test)]
mod tests {
use ntex_util::future::{lazy, Ready};
use std::task::{Context, Poll};
use std::task::Poll;
use super::*;
use crate::{pipeline, pipeline_factory, Service, ServiceFactory};
@ -224,13 +210,9 @@ mod tests {
impl Service<()> for Srv {
type Response = ();
type Error = ();
type Future = Ready<(), ()>;
type Future<'f> = Ready<(), ()>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&self, _: ()) -> Self::Future {
fn call(&self, _: ()) -> Self::Future<'_> {
Ready::Ok(())
}
}
@ -249,7 +231,7 @@ mod tests {
);
assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
let res = lazy(|cx| srv.poll_shutdown(cx, true)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
let res = srv.call("srv").await;
@ -258,7 +240,7 @@ mod tests {
}
#[ntex::test]
async fn test_new_service() {
async fn test_create() {
let new_srv = pipeline_factory(
apply_fn_factory(
|| Ready::<_, ()>::Ok(Srv),
@ -273,7 +255,7 @@ mod tests {
.clone(),
);
let srv = new_srv.new_service(()).await.unwrap();
let srv = new_srv.create(&()).await.unwrap();
assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));

311
ntex-service/src/boxed.rs
View file

@ -1,141 +1,174 @@
use std::{
future::Future, marker::PhantomData, pin::Pin, rc::Rc, task::Context, task::Poll,
};
use std::{future::Future, pin::Pin, rc::Rc, task::Context, task::Poll};
use crate::{Service, ServiceFactory};
pub type BoxFuture<I, E> = Pin<Box<dyn Future<Output = Result<I, E>>>>;
pub type BoxService<Req, Res, Err> =
Box<dyn Service<Req, Response = Res, Error = Err, Future = BoxFuture<Res, Err>>>;
pub type RcService<Req, Res, Err> =
Rc<dyn Service<Req, Response = Res, Error = Err, Future = BoxFuture<Res, Err>>>;
pub struct BoxServiceFactory<C, Req, Res, Err, InitErr>(Inner<C, Req, Res, Err, InitErr>);
pub type BoxFuture<'a, I, E> = Pin<Box<dyn Future<Output = Result<I, E>> + 'a>>;
/// Create boxed service factory
pub fn factory<T, R, C>(
factory: T,
) -> BoxServiceFactory<C, R, T::Response, T::Error, T::InitError>
pub fn factory<F, R, C>(
factory: F,
) -> BoxServiceFactory<C, R, F::Response, F::Error, F::InitError>
where
C: 'static,
R: 'static,
T: ServiceFactory<R, C> + 'static,
T::Response: 'static,
T::Error: 'static,
T::InitError: 'static,
C: 'static,
F: crate::ServiceFactory<R, C> + 'static,
F::Service: 'static,
{
BoxServiceFactory(Box::new(FactoryWrapper {
factory,
_t: std::marker::PhantomData,
}))
BoxServiceFactory(Box::new(factory))
}
/// Create rc boxed service factory
pub fn rcfactory<F, R, C>(
factory: F,
) -> RcServiceFactory<C, R, F::Response, F::Error, F::InitError>
where
R: 'static,
C: 'static,
F: crate::ServiceFactory<R, C> + 'static,
F::Service: 'static,
{
RcServiceFactory(Rc::new(factory))
}
/// Create boxed service
pub fn service<T, R>(service: T) -> BoxService<R, T::Response, T::Error>
pub fn service<S, R>(service: S) -> BoxService<R, S::Response, S::Error>
where
R: 'static,
T: Service<R> + 'static,
T::Future: 'static,
S: crate::Service<R> + 'static,
{
Box::new(ServiceWrapper(service, PhantomData))
BoxService(Box::new(service))
}
/// Create rc service
pub fn rcservice<T, R>(service: T) -> RcService<R, T::Response, T::Error>
pub fn rcservice<S, R>(service: S) -> RcService<R, S::Response, S::Error>
where
R: 'static,
T: Service<R> + 'static,
T::Future: 'static,
S: crate::Service<R> + 'static,
{
Rc::new(ServiceWrapper(service, PhantomData))
RcService(Rc::new(service))
}
type Inner<C, Req, Res, Err, InitErr> = Box<
dyn ServiceFactory<
Req,
C,
Response = Res,
Error = Err,
InitError = InitErr,
Service = BoxService<Req, Res, Err>,
Future = BoxFuture<BoxService<Req, Res, Err>, InitErr>,
>,
>;
pub trait ServiceObj<Req> {
type Response;
type Error;
impl<C, Req, Res, Err, InitErr> ServiceFactory<Req, C>
for BoxServiceFactory<C, Req, Res, Err, InitErr>
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()>;
fn call<'a>(&'a self, req: Req) -> BoxFuture<'a, Self::Response, Self::Error>
where
Req: 'a;
}
impl<S, Req> ServiceObj<Req> for S
where
Req: 'static,
Res: 'static,
Err: 'static,
InitErr: 'static,
S: crate::Service<Req>,
{
type Response = Res;
type Error = Err;
type InitError = InitErr;
type Service = BoxService<Req, Res, Err>;
type Response = S::Response;
type Error = S::Error;
type Future = BoxFuture<Self::Service, InitErr>;
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
crate::Service::poll_ready(self, cx)
}
fn new_service(&self, cfg: C) -> Self::Future {
self.0.new_service(cfg)
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
crate::Service::poll_shutdown(self, cx)
}
#[inline]
fn call<'a>(&'a self, req: Req) -> BoxFuture<'a, Self::Response, Self::Error>
where
Req: 'a,
{
Box::pin(crate::Service::call(self, req))
}
}
struct FactoryWrapper<C, R, T: ServiceFactory<R, C>> {
factory: T,
_t: std::marker::PhantomData<(C, R)>,
trait ServiceFactoryObj<Req, Cfg> {
type Response;
type Error;
type InitError;
fn create<'a>(
&'a self,
cfg: Cfg,
) -> BoxFuture<'a, BoxService<Req, Self::Response, Self::Error>, Self::InitError>
where
Cfg: 'a;
}
impl<C, R, T, Res, Err, InitErr> ServiceFactory<R, C> for FactoryWrapper<C, R, T>
where
R: 'static,
Res: 'static,
Err: 'static,
InitErr: 'static,
T: ServiceFactory<R, C, Response = Res, Error = Err, InitError = InitErr> + 'static,
T::Service: 'static,
T::Future: 'static,
<T::Service as Service<R>>::Future: 'static,
{
type Response = Res;
type Error = Err;
type InitError = InitErr;
type Service = BoxService<R, Res, Err>;
type Future = BoxFuture<Self::Service, Self::InitError>;
trait ServiceFactoryRcObj<Req, Cfg> {
type Response;
type Error;
type InitError;
fn new_service(&self, cfg: C) -> Self::Future {
let fut = self.factory.new_service(cfg);
Box::pin(async move {
let srv = fut.await?;
Ok(ServiceWrapper::boxed(srv))
})
fn create<'a>(
&'a self,
cfg: Cfg,
) -> BoxFuture<'a, RcService<Req, Self::Response, Self::Error>, Self::InitError>
where
Cfg: 'a;
}
impl<F, Req, Cfg> ServiceFactoryObj<Req, Cfg> for F
where
Cfg: 'static,
Req: 'static,
F: crate::ServiceFactory<Req, Cfg>,
F::Service: 'static,
{
type Response = F::Response;
type Error = F::Error;
type InitError = F::InitError;
#[inline]
fn create<'a>(
&'a self,
cfg: Cfg,
) -> BoxFuture<'a, BoxService<Req, Self::Response, Self::Error>, Self::InitError>
where
Cfg: 'a,
{
let fut = crate::ServiceFactory::create(self, cfg);
Box::pin(async move { fut.await.map(service) })
}
}
struct ServiceWrapper<T: Service<R>, R>(T, PhantomData<R>);
impl<T, R> ServiceWrapper<T, R>
impl<F, Req, Cfg> ServiceFactoryRcObj<Req, Cfg> for F
where
R: 'static,
T: Service<R> + 'static,
T::Future: 'static,
Cfg: 'static,
Req: 'static,
F: crate::ServiceFactory<Req, Cfg>,
F::Service: 'static,
{
fn boxed(service: T) -> BoxService<R, T::Response, T::Error> {
Box::new(ServiceWrapper(service, PhantomData))
type Response = F::Response;
type Error = F::Error;
type InitError = F::InitError;
#[inline]
fn create<'a>(
&'a self,
cfg: Cfg,
) -> BoxFuture<'a, RcService<Req, Self::Response, Self::Error>, Self::InitError>
where
Cfg: 'a,
{
let fut = crate::ServiceFactory::create(self, cfg);
Box::pin(async move { fut.await.map(rcservice) })
}
}
impl<T, R, Res, Err> Service<R> for ServiceWrapper<T, R>
pub struct BoxService<Req, Res, Err>(Box<dyn ServiceObj<Req, Response = Res, Error = Err>>);
impl<Req, Res, Err> crate::Service<Req> for BoxService<Req, Res, Err>
where
T: Service<R, Response = Res, Error = Err>,
T::Future: 'static,
Req: 'static,
{
type Response = Res;
type Error = Err;
type Future = BoxFuture<Res, Err>;
type Future<'f> = BoxFuture<'f, Res, Err> where Self: 'f, Req: 'f;
#[inline]
fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
@ -143,12 +176,94 @@ where
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
self.0.poll_shutdown(cx, is_error)
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
self.0.poll_shutdown(cx)
}
#[inline]
fn call(&self, req: R) -> Self::Future {
Box::pin(self.0.call(req))
fn call(&self, req: Req) -> Self::Future<'_> {
self.0.call(req)
}
}
pub struct RcService<Req, Res, Err>(Rc<dyn ServiceObj<Req, Response = Res, Error = Err>>);
impl<Req, Res, Err> Clone for RcService<Req, Res, Err> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl<Req, Res, Err> crate::Service<Req> for RcService<Req, Res, Err>
where
Req: 'static,
{
type Response = Res;
type Error = Err;
type Future<'f> = BoxFuture<'f, Res, Err> where Self: 'f, Req: 'f;
#[inline]
fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.poll_ready(ctx)
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
self.0.poll_shutdown(cx)
}
#[inline]
fn call(&self, req: Req) -> Self::Future<'_> {
self.0.call(req)
}
}
pub struct BoxServiceFactory<Cfg, Req, Res, Err, InitErr>(
Box<dyn ServiceFactoryObj<Req, Cfg, Response = Res, Error = Err, InitError = InitErr>>,
);
impl<C, Req, Res, Err, InitErr> crate::ServiceFactory<Req, C>
for BoxServiceFactory<C, Req, Res, Err, InitErr>
where
Req: 'static,
{
type Response = Res;
type Error = Err;
type Service = BoxService<Req, Res, Err>;
type InitError = InitErr;
type Future<'f> = BoxFuture<'f, Self::Service, InitErr> where Self: 'f, C: 'f;
#[inline]
fn create(&self, cfg: C) -> Self::Future<'_> {
self.0.create(cfg)
}
}
pub struct RcServiceFactory<Cfg, Req, Res, Err, InitErr>(
Rc<dyn ServiceFactoryRcObj<Req, Cfg, Response = Res, Error = Err, InitError = InitErr>>,
);
impl<C, Req, Res, Err, InitErr> Clone for RcServiceFactory<C, Req, Res, Err, InitErr> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl<C, Req, Res, Err, InitErr> crate::ServiceFactory<Req, C>
for RcServiceFactory<C, Req, Res, Err, InitErr>
where
Req: 'static,
{
type Response = Res;
type Error = Err;
type Service = RcService<Req, Res, Err>;
type InitError = InitErr;
type Future<'f> = BoxFuture<'f, Self::Service, InitErr> where Self: 'f, C: 'f;
#[inline]
fn create(&self, cfg: C) -> Self::Future<'_> {
self.0.create(cfg)
}
}

247
ntex-service/src/fn_service.rs
View file

@ -1,5 +1,4 @@
use std::task::{Context, Poll};
use std::{cell::Cell, future::ready, future::Future, future::Ready, marker::PhantomData};
use std::{future::ready, future::Future, future::Ready, marker::PhantomData};
use crate::{IntoService, IntoServiceFactory, Service, ServiceFactory};
@ -41,7 +40,7 @@ where
/// });
///
/// // construct new service
/// let srv = factory.new_service(()).await?;
/// let srv = factory.create(&()).await?;
///
/// // now we can use `div` service
/// let result = srv.call((10, 20)).await?;
@ -51,12 +50,10 @@ where
/// Ok(())
/// }
/// ```
pub fn fn_factory<F, Cfg, Srv, Fut, Req, Err>(
f: F,
) -> FnServiceNoConfig<F, Cfg, Srv, Fut, Req, Err>
pub fn fn_factory<F, Srv, Fut, Req, Err>(f: F) -> FnServiceNoConfig<F, Srv, Fut, Req, Err>
where
Srv: Service<Req>,
F: Fn() -> Fut,
Srv: Service<Req>,
Fut: Future<Output = Result<Srv, Err>>,
{
FnServiceNoConfig::new(f)
@ -78,12 +75,13 @@ where
/// async fn main() -> io::Result<()> {
/// // Create service factory. factory uses config argument for
/// // services it generates.
/// let factory = fn_factory_with_config(|y: usize| {
/// let factory = fn_factory_with_config(|y: &usize| {
/// let y = *y;
/// async move { Ok::<_, io::Error>(fn_service(move |x: usize| async move { Ok::<_, io::Error>(x * y) })) }
/// });
///
/// // construct new service with config argument
/// let srv = factory.new_service(10).await?;
/// let srv = factory.create(&10).await?;
///
/// let result = srv.call(10).await?;
/// assert_eq!(result, 100);
@ -100,114 +98,61 @@ where
Fut: Future<Output = Result<Srv, Err>>,
Srv: Service<Req>,
{
FnServiceConfig::new(f)
FnServiceConfig { f, _t: PhantomData }
}
#[inline]
pub fn fn_shutdown() {}
pub struct FnService<F, Fut, Req, Res, Err, FShut = fn()>
where
F: Fn(Req) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
{
pub struct FnService<F, Req> {
f: F,
f_shutdown: Cell<Option<FShut>>,
_t: PhantomData<Req>,
}
impl<F, Fut, Req, Res, Err> FnService<F, Fut, Req, Res, Err>
impl<F, Req> Clone for FnService<F, Req>
where
F: Fn(Req) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
F: Clone,
{
pub(crate) fn new(f: F) -> Self {
Self {
f,
f_shutdown: Cell::new(Some(fn_shutdown)),
_t: PhantomData,
}
}
/// Set function that get called oin poll_shutdown method of Service trait.
pub fn on_shutdown<FShut>(self, f: FShut) -> FnService<F, Fut, Req, Res, Err, FShut>
where
FShut: FnOnce(),
{
FnService {
f: self.f,
f_shutdown: Cell::new(Some(f)),
_t: PhantomData,
}
}
}
impl<F, Fut, Req, Res, Err, FShut> Clone for FnService<F, Fut, Req, Res, Err, FShut>
where
F: Fn(Req) -> Fut + Clone,
FShut: FnOnce() + Clone,
Fut: Future<Output = Result<Res, Err>>,
{
#[inline]
fn clone(&self) -> Self {
let f = self.f_shutdown.take();
self.f_shutdown.set(f.clone());
Self {
f: self.f.clone(),
f_shutdown: Cell::new(f),
_t: PhantomData,
}
}
}
impl<F, Fut, Req, Res, Err, FShut> Service<Req> for FnService<F, Fut, Req, Res, Err, FShut>
impl<F, Fut, Req, Res, Err> Service<Req> for FnService<F, Req>
where
F: Fn(Req) -> Fut,
FShut: FnOnce(),
Fut: Future<Output = Result<Res, Err>>,
{
type Response = Res;
type Error = Err;
type Future = Fut;
type Future<'f> = Fut where Self: 'f, Req: 'f;
#[inline]
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
#[inline]
fn poll_shutdown(&self, _: &mut Context<'_>, _: bool) -> Poll<()> {
if let Some(f) = self.f_shutdown.take() {
(f)()
}
Poll::Ready(())
}
#[inline]
fn call(&self, req: Req) -> Self::Future {
fn call(&self, req: Req) -> Self::Future<'_> {
(self.f)(req)
}
}
impl<F, Fut, Req, Res, Err> IntoService<FnService<F, Fut, Req, Res, Err>, Req> for F
impl<F, Fut, Req, Res, Err> IntoService<FnService<F, Req>, Req> for F
where
F: Fn(Req) -> Fut,
F: Fn(Req) -> Fut + 'static,
Fut: Future<Output = Result<Res, Err>>,
{
#[inline]
fn into_service(self) -> FnService<F, Fut, Req, Res, Err> {
FnService::new(self)
fn into_service(self) -> FnService<F, Req> {
FnService {
f: self,
_t: PhantomData,
}
}
}
pub struct FnServiceFactory<F, Fut, Req, Res, Err, Cfg, FShut = fn()>
pub struct FnServiceFactory<F, Fut, Req, Res, Err, Cfg>
where
F: Fn(Req) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
{
f: F,
f_shutdown: Cell<Option<FShut>>,
_t: PhantomData<(Req, Cfg)>,
}
@ -217,101 +162,56 @@ where
Fut: Future<Output = Result<Res, Err>>,
{
fn new(f: F) -> Self {
FnServiceFactory {
f,
f_shutdown: Cell::new(Some(fn_shutdown)),
_t: PhantomData,
}
}
/// Set function that get called oin poll_shutdown method of Service trait.
pub fn on_shutdown<FShut>(
self,
f: FShut,
) -> FnServiceFactory<F, Fut, Req, Res, Err, Cfg, FShut>
where
FShut: FnOnce(),
{
FnServiceFactory {
f: self.f,
f_shutdown: Cell::new(Some(f)),
_t: PhantomData,
}
FnServiceFactory { f, _t: PhantomData }
}
}
impl<F, Fut, Req, Res, Err, Cfg, FShut> Clone
for FnServiceFactory<F, Fut, Req, Res, Err, Cfg, FShut>
impl<F, Fut, Req, Res, Err, Cfg> Clone for FnServiceFactory<F, Fut, Req, Res, Err, Cfg>
where
F: Fn(Req) -> Fut + Clone,
FShut: FnOnce() + Clone,
Fut: Future<Output = Result<Res, Err>>,
{
#[inline]
fn clone(&self) -> Self {
let f = self.f_shutdown.take();
self.f_shutdown.set(f.clone());
Self {
f: self.f.clone(),
f_shutdown: Cell::new(f),
_t: PhantomData,
}
}
}
impl<F, Fut, Req, Res, Err, FShut> Service<Req>
for FnServiceFactory<F, Fut, Req, Res, Err, (), FShut>
impl<F, Fut, Req, Res, Err> Service<Req> for FnServiceFactory<F, Fut, Req, Res, Err, ()>
where
F: Fn(Req) -> Fut,
FShut: FnOnce(),
Fut: Future<Output = Result<Res, Err>>,
{
type Response = Res;
type Error = Err;
type Future = Fut;
type Future<'f> = Fut where Self: 'f;
#[inline]
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
#[inline]
fn poll_shutdown(&self, _: &mut Context<'_>, _: bool) -> Poll<()> {
if let Some(f) = self.f_shutdown.take() {
(f)()
}
Poll::Ready(())
}
#[inline]
fn call(&self, req: Req) -> Self::Future {
fn call(&self, req: Req) -> Self::Future<'_> {
(self.f)(req)
}
}
impl<F, Fut, Req, Res, Err, Cfg, FShut> ServiceFactory<Req, Cfg>
for FnServiceFactory<F, Fut, Req, Res, Err, Cfg, FShut>
impl<F, Fut, Req, Res, Err, Cfg> ServiceFactory<Req, Cfg>
for FnServiceFactory<F, Fut, Req, Res, Err, Cfg>
where
F: Fn(Req) -> Fut + Clone,
FShut: FnOnce() + Clone,
F: Fn(Req) -> Fut + Clone + 'static,
Fut: Future<Output = Result<Res, Err>>,
{
type Response = Res;
type Error = Err;
type Service = FnService<F, Fut, Req, Res, Err, FShut>;
type Service = FnService<F, Req>;
type InitError = ();
type Future = Ready<Result<Self::Service, Self::InitError>>;
type Future<'f> = Ready<Result<Self::Service, Self::InitError>> where Self: 'f;
#[inline]
fn new_service(&self, _: Cfg) -> Self::Future {
let f = self.f_shutdown.take();
self.f_shutdown.set(f.clone());
fn create(&self, _: Cfg) -> Self::Future<'_> {
ready(Ok(FnService {
f: self.f.clone(),
f_shutdown: Cell::new(f),
_t: PhantomData,
}))
}
@ -320,7 +220,7 @@ where
impl<F, Fut, Req, Res, Err, Cfg>
IntoServiceFactory<FnServiceFactory<F, Fut, Req, Res, Err, Cfg>, Req, Cfg> for F
where
F: Fn(Req) -> Fut + Clone,
F: Fn(Req) -> Fut + Clone + 'static,
Fut: Future<Output = Result<Res, Err>>,
{
#[inline]
@ -329,7 +229,7 @@ where
}
}
/// Convert `Fn(&Config) -> Future<Service>` fn to NewService
/// Convert `Fn(Config) -> Future<Service>` fn to NewService
pub struct FnServiceConfig<F, Fut, Cfg, Srv, Req, Err>
where
F: Fn(Cfg) -> Fut,
@ -340,17 +240,6 @@ where
_t: PhantomData<(Fut, Cfg, Srv, Req, Err)>,
}
impl<F, Fut, Cfg, Srv, Req, Err> FnServiceConfig<F, Fut, Cfg, Srv, Req, Err>
where
F: Fn(Cfg) -> Fut,
Fut: Future<Output = Result<Srv, Err>>,
Srv: Service<Req>,
{
fn new(f: F) -> Self {
FnServiceConfig { f, _t: PhantomData }
}
}
impl<F, Fut, Cfg, Srv, Req, Err> Clone for FnServiceConfig<F, Fut, Cfg, Srv, Req, Err>
where
F: Fn(Cfg) -> Fut + Clone,
@ -378,26 +267,26 @@ where
type Service = Srv;
type InitError = Err;
type Future = Fut;
type Future<'f> = Fut where Self: 'f, Fut: 'f;
#[inline]
fn new_service(&self, cfg: Cfg) -> Self::Future {
fn create(&self, cfg: Cfg) -> Self::Future<'_> {
(self.f)(cfg)
}
}
/// Converter for `Fn() -> Future<Service>` fn
pub struct FnServiceNoConfig<F, C, S, R, Req, E>
pub struct FnServiceNoConfig<F, S, R, Req, E>
where
F: Fn() -> R,
S: Service<Req>,
R: Future<Output = Result<S, E>>,
{
f: F,
_t: PhantomData<(Req, C)>,
_t: PhantomData<Req>,
}
impl<F, C, S, R, Req, E> FnServiceNoConfig<F, C, S, R, Req, E>
impl<F, S, R, Req, E> FnServiceNoConfig<F, S, R, Req, E>
where
F: Fn() -> R,
R: Future<Output = Result<S, E>>,
@ -408,25 +297,26 @@ where
}
}
impl<F, C, S, R, Req, E> ServiceFactory<Req, C> for FnServiceNoConfig<F, C, S, R, Req, E>
impl<F, S, R, Req, E, C> ServiceFactory<Req, C> for FnServiceNoConfig<F, S, R, Req, E>
where
F: Fn() -> R,
R: Future<Output = Result<S, E>>,
S: Service<Req>,
C: 'static,
{
type Response = S::Response;
type Error = S::Error;
type Service = S;
type InitError = E;
type Future = R;
type Future<'f> = R where Self: 'f, R: 'f;
#[inline]
fn new_service(&self, _: C) -> Self::Future {
fn create(&self, _: C) -> Self::Future<'_> {
(self.f)()
}
}
impl<F, C, S, R, Req, E> Clone for FnServiceNoConfig<F, C, S, R, Req, E>
impl<F, S, R, Req, E> Clone for FnServiceNoConfig<F, S, R, Req, E>
where
F: Fn() -> R + Clone,
R: Future<Output = Result<S, E>>,
@ -438,15 +328,16 @@ where
}
}
impl<F, C, S, R, Req, E> IntoServiceFactory<FnServiceNoConfig<F, C, S, R, Req, E>, Req, C>
impl<F, S, R, Req, E, C> IntoServiceFactory<FnServiceNoConfig<F, S, R, Req, E>, Req, C>
for F
where
F: Fn() -> R,
R: Future<Output = Result<S, E>>,
S: Service<Req>,
C: 'static,
{
#[inline]
fn into_factory(self) -> FnServiceNoConfig<F, C, S, R, Req, E> {
fn into_factory(self) -> FnServiceNoConfig<F, S, R, Req, E> {
FnServiceNoConfig::new(self)
}
}
@ -454,21 +345,16 @@ where
#[cfg(test)]
mod tests {
use ntex_util::future::lazy;
use std::{cell::RefCell, rc::Rc, task::Poll};
use std::task::Poll;
use super::*;
use crate::{Service, ServiceFactory};
#[ntex::test]
async fn test_fn_service() {
let shutdown = Rc::new(RefCell::new(false));
let new_srv = fn_service(|()| async { Ok::<_, ()>("srv") })
.on_shutdown(|| {
*shutdown.borrow_mut() = true;
})
.clone();
let new_srv = fn_service(|()| async { Ok::<_, ()>("srv") }).clone();
let srv = new_srv.new_service(()).await.unwrap();
let srv = new_srv.create(()).await.unwrap();
let res = srv.call(()).await;
assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
assert!(res.is_ok());
@ -479,22 +365,14 @@ mod tests {
assert!(res.is_ok());
assert_eq!(res.unwrap(), "srv");
assert_eq!(
lazy(|cx| srv2.poll_shutdown(cx, false)).await,
Poll::Ready(())
);
assert!(*shutdown.borrow());
assert_eq!(lazy(|cx| srv2.poll_shutdown(cx)).await, Poll::Ready(()));
}
#[ntex::test]
async fn test_fn_service_service() {
let shutdown = Rc::new(RefCell::new(false));
let srv = fn_service(|()| async { Ok::<_, ()>("srv") })
.on_shutdown(|| {
*shutdown.borrow_mut() = true;
})
.clone()
.new_service(())
.create(&())
.await
.unwrap()
.clone();
@ -503,23 +381,22 @@ mod tests {
assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
assert!(res.is_ok());
assert_eq!(res.unwrap(), "srv");
assert_eq!(
lazy(|cx| srv.poll_shutdown(cx, false)).await,
Poll::Ready(())
);
assert!(*shutdown.borrow());
assert_eq!(lazy(|cx| srv.poll_shutdown(cx)).await, Poll::Ready(()));
}
#[ntex::test]
async fn test_fn_service_with_config() {
let new_srv = fn_factory_with_config(|cfg: usize| async move {
Ok::<_, ()>(fn_service(
move |()| async move { Ok::<_, ()>(("srv", cfg)) },
))
let new_srv = fn_factory_with_config(|cfg: &usize| {
let cfg = *cfg;
async move {
Ok::<_, ()>(fn_service(
move |()| async move { Ok::<_, ()>(("srv", cfg)) },
))
}
})
.clone();
let srv = new_srv.new_service(1).await.unwrap();
let srv = new_srv.create(&1).await.unwrap();
let res = srv.call(()).await;
assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
assert!(res.is_ok());

143
ntex-service/src/lib.rs
View file

@ -11,6 +11,7 @@ mod and_then;
mod apply;
pub mod boxed;
mod fn_service;
mod macros;
mod map;
mod map_config;
mod map_err;
@ -21,10 +22,11 @@ mod transform;
pub use self::apply::{apply_fn, apply_fn_factory};
pub use self::fn_service::{fn_factory, fn_factory_with_config, fn_service};
pub use self::map_config::{map_config, map_config_service, unit_config};
pub use self::map_config::{map_config, unit_config};
pub use self::pipeline::{pipeline, pipeline_factory, Pipeline, PipelineFactory};
pub use self::transform::{apply, Identity, Transform};
pub use self::transform::{apply, Identity, Middleware, Stack};
#[allow(unused_variables)]
/// An asynchronous function of `Request` to a `Response`.
///
/// The `Service` trait represents a request/response interaction, receiving requests and returning
@ -62,13 +64,9 @@ pub use self::transform::{apply, Identity, Transform};
/// impl Service<u8> for MyService {
/// type Response = u64;
/// type Error = Infallible;
/// type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
/// type Future<'f> = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
///
/// fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
/// Poll::Ready(Ok(()))
/// }
///
/// fn call(&self, req: u8) -> Self::Future {
/// fn call(&self, req: u8) -> Self::Future<'_> {
/// Box::pin(std::future::ready(Ok(req as u64)))
/// }
/// }
@ -88,8 +86,22 @@ pub trait Service<Req> {
type Error;
/// The future response value.
type Future: Future<Output = Result<Self::Response, Self::Error>>;
type Future<'f>: Future<Output = Result<Self::Response, Self::Error>>
where
Req: 'f,
Self: 'f;
/// Process the request and return the response asynchronously.
///
/// This function is expected to be callable off-task. As such, implementations of `call`
/// should take care to not call `poll_ready`. If the service is at capacity and the request
/// is unable to be handled, the returned `Future` should resolve to an error.
///
/// Invoking `call` without first invoking `poll_ready` is permitted. Implementations must be
/// resilient to this fact.
fn call(&self, req: Req) -> Self::Future<'_>;
#[inline]
/// Returns `Ready` when the service is able to process requests.
///
/// If the service is at capacity, then `Pending` is returned and the task is notified when
@ -103,28 +115,19 @@ pub trait Service<Req> {
///
/// 1. `.poll_ready()` might be called on different task from actual service call.
/// 1. In case of chained services, `.poll_ready()` is called for all services at once.
fn poll_ready(&self, ctx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>>;
fn poll_ready(&self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
#[inline]
#[allow(unused_variables)]
/// Shutdown service.
///
/// Returns `Ready` when the service is properly shutdowns. This method might be called
/// after it returns `Ready`.
fn poll_shutdown(&self, ctx: &mut task::Context<'_>, is_error: bool) -> Poll<()> {
fn poll_shutdown(&self, cx: &mut task::Context<'_>) -> Poll<()> {
Poll::Ready(())
}
/// Process the request and return the response asynchronously.
///
/// This function is expected to be callable off-task. As such, implementations of `call`
/// should take care to not call `poll_ready`. If the service is at capacity and the request
/// is unable to be handled, the returned `Future` should resolve to an error.
///
/// Invoking `call` without first invoking `poll_ready` is permitted. Implementations must be
/// resilient to this fact.
fn call(&self, req: Req) -> Self::Future;
#[inline]
/// Map this service's output to a different type, returning a new service of the resulting type.
///
@ -136,7 +139,7 @@ pub trait Service<Req> {
fn map<F, Res>(self, f: F) -> crate::dev::Map<Self, F, Req, Res>
where
Self: Sized,
F: FnMut(Self::Response) -> Res,
F: Fn(Self::Response) -> Res,
{
crate::dev::Map::new(self, f)
}
@ -183,33 +186,33 @@ pub trait ServiceFactory<Req, Cfg = ()> {
type InitError;
/// The future of the `ServiceFactory` instance.
type Future: Future<Output = Result<Self::Service, Self::InitError>>;
type Future<'f>: Future<Output = Result<Self::Service, Self::InitError>>
where
Cfg: 'f,
Self: 'f;
/// Create and return a new service value asynchronously.
fn new_service(&self, cfg: Cfg) -> Self::Future;
fn create(&self, cfg: Cfg) -> Self::Future<'_>;
#[inline]
/// Map this service's output to a different type, returning a new service
/// of the resulting type.
fn map<F, Res>(self, f: F) -> crate::map::MapServiceFactory<Self, F, Req, Res, Cfg>
fn map<F, Res>(self, f: F) -> crate::map::MapFactory<Self, F, Req, Res, Cfg>
where
Self: Sized,
F: FnMut(Self::Response) -> Res + Clone,
F: Fn(Self::Response) -> Res + Clone,
{
crate::map::MapServiceFactory::new(self, f)
crate::map::MapFactory::new(self, f)
}
#[inline]
/// Map this service's error to a different error, returning a new service.
fn map_err<F, E>(
self,
f: F,
) -> crate::map_err::MapErrServiceFactory<Self, Req, Cfg, F, E>
fn map_err<F, E>(self, f: F) -> crate::map_err::MapErrFactory<Self, Req, Cfg, F, E>
where
Self: Sized,
F: Fn(Self::Error) -> E + Clone,
{
crate::map_err::MapErrServiceFactory::<_, _, Cfg, _, _>::new(self, f)
crate::map_err::MapErrFactory::new(self, f)
}
#[inline]
@ -226,13 +229,13 @@ pub trait ServiceFactory<Req, Cfg = ()> {
}
}
impl<S, Req> Service<Req> for Box<S>
impl<'a, S, Req> Service<Req> for &'a S
where
S: Service<Req> + ?Sized,
{
type Response = S::Response;
type Error = S::Error;
type Future = S::Future;
type Future<'f> = S::Future<'f> where 'a: 'f, Req: 'f;
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
@ -240,12 +243,36 @@ where
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
(**self).poll_shutdown(cx, is_error)
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
(**self).poll_shutdown(cx)
}
#[inline]
fn call(&self, request: Req) -> S::Future {
fn call(&self, request: Req) -> S::Future<'_> {
(**self).call(request)
}
}
impl<S, Req> Service<Req> for Box<S>
where
S: Service<Req> + ?Sized,
{
type Response = S::Response;
type Error = S::Error;
type Future<'f> = S::Future<'f> where S: 'f, Req: 'f;
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
(**self).poll_ready(cx)
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
(**self).poll_shutdown(cx)
}
#[inline]
fn call(&self, request: Req) -> S::Future<'_> {
(**self).call(request)
}
}
@ -256,18 +283,18 @@ where
{
type Response = S::Response;
type Error = S::Error;
type Future = S::Future;
type Future<'f> = S::Future<'f> where S: 'f, Req: 'f;
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
(**self).poll_ready(cx)
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
(**self).poll_shutdown(cx, is_error)
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
(**self).poll_shutdown(cx)
}
fn call(&self, request: Req) -> S::Future {
fn call(&self, request: Req) -> S::Future<'_> {
(**self).call(request)
}
}
@ -280,20 +307,20 @@ where
type Error = S::Error;
type Service = S::Service;
type InitError = S::InitError;
type Future = S::Future;
type Future<'f> = S::Future<'f> where S: 'f, Cfg: 'f;
fn new_service(&self, cfg: Cfg) -> S::Future {
self.as_ref().new_service(cfg)
fn create(&self, cfg: Cfg) -> S::Future<'_> {
self.as_ref().create(cfg)
}
}
/// Trait for types that can be converted to a `Service`
pub trait IntoService<T, Req>
pub trait IntoService<Svc, Req>
where
T: Service<Req>,
Svc: Service<Req>,
{
/// Convert to a `Service`
fn into_service(self) -> T;
fn into_service(self) -> Svc;
}
/// Trait for types that can be converted to a `ServiceFactory`
@ -305,11 +332,11 @@ where
fn into_factory(self) -> T;
}
impl<T, Req> IntoService<T, Req> for T
impl<Svc, Req> IntoService<Svc, Req> for Svc
where
T: Service<Req>,
Svc: Service<Req>,
{
fn into_service(self) -> T {
fn into_service(self) -> Svc {
self
}
}
@ -324,24 +351,24 @@ where
}
/// Convert object of type `T` to a service `S`
pub fn into_service<T, S, Req>(tp: T) -> S
pub fn into_service<Svc, Req, F>(tp: F) -> Svc
where
S: Service<Req>,
T: IntoService<S, Req>,
Svc: Service<Req>,
F: IntoService<Svc, Req>,
{
tp.into_service()
}
pub mod dev {
pub use crate::and_then::{AndThen, AndThenFactory};
pub use crate::apply::{Apply, ApplyServiceFactory};
pub use crate::apply::{Apply, ApplyFactory};
pub use crate::fn_service::{
FnService, FnServiceConfig, FnServiceFactory, FnServiceNoConfig,
};
pub use crate::map::{Map, MapServiceFactory};
pub use crate::map::{Map, MapFactory};
pub use crate::map_config::{MapConfig, UnitConfig};
pub use crate::map_err::{MapErr, MapErrServiceFactory};
pub use crate::map_err::{MapErr, MapErrFactory};
pub use crate::map_init_err::MapInitErr;
pub use crate::then::{Then, ThenFactory};
pub use crate::transform::ApplyTransform;
pub use crate::transform::ApplyMiddleware;
}

38
ntex-service/src/macros.rs Normal file
View file

@ -0,0 +1,38 @@
/// An implementation of [`poll_ready`] that forwards readiness checks to a field.
#[macro_export]
macro_rules! forward_poll_ready {
($field:ident) => {
#[inline]
fn poll_ready(
&self,
cx: &mut ::core::task::Context<'_>,
) -> ::core::task::Poll<Result<(), Self::Error>> {
self.$field
.poll_ready(cx)
.map_err(::core::convert::Into::into)
}
};
($field:ident, $err:expr) => {
#[inline]
fn poll_ready(
&self,
cx: &mut ::core::task::Context<'_>,
) -> ::core::task::Poll<Result<(), Self::Error>> {
self.$field.poll_ready(cx).map_err($err)
}
};
}
/// An implementation of [`poll_shutdown`] that forwards readiness checks to a field.
#[macro_export]
macro_rules! forward_poll_shutdown {
($field:ident) => {
#[inline]
fn poll_shutdown(
&self,
cx: &mut ::core::task::Context<'_>,
) -> ::core::task::Poll<()> {
self.$field.poll_shutdown(cx)
}
};
}

135
ntex-service/src/map.rs
View file

@ -16,7 +16,7 @@ impl<A, F, Req, Res> Map<A, F, Req, Res> {
pub(crate) fn new(service: A, f: F) -> Self
where
A: Service<Req>,
F: FnMut(A::Response) -> Res,
F: Fn(A::Response) -> Res,
{
Self {
service,
@ -44,62 +44,51 @@ where
impl<A, F, Req, Res> Service<Req> for Map<A, F, Req, Res>
where
A: Service<Req>,
F: FnMut(A::Response) -> Res + Clone,
F: Fn(A::Response) -> Res,
{
type Response = Res;
type Error = A::Error;
type Future = MapFuture<A, F, Req, Res>;
type Future<'f> = MapFuture<'f, A, F, Req, Res> where Self: 'f, Req: 'f;
crate::forward_poll_ready!(service);
crate::forward_poll_shutdown!(service);
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.service.poll_ready(cx)
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
self.service.poll_shutdown(cx, is_error)
}
#[inline]
fn call(&self, req: Req) -> Self::Future {
MapFuture::new(self.service.call(req), self.f.clone())
fn call(&self, req: Req) -> Self::Future<'_> {
MapFuture {
fut: self.service.call(req),
slf: self,
}
}
}
pin_project_lite::pin_project! {
pub struct MapFuture<A, F, Req, Res>
pub struct MapFuture<'f, A, F, Req, Res>
where
A: Service<Req>,
F: FnMut(A::Response) -> Res,
A: 'f,
Req: 'f,
F: Fn(A::Response) -> Res,
{
f: F,
slf: &'f Map<A, F, Req, Res>,
#[pin]
fut: A::Future,
fut: A::Future<'f>,
}
}
impl<A, F, Req, Res> MapFuture<A, F, Req, Res>
impl<'f, A, F, Req, Res> Future for MapFuture<'f, A, F, Req, Res>
where
A: Service<Req>,
F: FnMut(A::Response) -> Res,
{
fn new(fut: A::Future, f: F) -> Self {
MapFuture { f, fut }
}
}
impl<A, F, Req, Res> Future for MapFuture<A, F, Req, Res>
where
A: Service<Req>,
F: FnMut(A::Response) -> Res,
A: Service<Req> + 'f,
Req: 'f,
F: Fn(A::Response) -> Res,
{
type Output = Result<Res, A::Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.as_mut().project();
match this.fut.poll(cx) {
Poll::Ready(Ok(resp)) => Poll::Ready(Ok((this.f)(resp))),
Poll::Ready(Ok(resp)) => Poll::Ready(Ok((self.project().slf.f)(resp))),
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
Poll::Pending => Poll::Pending,
}
@ -107,19 +96,19 @@ where
}
/// `MapNewService` new service combinator
pub struct MapServiceFactory<A, F, Req, Res, Cfg> {
pub struct MapFactory<A, F, Req, Res, Cfg> {
a: A,
f: F,
r: PhantomData<fn(Req, Cfg) -> Res>,
}
impl<A, F, Req, Res, Cfg> MapServiceFactory<A, F, Req, Res, Cfg> {
impl<A, F, Req, Res, Cfg> MapFactory<A, F, Req, Res, Cfg>
where
A: ServiceFactory<Req, Cfg>,
F: Fn(A::Response) -> Res,
{
/// Create new `Map` new service instance
pub(crate) fn new(a: A, f: F) -> Self
where
A: ServiceFactory<Req, Cfg>,
F: FnMut(A::Response) -> Res,
{
pub(crate) fn new(a: A, f: F) -> Self {
Self {
a,
f,
@ -128,7 +117,7 @@ impl<A, F, Req, Res, Cfg> MapServiceFactory<A, F, Req, Res, Cfg> {
}
}
impl<A, F, Req, Res, Cfg> Clone for MapServiceFactory<A, F, Req, Res, Cfg>
impl<A, F, Req, Res, Cfg> Clone for MapFactory<A, F, Req, Res, Cfg>
where
A: Clone,
F: Clone,
@ -143,51 +132,45 @@ where
}
}
impl<A, F, Req, Res, Cfg> ServiceFactory<Req, Cfg>
for MapServiceFactory<A, F, Req, Res, Cfg>
impl<A, F, Req, Res, Cfg> ServiceFactory<Req, Cfg> for MapFactory<A, F, Req, Res, Cfg>
where
A: ServiceFactory<Req, Cfg>,
F: FnMut(A::Response) -> Res + Clone,
F: Fn(A::Response) -> Res + Clone,
{
type Response = Res;
type Error = A::Error;
type Service = Map<A::Service, F, Req, Res>;
type InitError = A::InitError;
type Future = MapServiceFuture<A, F, Req, Res, Cfg>;
type Future<'f> = MapFactoryFuture<'f, A, F, Req, Res, Cfg> where Self: 'f, Cfg: 'f;
#[inline]
fn new_service(&self, cfg: Cfg) -> Self::Future {
MapServiceFuture::new(self.a.new_service(cfg), self.f.clone())
fn create(&self, cfg: Cfg) -> Self::Future<'_> {
MapFactoryFuture {
fut: self.a.create(cfg),
f: Some(self.f.clone()),
}
}
}
pin_project_lite::pin_project! {
pub struct MapServiceFuture<A, F, Req, Res, Cfg>
pub struct MapFactoryFuture<'f, A, F, Req, Res, Cfg>
where
A: ServiceFactory<Req, Cfg>,
F: FnMut(A::Response) -> Res,
A: 'f,
F: Fn(A::Response) -> Res,
Cfg: 'f,
{
#[pin]
fut: A::Future,
fut: A::Future<'f>,
f: Option<F>,
}
}
impl<A, F, Req, Res, Cfg> MapServiceFuture<A, F, Req, Res, Cfg>
impl<'f, A, F, Req, Res, Cfg> Future for MapFactoryFuture<'f, A, F, Req, Res, Cfg>
where
A: ServiceFactory<Req, Cfg>,
F: FnMut(A::Response) -> Res,
{
fn new(fut: A::Future, f: F) -> Self {
MapServiceFuture { f: Some(f), fut }
}
}
impl<A, F, Req, Res, Cfg> Future for MapServiceFuture<A, F, Req, Res, Cfg>
where
A: ServiceFactory<Req, Cfg>,
F: FnMut(A::Response) -> Res,
F: Fn(A::Response) -> Res,
{
type Output = Result<Map<A::Service, F, Req, Res>, A::InitError>;
@ -207,7 +190,7 @@ mod tests {
use ntex_util::future::{lazy, Ready};
use super::*;
use crate::{IntoServiceFactory, Service, ServiceFactory};
use crate::{fn_factory, Service, ServiceFactory};
#[derive(Clone)]
struct Srv;
@ -215,13 +198,13 @@ mod tests {
impl Service<()> for Srv {
type Response = ();
type Error = ();
type Future = Ready<(), ()>;
type Future<'f> = Ready<(), ()>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&self, _: ()) -> Self::Future {
fn call(&self, _: ()) -> Self::Future<'_> {
Ready::Ok(())
}
}
@ -236,7 +219,7 @@ mod tests {
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert_eq!(res, Poll::Ready(Ok(())));
let res = lazy(|cx| srv.poll_shutdown(cx, true)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
}
@ -250,17 +233,16 @@ mod tests {
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert_eq!(res, Poll::Ready(Ok(())));
let res = lazy(|cx| srv.poll_shutdown(cx, true)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
}
#[ntex::test]
async fn test_factory() {
let new_srv = (|| async { Ok::<_, ()>(Srv) })
.into_factory()
let new_srv = fn_factory(|| async { Ok::<_, ()>(Srv) })
.map(|_| "ok")
.clone();
let srv = new_srv.new_service(&()).await.unwrap();
let srv = new_srv.create(&()).await.unwrap();
let res = srv.call(()).await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), ("ok"));
@ -268,11 +250,10 @@ mod tests {
#[ntex::test]
async fn test_pipeline_factory() {
let new_srv =
crate::pipeline_factory((|| async { Ok::<_, ()>(Srv) }).into_factory())
.map(|_| "ok")
.clone();
let srv = new_srv.new_service(&()).await.unwrap();
let new_srv = crate::pipeline_factory(fn_factory(|| async { Ok::<_, ()>(Srv) }))
.map(|_| "ok")
.clone();
let srv = new_srv.create(&()).await.unwrap();
let res = srv.call(()).await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), ("ok"));

270
ntex-service/src/map_config.rs
View file

@ -1,13 +1,12 @@
use std::task::{Context, Poll};
use std::{cell::RefCell, future::Future, marker::PhantomData, pin::Pin, rc::Rc};
use std::{future::Future, marker::PhantomData, pin::Pin, task::Context, task::Poll};
use super::{IntoServiceFactory, Service, ServiceFactory};
use super::{IntoServiceFactory, ServiceFactory};
/// Adapt external config argument to a config for provided service factory
///
/// Note that this function consumes the receiving service factory and returns
/// a wrapped version of it.
pub fn map_config<T, R, U, F, C, C2>(factory: U, f: F) -> MapConfig<T, R, F, C, C2>
pub fn map_config<T, R, U, F, C, C2>(factory: U, f: F) -> MapConfig<T, F, C, C2>
where
T: ServiceFactory<R, C2>,
U: IntoServiceFactory<T, R, C2>,
@ -16,45 +15,25 @@ where
MapConfig::new(factory.into_factory(), f)
}
/// Adapt external config argument to a config for provided service factory
///
/// This function uses service for converting config.
pub fn map_config_service<T, R, M, C, C2, U1, U2>(
factory: U1,
mapper: U2,
) -> MapConfigService<T, R, M, C, C2>
where
T: ServiceFactory<R, C2>,
M: ServiceFactory<C, (), Response = C2, Error = T::InitError, InitError = T::InitError>,
U1: IntoServiceFactory<T, R, C2>,
U2: IntoServiceFactory<M, C, ()>,
{
MapConfigService::new(factory.into_factory(), mapper.into_factory())
}
/// Replace config with unit
pub fn unit_config<T, R, U, C>(factory: U) -> UnitConfig<T, R, C>
pub fn unit_config<T, R, U>(factory: U) -> UnitConfig<T>
where
T: ServiceFactory<R, ()>,
U: IntoServiceFactory<T, R, ()>,
T: ServiceFactory<R>,
U: IntoServiceFactory<T, R>,
{
UnitConfig::new(factory.into_factory())
}
/// `map_config()` adapter service factory
pub struct MapConfig<A, R, F, C, C2> {
pub struct MapConfig<A, F, C, C2> {
a: A,
f: F,
e: PhantomData<(R, C, C2)>,
e: PhantomData<(C, C2)>,
}
impl<A, R, F, C, C2> MapConfig<A, R, F, C, C2> {
impl<A, F, C, C2> MapConfig<A, F, C, C2> {
/// Create new `MapConfig` combinator
pub(crate) fn new(a: A, f: F) -> Self
where
A: ServiceFactory<R, C2>,
F: Fn(C) -> C2,
{
pub(crate) fn new(a: A, f: F) -> Self {
Self {
a,
f,
@ -63,7 +42,7 @@ impl<A, R, F, C, C2> MapConfig<A, R, F, C, C2> {
}
}
impl<A, R, F, C, C2> Clone for MapConfig<A, R, F, C, C2>
impl<A, F, C, C2> Clone for MapConfig<A, F, C, C2>
where
A: Clone,
F: Clone,
@ -77,7 +56,7 @@ where
}
}
impl<A, R, F, C, C2> ServiceFactory<R, C> for MapConfig<A, R, F, C, C2>
impl<A, F, R, C, C2> ServiceFactory<R, C> for MapConfig<A, F, C, C2>
where
A: ServiceFactory<R, C2>,
F: Fn(C) -> C2,
@ -87,195 +66,74 @@ where
type Service = A::Service;
type InitError = A::InitError;
type Future = A::Future;
type Future<'f> = A::Future<'f> where Self: 'f;
fn new_service(&self, cfg: C) -> Self::Future {
self.a.new_service((self.f)(cfg))
fn create(&self, cfg: C) -> Self::Future<'_> {
let cfg = (self.f)(cfg);
self.a.create(cfg)
}
}
/// `unit_config()` config combinator
pub struct UnitConfig<A, R, C> {
pub struct UnitConfig<A> {
a: A,
e: PhantomData<(C, R)>,
}
impl<A, R, C> UnitConfig<A, R, C>
where
A: ServiceFactory<R, ()>,
{
impl<A> UnitConfig<A> {
/// Create new `UnitConfig` combinator
pub(crate) fn new(a: A) -> Self {
Self { a, e: PhantomData }
Self { a }
}
}
impl<A, R, C> Clone for UnitConfig<A, R, C>
impl<A> Clone for UnitConfig<A>
where
A: Clone,
{
fn clone(&self) -> Self {
Self {
a: self.a.clone(),
e: PhantomData,
}
Self { a: self.a.clone() }
}
}
impl<A, R, C> ServiceFactory<R, C> for UnitConfig<A, R, C>
impl<A, R, C> ServiceFactory<R, C> for UnitConfig<A>
where
A: ServiceFactory<R, ()>,
A: ServiceFactory<R>,
{
type Response = A::Response;
type Error = A::Error;
type Service = A::Service;
type InitError = A::InitError;
type Future = A::Future;
type Future<'f> = UnitConfigFuture<'f, A, R, C> where Self: 'f, C: 'f;
fn new_service(&self, _: C) -> Self::Future {
self.a.new_service(())
}
}
/// `map_config_service()` adapter service factory
pub struct MapConfigService<A, R, M: ServiceFactory<C, ()>, C, C2>(
Rc<Inner<A, R, M, C, C2>>,
);
struct Inner<A, R, M: ServiceFactory<C, ()>, C, C2> {
a: A,
m: M,
mapper: RefCell<Option<M::Service>>,
e: PhantomData<(R, C, C2)>,
}
impl<A, R, M: ServiceFactory<C, ()>, C, C2> Clone for MapConfigService<A, R, M, C, C2> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl<A, R, M: ServiceFactory<C, ()>, C, C2> MapConfigService<A, R, M, C, C2> {
/// Create new `MapConfigService` combinator
pub(crate) fn new(a: A, m: M) -> Self
where
A: ServiceFactory<R, C2>,
M: ServiceFactory<
C,
(),
Response = C2,
Error = A::InitError,
InitError = A::InitError,
>,
{
Self(Rc::new(Inner {
a,
m,
mapper: RefCell::new(None),
e: PhantomData,
}))
}
}
impl<A, R, M, C, C2> ServiceFactory<R, C> for MapConfigService<A, R, M, C, C2>
where
A: ServiceFactory<R, C2>,
M: ServiceFactory<C, (), Response = C2, Error = A::InitError, InitError = A::InitError>,
{
type Response = A::Response;
type Error = A::Error;
type Service = A::Service;
type InitError = A::InitError;
type Future = MapConfigServiceResponse<A, R, M, C, C2>;
fn new_service(&self, cfg: C) -> Self::Future {
let inner = self.0.clone();
if self.0.mapper.borrow().is_some() {
MapConfigServiceResponse {
inner,
config: Some(cfg),
state: ResponseState::MapReady,
}
} else {
MapConfigServiceResponse {
inner,
config: Some(cfg),
state: ResponseState::CreateMapper {
fut: self.0.m.new_service(()),
},
}
fn create(&self, _: C) -> Self::Future<'_> {
UnitConfigFuture {
fut: self.a.create(()),
_t: PhantomData,
}
}
}
pin_project_lite::pin_project! {
pub struct MapConfigServiceResponse<A, R, M, C, C2>
where
A: ServiceFactory<R, C2>,
M: ServiceFactory<C, ()>,
pub struct UnitConfigFuture<'f, A, R, C>
where A: ServiceFactory<R>,
A: 'f,
C: 'f,
{
inner: Rc<Inner<A, R, M, C, C2>>,
config: Option<C>,
#[pin]
state: ResponseState<A, R, M, C, C2>,
fut: A::Future<'f>,
_t: PhantomData<C>,
}
}
pin_project_lite::pin_project! {
#[project = ResponseStateProject]
enum ResponseState<A: ServiceFactory<R, C2>, R, M: ServiceFactory<C, ()>, C, C2> {
CreateMapper { #[pin] fut: M::Future },
MapReady,
MapConfig { #[pin] fut: <M::Service as Service<C>>::Future },
CreateService { #[pin] fut: A::Future },
}
}
impl<A, R, M, C, C2> Future for MapConfigServiceResponse<A, R, M, C, C2>
impl<'f, A, R, C> Future for UnitConfigFuture<'f, A, R, C>
where
A: ServiceFactory<R, C2>,
M: ServiceFactory<C, (), Response = C2, Error = A::InitError, InitError = A::InitError>,
A: ServiceFactory<R>,
{
type Output = Result<A::Service, A::InitError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
ResponseStateProject::CreateMapper { fut } => {
let mapper = match fut.poll(cx) {
Poll::Ready(result) => result?,
Poll::Pending => return Poll::Pending,
};
*this.inner.mapper.borrow_mut() = Some(mapper);
this.state.set(ResponseState::MapReady);
self.poll(cx)
}
ResponseStateProject::MapReady => {
let mapper = this.inner.mapper.borrow();
match mapper.as_ref().unwrap().poll_ready(cx) {
Poll::Ready(result) => result?,
Poll::Pending => return Poll::Pending,
};
let fut = mapper.as_ref().unwrap().call(this.config.take().unwrap());
this.state.set(ResponseState::MapConfig { fut });
drop(mapper);
self.poll(cx)
}
ResponseStateProject::MapConfig { fut } => {
let config = match fut.poll(cx) {
Poll::Ready(result) => result?,
Poll::Pending => return Poll::Pending,
};
let fut = this.inner.a.new_service(config);
this.state.set(ResponseState::CreateService { fut });
self.poll(cx)
}
ResponseStateProject::CreateService { fut } => fut.poll(cx),
}
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.project().fut.poll(cx)
}
}
@ -286,7 +144,7 @@ mod tests {
use std::{cell::Cell, rc::Rc};
use super::*;
use crate::{fn_factory_with_config, fn_service, ServiceFactory};
use crate::{fn_service, ServiceFactory};
#[ntex::test]
async fn test_map_config() {
@ -294,13 +152,13 @@ mod tests {
let factory = map_config(
fn_service(|item: usize| Ready::<_, ()>::Ok(item)),
|t: usize| {
item.set(item.get() + t);
|t: &usize| {
item.set(item.get() + *t);
},
)
.clone();
let _ = factory.new_service(10).await;
let _ = factory.create(&10).await;
assert_eq!(item.get(), 11);
}
@ -308,31 +166,31 @@ mod tests {
async fn test_unit_config() {
let _ = unit_config(fn_service(|item: usize| Ready::<_, ()>::Ok(item)))
.clone()
.new_service(10)
.create(&10)
.await;
}
#[ntex::test]
async fn test_map_config_service() {
let item = Rc::new(Cell::new(10usize));
let item2 = item.clone();
// #[ntex::test]
// async fn test_map_config_service() {
// let item = Rc::new(Cell::new(10usize));
// let item2 = item.clone();
let srv = map_config_service(
fn_factory_with_config(move |next: usize| {
let item = item2.clone();
async move {
item.set(next);
Ok::<_, ()>(fn_service(|id: usize| Ready::<_, ()>::Ok(id * 2)))
}
}),
fn_service(move |item: usize| Ready::<_, ()>::Ok(item + 1)),
)
.clone()
.new_service(10)
.await
.unwrap();
// let srv = map_config_service(
// fn_factory_with_config(move |next: usize| {
// let item = item2.clone();
// async move {
// item.set(next);
// Ok::<_, ()>(fn_service(|id: usize| Ready::<_, ()>::Ok(id * 2)))
// }
// }),
// fn_service(move |item: usize| Ready::<_, ()>::Ok(item + 1)),
// )
// .clone()
// .create(10)
// .await
// .unwrap();
assert_eq!(srv.call(10usize).await.unwrap(), 20);
assert_eq!(item.get(), 11);
}
// assert_eq!(srv.call(10usize).await.unwrap(), 20);
// assert_eq!(item.get(), 11);
// }
}

103
ntex-service/src/map_err.rs
View file

@ -45,11 +45,11 @@ where
impl<A, R, F, E> Service<R> for MapErr<A, R, F, E>
where
A: Service<R>,
F: Fn(A::Error) -> E + Clone,
F: Fn(A::Error) -> E,
{
type Response = A::Response;
type Error = E;
type Future = MapErrFuture<A, R, F, E>;
type Future<'f> = MapErrFuture<'f, A, R, F, E> where A: 'f, R: 'f, F: 'f, E: 'f;
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
@ -57,48 +57,40 @@ where
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
self.service.poll_shutdown(cx, is_error)
fn call(&self, req: R) -> Self::Future<'_> {
MapErrFuture {
slf: self,
fut: self.service.call(req),
}
}
#[inline]
fn call(&self, req: R) -> Self::Future {
MapErrFuture::new(self.service.call(req), self.f.clone())
}
crate::forward_poll_shutdown!(service);
}
pin_project_lite::pin_project! {
pub struct MapErrFuture<A, R, F, E>
pub struct MapErrFuture<'f, A, R, F, E>
where
A: Service<R>,
A: 'f,
R: 'f,
F: Fn(A::Error) -> E,
{
f: F,
slf: &'f MapErr<A, R, F, E>,
#[pin]
fut: A::Future,
fut: A::Future<'f>,
}
}
impl<A, R, F, E> MapErrFuture<A, R, F, E>
impl<'f, A, R, F, E> Future for MapErrFuture<'f, A, R, F, E>
where
A: Service<R>,
F: Fn(A::Error) -> E,
{
fn new(fut: A::Future, f: F) -> Self {
MapErrFuture { f, fut }
}
}
impl<A, R, F, E> Future for MapErrFuture<A, R, F, E>
where
A: Service<R>,
A: Service<R> + 'f,
F: Fn(A::Error) -> E,
{
type Output = Result<A::Response, E>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
this.fut.poll(cx).map_err(this.f)
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.as_mut().project();
this.fut.poll(cx).map_err(|e| (self.project().slf.f)(e))
}
}
@ -106,7 +98,7 @@ where
/// service's error.
///
/// This is created by the `NewServiceExt::map_err` method.
pub struct MapErrServiceFactory<A, R, C, F, E>
pub struct MapErrFactory<A, R, C, F, E>
where
A: ServiceFactory<R, C>,
F: Fn(A::Error) -> E + Clone,
@ -116,7 +108,7 @@ where
e: PhantomData<fn(R, C) -> E>,
}
impl<A, R, C, F, E> MapErrServiceFactory<A, R, C, F, E>
impl<A, R, C, F, E> MapErrFactory<A, R, C, F, E>
where
A: ServiceFactory<R, C>,
F: Fn(A::Error) -> E + Clone,
@ -131,7 +123,7 @@ where
}
}
impl<A, R, C, F, E> Clone for MapErrServiceFactory<A, R, C, F, E>
impl<A, R, C, F, E> Clone for MapErrFactory<A, R, C, F, E>
where
A: ServiceFactory<R, C> + Clone,
F: Fn(A::Error) -> E + Clone,
@ -145,7 +137,7 @@ where
}
}
impl<A, R, C, F, E> ServiceFactory<R, C> for MapErrServiceFactory<A, R, C, F, E>
impl<A, R, C, F, E> ServiceFactory<R, C> for MapErrFactory<A, R, C, F, E>
where
A: ServiceFactory<R, C>,
F: Fn(A::Error) -> E + Clone,
@ -155,37 +147,32 @@ where
type Service = MapErr<A::Service, R, F, E>;
type InitError = A::InitError;
type Future = MapErrServiceFuture<A, R, C, F, E>;
type Future<'f> = MapErrFactoryFuture<'f, A, R, C, F, E> where Self: 'f, C: 'f;
#[inline]
fn new_service(&self, cfg: C) -> Self::Future {
MapErrServiceFuture::new(self.a.new_service(cfg), self.f.clone())
fn create(&self, cfg: C) -> Self::Future<'_> {
MapErrFactoryFuture {
f: self.f.clone(),
fut: self.a.create(cfg),
}
}
}
pin_project_lite::pin_project! {
pub struct MapErrServiceFuture<A, R, C, F, E>
pub struct MapErrFactoryFuture<'f, A, R, C, F, E>
where
A: ServiceFactory<R, C>,
A: 'f,
F: Fn(A::Error) -> E,
C: 'f,
{
#[pin]
fut: A::Future,
f: F,
#[pin]
fut: A::Future<'f>,
}
}
impl<A, R, C, F, E> MapErrServiceFuture<A, R, C, F, E>
where
A: ServiceFactory<R, C>,
F: Fn(A::Error) -> E,
{
fn new(fut: A::Future, f: F) -> Self {
MapErrServiceFuture { fut, f }
}
}
impl<A, R, C, F, E> Future for MapErrServiceFuture<A, R, C, F, E>
impl<'f, A, R, C, F, E> Future for MapErrFactoryFuture<'f, A, R, C, F, E>
where
A: ServiceFactory<R, C>,
F: Fn(A::Error) -> E + Clone,
@ -205,7 +192,7 @@ where
#[cfg(test)]
mod tests {
use super::*;
use crate::{IntoServiceFactory, Service, ServiceFactory};
use crate::{fn_factory, Service, ServiceFactory};
use ntex_util::future::{lazy, Ready};
#[derive(Clone)]
@ -214,13 +201,13 @@ mod tests {
impl Service<()> for Srv {
type Response = ();
type Error = ();
type Future = Ready<(), ()>;
type Future<'f> = Ready<(), ()>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Err(()))
}
fn call(&self, _: ()) -> Self::Future {
fn call(&self, _: ()) -> Self::Future<'_> {
Ready::Err(())
}
}
@ -231,7 +218,7 @@ mod tests {
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert_eq!(res, Poll::Ready(Err("error")));
let res = lazy(|cx| srv.poll_shutdown(cx, true)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
}
@ -253,11 +240,10 @@ mod tests {
#[ntex::test]
async fn test_factory() {
let new_srv = (|| Ready::<_, ()>::Ok(Srv))
.into_factory()
let new_srv = fn_factory(|| Ready::<_, ()>::Ok(Srv))
.map_err(|_| "error")
.clone();
let srv = new_srv.new_service(&()).await.unwrap();
let srv = new_srv.create(&()).await.unwrap();
let res = srv.call(()).await;
assert!(res.is_err());
assert_eq!(res.err().unwrap(), "error");
@ -265,11 +251,10 @@ mod tests {
#[ntex::test]
async fn test_pipeline_factory() {
let new_srv =
crate::pipeline_factory((|| async { Ok::<_, ()>(Srv) }).into_factory())
.map_err(|_| "error")
.clone();
let srv = new_srv.new_service(&()).await.unwrap();
let new_srv = crate::pipeline_factory(fn_factory(|| async { Ok::<Srv, ()>(Srv) }))
.map_err(|_| "error")
.clone();
let srv = new_srv.create(&()).await.unwrap();
let res = srv.call(()).await;
assert!(res.is_err());
assert_eq!(res.err().unwrap(), "error");

40
ntex-service/src/map_init_err.rs
View file

@ -48,58 +48,64 @@ where
type Service = A::Service;
type InitError = E;
type Future = MapInitErrFuture<A, R, C, F, E>;
type Future<'f> = MapInitErrFuture<'f, A, R, C, F, E> where Self: 'f, C: 'f;
fn new_service(&self, cfg: C) -> Self::Future {
#[inline]
fn create(&self, cfg: C) -> Self::Future<'_> {
MapInitErrFuture {
fut: self.a.new_service(cfg),
f: self.f.clone(),
fut: self.a.create(cfg),
}
}
}
pin_project_lite::pin_project! {
pub struct MapInitErrFuture<A, R, C, F, E>
pub struct MapInitErrFuture<'f, A, R, C, F, E>
where
A: ServiceFactory<R, C>,
A: 'f,
F: Fn(A::InitError) -> E,
C: 'f,
{
f: F,
#[pin]
fut: A::Future,
fut: A::Future<'f>,
}
}
impl<A, R, C, F, E> Future for MapInitErrFuture<A, R, C, F, E>
impl<'f, A, R, C, F, E> Future for MapInitErrFuture<'f, A, R, C, F, E>
where
A: ServiceFactory<R, C>,
F: Fn(A::InitError) -> E,
{
type Output = Result<A::Service, E>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
this.fut.poll(cx).map_err(this.f)
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.as_mut().project();
this.fut.poll(cx).map_err(|e| (self.project().f)(e))
}
}
#[cfg(test)]
mod tests {
use crate::{fn_factory_with_config, fn_service, pipeline_factory, ServiceFactory};
use crate::{fn_factory_with_config, into_service, pipeline_factory, ServiceFactory};
#[ntex::test]
async fn map_init_err() {
let factory = pipeline_factory(fn_factory_with_config(|err: bool| async move {
if err {
Err(())
} else {
Ok(fn_service(|i: usize| async move { Ok::<_, ()>(i * 2) }))
let factory = pipeline_factory(fn_factory_with_config(|err: &bool| {
let err = *err;
async move {
if err {
Err(())
} else {
Ok(into_service(|i: usize| async move { Ok::<_, ()>(i * 2) }))
}
}
}))
.map_init_err(|_| std::io::Error::new(std::io::ErrorKind::Other, "err"))
.clone();
assert!(factory.new_service(true).await.is_err());
assert!(factory.new_service(false).await.is_ok());
assert!(factory.create(&true).await.is_err());
assert!(factory.create(&false).await.is_ok());
}
}

122
ntex-service/src/pipeline.rs
View file

@ -1,18 +1,18 @@
use std::{marker::PhantomData, task::Context, task::Poll};
use std::marker::PhantomData;
use crate::and_then::{AndThen, AndThenFactory};
use crate::map::{Map, MapServiceFactory};
use crate::map_err::{MapErr, MapErrServiceFactory};
use crate::map::{Map, MapFactory};
use crate::map_err::{MapErr, MapErrFactory};
use crate::map_init_err::MapInitErr;
use crate::then::{Then, ThenFactory};
use crate::transform::{ApplyTransform, Transform};
use crate::transform::{ApplyMiddleware, Middleware};
use crate::{IntoService, IntoServiceFactory, Service, ServiceFactory};
/// Constructs new pipeline with one service in pipeline chain.
pub fn pipeline<T, R, F>(service: F) -> Pipeline<T, R>
pub fn pipeline<Svc, Req, F>(service: F) -> Pipeline<Req, Svc>
where
T: Service<R>,
F: IntoService<T, R>,
Svc: Service<Req>,
F: IntoService<Svc, Req>,
{
Pipeline {
service: service.into_service(),
@ -21,7 +21,7 @@ where
}
/// Constructs new pipeline factory with one service factory.
pub fn pipeline_factory<T, R, C, F>(factory: F) -> PipelineFactory<T, R, C>
pub fn pipeline_factory<T, R, C, F>(factory: F) -> PipelineFactory<R, T, C>
where
T: ServiceFactory<R, C>,
F: IntoServiceFactory<T, R, C>,
@ -33,12 +33,12 @@ where
}
/// Pipeline service - pipeline allows to compose multiple service into one service.
pub struct Pipeline<T, R> {
service: T,
_t: PhantomData<R>,
pub struct Pipeline<Req, Svc> {
service: Svc,
_t: PhantomData<Req>,
}
impl<T: Service<R>, R> Pipeline<T, R> {
impl<Req, Svc: Service<Req>> Pipeline<Req, Svc> {
/// Call another service after call to this one has resolved successfully.
///
/// This function can be used to chain two services together and ensure that
@ -48,11 +48,11 @@ impl<T: Service<R>, R> Pipeline<T, R> {
///
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it.
pub fn and_then<F, U>(self, service: F) -> Pipeline<AndThen<T, U, R>, R>
pub fn and_then<Next, F>(self, service: F) -> Pipeline<Req, AndThen<Svc, Next>>
where
Self: Sized,
F: IntoService<U, T::Response>,
U: Service<T::Response, Error = T::Error>,
F: IntoService<Next, Svc::Response>,
Next: Service<Svc::Response, Error = Svc::Error>,
{
Pipeline {
service: AndThen::new(self.service, service.into_service()),
@ -65,11 +65,11 @@ impl<T: Service<R>, R> Pipeline<T, R> {
///
/// Note that this function consumes the receiving pipeline and returns a
/// wrapped version of it.
pub fn then<F, U>(self, service: F) -> Pipeline<Then<T, U, R>, R>
pub fn then<Next, F>(self, service: F) -> Pipeline<Req, Then<Svc, Next>>
where
Self: Sized,
F: IntoService<U, Result<T::Response, T::Error>>,
U: Service<Result<T::Response, T::Error>, Error = T::Error>,
F: IntoService<Next, Result<Svc::Response, Svc::Error>>,
Next: Service<Result<Svc::Response, Svc::Error>, Error = Svc::Error>,
{
Pipeline {
service: Then::new(self.service, service.into_service()),
@ -86,10 +86,10 @@ impl<T: Service<R>, R> Pipeline<T, R> {
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it, similar to the existing `map` methods in the
/// standard library.
pub fn map<F, Res>(self, f: F) -> Pipeline<Map<T, F, R, Res>, R>
pub fn map<F, Res>(self, f: F) -> Pipeline<Req, Map<Svc, F, Req, Res>>
where
Self: Sized,
F: FnMut(T::Response) -> Res,
F: Fn(Svc::Response) -> Res,
{
Pipeline {
service: Map::new(self.service, f),
@ -105,10 +105,10 @@ impl<T: Service<R>, R> Pipeline<T, R> {
///
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it.
pub fn map_err<F, E>(self, f: F) -> Pipeline<MapErr<T, R, F, E>, R>
pub fn map_err<F, Err>(self, f: F) -> Pipeline<Req, MapErr<Svc, Req, F, Err>>
where
Self: Sized,
F: Fn(T::Error) -> E,
F: Fn(Svc::Error) -> Err,
{
Pipeline {
service: MapErr::new(self.service, f),
@ -117,9 +117,9 @@ impl<T: Service<R>, R> Pipeline<T, R> {
}
}
impl<T, R> Clone for Pipeline<T, R>
impl<Req, Svc> Clone for Pipeline<Req, Svc>
where
T: Clone,
Svc: Clone,
{
fn clone(&self) -> Self {
Pipeline {
@ -129,42 +129,31 @@ where
}
}
impl<T: Service<R>, R> Service<R> for Pipeline<T, R> {
type Response = T::Response;
type Error = T::Error;
type Future = T::Future;
impl<Req, Svc: Service<Req>> Service<Req> for Pipeline<Req, Svc> {
type Response = Svc::Response;
type Error = Svc::Error;
type Future<'f> = Svc::Future<'f> where Self: 'f, Req: 'f;
crate::forward_poll_ready!(service);
crate::forward_poll_shutdown!(service);
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), T::Error>> {
self.service.poll_ready(cx)
}
#[inline]
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
self.service.poll_shutdown(cx, is_error)
}
#[inline]
fn call(&self, req: R) -> Self::Future {
fn call(&self, req: Req) -> Self::Future<'_> {
self.service.call(req)
}
}
/// Pipeline factory
pub struct PipelineFactory<T, R, C = ()> {
pub struct PipelineFactory<Req, T, C = ()> {
factory: T,
_t: PhantomData<(R, C)>,
_t: PhantomData<(Req, C)>,
}
impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
impl<Req, T: ServiceFactory<Req, C>, C> PipelineFactory<Req, T, C> {
/// Call another service after call to this one has resolved successfully.
pub fn and_then<F, U>(
self,
factory: F,
) -> PipelineFactory<AndThenFactory<T, U, R, C>, R, C>
pub fn and_then<F, U>(self, factory: F) -> PipelineFactory<Req, AndThenFactory<T, U>, C>
where
Self: Sized,
C: Clone,
F: IntoServiceFactory<U, T::Response, C>,
U: ServiceFactory<T::Response, C, Error = T::Error, InitError = T::InitError>,
{
@ -174,15 +163,15 @@ impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
}
}
/// Apply transform to current service factory.
/// Apply middleware to current service factory.
///
/// Short version of `apply(transform, pipeline_factory(...))`
pub fn apply<U>(self, tr: U) -> PipelineFactory<ApplyTransform<U, T, R, C>, R, C>
/// Short version of `apply(middleware, pipeline_factory(...))`
pub fn apply<U>(self, tr: U) -> PipelineFactory<Req, ApplyMiddleware<U, T, C>, C>
where
U: Transform<T::Service>,
U: Middleware<T::Service>,
{
PipelineFactory {
factory: ApplyTransform::new(tr, self.factory),
factory: ApplyMiddleware::new(tr, self.factory),
_t: PhantomData,
}
}
@ -193,7 +182,7 @@ impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
///
/// Note that this function consumes the receiving pipeline and returns a
/// wrapped version of it.
pub fn then<F, U>(self, factory: F) -> PipelineFactory<ThenFactory<T, U, R>, R, C>
pub fn then<F, U>(self, factory: F) -> PipelineFactory<Req, ThenFactory<T, U>, C>
where
Self: Sized,
C: Clone,
@ -213,16 +202,13 @@ impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
/// Map this service's output to a different type, returning a new service
/// of the resulting type.
pub fn map<F, Res>(
self,
f: F,
) -> PipelineFactory<MapServiceFactory<T, F, R, Res, C>, R, C>
pub fn map<F, Res>(self, f: F) -> PipelineFactory<Req, MapFactory<T, F, Req, Res, C>, C>
where
Self: Sized,
F: FnMut(T::Response) -> Res + Clone,
F: Fn(T::Response) -> Res + Clone,
{
PipelineFactory {
factory: MapServiceFactory::new(self.factory, f),
factory: MapFactory::new(self.factory, f),
_t: PhantomData,
}
}
@ -231,13 +217,13 @@ impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
pub fn map_err<F, E>(
self,
f: F,
) -> PipelineFactory<MapErrServiceFactory<T, R, C, F, E>, R, C>
) -> PipelineFactory<Req, MapErrFactory<T, Req, C, F, E>, C>
where
Self: Sized,
F: Fn(T::Error) -> E + Clone,
{
PipelineFactory {
factory: MapErrServiceFactory::new(self.factory, f),
factory: MapErrFactory::new(self.factory, f),
_t: PhantomData,
}
}
@ -246,7 +232,7 @@ impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
pub fn map_init_err<F, E>(
self,
f: F,
) -> PipelineFactory<MapInitErr<T, R, C, F, E>, R, C>
) -> PipelineFactory<Req, MapInitErr<T, Req, C, F, E>, C>
where
Self: Sized,
F: Fn(T::InitError) -> E + Clone,
@ -258,7 +244,7 @@ impl<T: ServiceFactory<R, C>, R, C> PipelineFactory<T, R, C> {
}
}
impl<T, R, C> Clone for PipelineFactory<T, R, C>
impl<Req, T, C> Clone for PipelineFactory<Req, T, C>
where
T: Clone,
{
@ -270,15 +256,17 @@ where
}
}
impl<T: ServiceFactory<R, C>, R, C> ServiceFactory<R, C> for PipelineFactory<T, R, C> {
impl<Req, T: ServiceFactory<Req, C>, C> ServiceFactory<Req, C>
for PipelineFactory<Req, T, C>
{
type Response = T::Response;
type Error = T::Error;
type Service = T::Service;
type InitError = T::InitError;
type Future = T::Future;
type Future<'f> = T::Future<'f> where Self: 'f;
#[inline]
fn new_service(&self, cfg: C) -> Self::Future {
self.factory.new_service(cfg)
fn create(&self, cfg: C) -> Self::Future<'_> {
self.factory.create(cfg)
}
}

195
ntex-service/src/then.rs
View file

@ -1,6 +1,4 @@
use std::{
future::Future, marker::PhantomData, pin::Pin, rc::Rc, task::Context, task::Poll,
};
use std::{future::Future, pin::Pin, task::Context, task::Poll};
use super::{Service, ServiceFactory};
@ -8,50 +6,52 @@ use super::{Service, ServiceFactory};
/// another service.
///
/// This is created by the `Pipeline::then` method.
pub struct Then<A, B, R>(Rc<(A, B)>, PhantomData<R>);
pub struct Then<A, B> {
svc1: A,
svc2: B,
}
impl<A, B, R> Then<A, B, R> {
impl<A, B> Then<A, B> {
/// Create new `.then()` combinator
pub(crate) fn new(a: A, b: B) -> Then<A, B, R>
where
A: Service<R>,
B: Service<Result<A::Response, A::Error>, Error = A::Error>,
{
Self(Rc::new((a, b)), PhantomData)
pub(crate) fn new(svc1: A, svc2: B) -> Then<A, B> {
Self { svc1, svc2 }
}
}
impl<A, B, R> Clone for Then<A, B, R> {
impl<A, B> Clone for Then<A, B>
where
A: Clone,
B: Clone,
{
fn clone(&self) -> Self {
Then(self.0.clone(), PhantomData)
Then {
svc1: self.svc1.clone(),
svc2: self.svc2.clone(),
}
}
}
impl<A, B, R> Service<R> for Then<A, B, R>
impl<A, B, R> Service<R> for Then<A, B>
where
A: Service<R>,
B: Service<Result<A::Response, A::Error>, Error = A::Error>,
{
type Response = B::Response;
type Error = B::Error;
type Future = ThenServiceResponse<A, B, R>;
type Future<'f> = ThenServiceResponse<'f, A, B, R> where Self: 'f, R: 'f;
#[inline]
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let srv = self.0.as_ref();
let not_ready = !srv.0.poll_ready(cx)?.is_ready();
if !srv.1.poll_ready(cx)?.is_ready() || not_ready {
let not_ready = !self.svc1.poll_ready(cx)?.is_ready();
if !self.svc2.poll_ready(cx)?.is_ready() || not_ready {
Poll::Pending
} else {
Poll::Ready(Ok(()))
}
}
fn poll_shutdown(&self, cx: &mut Context<'_>, is_error: bool) -> Poll<()> {
let srv = self.0.as_ref();
if srv.0.poll_shutdown(cx, is_error).is_ready()
&& srv.1.poll_shutdown(cx, is_error).is_ready()
fn poll_shutdown(&self, cx: &mut Context<'_>) -> Poll<()> {
if self.svc1.poll_shutdown(cx).is_ready() && self.svc2.poll_shutdown(cx).is_ready()
{
Poll::Ready(())
} else {
@ -60,41 +60,46 @@ where
}
#[inline]
fn call(&self, req: R) -> Self::Future {
fn call(&self, req: R) -> Self::Future<'_> {
ThenServiceResponse {
slf: self,
state: State::A {
fut: self.0.as_ref().0.call(req),
b: Some(self.0.clone()),
fut: self.svc1.call(req),
},
}
}
}
pin_project_lite::pin_project! {
pub struct ThenServiceResponse<A, B, R>
pub struct ThenServiceResponse<'f, A, B, R>
where
A: Service<R>,
B: Service<Result<A::Response, A::Error>>,
{
slf: &'f Then<A, B>,
#[pin]
state: State<A, B, R>,
state: State<'f, A, B, R>,
}
}
pin_project_lite::pin_project! {
#[project = StateProject]
enum State<A, B, R>
enum State<'f, A, B, R>
where
A: Service<R>,
A: 'f,
A::Response: 'f,
B: Service<Result<A::Response, A::Error>>,
B: 'f,
R: 'f,
{
A { #[pin] fut: A::Future, b: Option<Rc<(A, B)>> },
B { #[pin] fut: B::Future },
A { #[pin] fut: A::Future<'f> },
B { #[pin] fut: B::Future<'f> },
Empty,
}
}
impl<A, B, R> Future for ThenServiceResponse<A, B, R>
impl<'a, A, B, R> Future for ThenServiceResponse<'a, A, B, R>
where
A: Service<R>,
B: Service<Result<A::Response, A::Error>>,
@ -105,11 +110,9 @@ where
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProject::A { fut, b } => match fut.poll(cx) {
StateProject::A { fut } => match fut.poll(cx) {
Poll::Ready(res) => {
let b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A
let fut = b.as_ref().1.call(res);
let fut = this.slf.svc2.call(res);
this.state.set(State::B { fut });
self.poll(cx)
}
@ -127,84 +130,81 @@ where
}
/// `.then()` service factory combinator
pub struct ThenFactory<A, B, R>(Rc<(A, B)>, PhantomData<R>);
pub struct ThenFactory<A, B> {
svc1: A,
svc2: B,
}
impl<A, B, R> ThenFactory<A, B, R> {
/// Create new `AndThen` combinator
pub(crate) fn new(a: A, b: B) -> Self {
Self(Rc::new((a, b)), PhantomData)
impl<A, B> ThenFactory<A, B> {
/// Create new factory for `Then` combinator
pub(crate) fn new(svc1: A, svc2: B) -> Self {
Self { svc1, svc2 }
}
}
impl<A, B, R, C> ServiceFactory<R, C> for ThenFactory<A, B, R>
impl<A, B, R, C> ServiceFactory<R, C> for ThenFactory<A, B>
where
A: ServiceFactory<R, C>,
C: Clone,
B: ServiceFactory<
Result<A::Response, A::Error>,
C,
Error = A::Error,
InitError = A::InitError,
>,
C: Clone,
{
type Response = B::Response;
type Error = A::Error;
type Service = Then<A::Service, B::Service, R>;
type Service = Then<A::Service, B::Service>;
type InitError = A::InitError;
type Future = ThenFactoryResponse<A, B, R, C>;
type Future<'f> = ThenFactoryResponse<'f, A, B, R, C> where Self: 'f, C: 'f;
fn new_service(&self, cfg: C) -> Self::Future {
let srv = &*self.0;
ThenFactoryResponse::new(srv.0.new_service(cfg.clone()), srv.1.new_service(cfg))
}
}
impl<A, B, R> Clone for ThenFactory<A, B, R> {
fn clone(&self) -> Self {
Self(self.0.clone(), PhantomData)
}
}
pin_project_lite::pin_project! {
pub struct ThenFactoryResponse<A, B, R, C>
where
A: ServiceFactory<R, C>,
B: ServiceFactory<Result<A::Response, A::Error>, C,
Error = A::Error,
InitError = A::InitError,
>,
{
#[pin]
fut_b: B::Future,
#[pin]
fut_a: A::Future,
a: Option<A::Service>,
b: Option<B::Service>,
}
}
impl<A, B, R, C> ThenFactoryResponse<A, B, R, C>
where
A: ServiceFactory<R, C>,
B: ServiceFactory<
Result<A::Response, A::Error>,
C,
Error = A::Error,
InitError = A::InitError,
>,
{
fn new(fut_a: A::Future, fut_b: B::Future) -> Self {
Self {
fut_a,
fut_b,
fn create(&self, cfg: C) -> Self::Future<'_> {
ThenFactoryResponse {
fut_a: self.svc1.create(cfg.clone()),
fut_b: self.svc2.create(cfg),
a: None,
b: None,
}
}
}
impl<A, B, R, C> Future for ThenFactoryResponse<A, B, R, C>
impl<A, B> Clone for ThenFactory<A, B>
where
A: Clone,
B: Clone,
{
fn clone(&self) -> Self {
Self {
svc1: self.svc1.clone(),
svc2: self.svc2.clone(),
}
}
}
pin_project_lite::pin_project! {
pub struct ThenFactoryResponse<'f, A, B, R, C>
where
A: ServiceFactory<R, C>,
B: ServiceFactory<Result<A::Response, A::Error>, C,
Error = A::Error,
InitError = A::InitError,
>,
A: 'f,
B: 'f,
C: 'f,
{
#[pin]
fut_b: B::Future<'f>,
#[pin]
fut_a: A::Future<'f>,
a: Option<A::Service>,
b: Option<B::Service>,
}
}
impl<'f, A, B, R, C> Future for ThenFactoryResponse<'f, A, B, R, C>
where
A: ServiceFactory<R, C>,
B: ServiceFactory<
@ -214,7 +214,7 @@ where
InitError = A::InitError,
>,
{
type Output = Result<Then<A::Service, B::Service, R>, A::InitError>;
type Output = Result<Then<A::Service, B::Service>, A::InitError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
@ -253,14 +253,14 @@ mod tests {
impl Service<Result<&'static str, &'static str>> for Srv1 {
type Response = &'static str;
type Error = ();
type Future = Ready<Self::Response, Self::Error>;
type Future<'f> = Ready<Self::Response, Self::Error>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.set(self.0.get() + 1);
Poll::Ready(Ok(()))
}
fn call(&self, req: Result<&'static str, &'static str>) -> Self::Future {
fn call(&self, req: Result<&'static str, &'static str>) -> Self::Future<'_> {
match req {
Ok(msg) => Ready::Ok(msg),
Err(_) => Ready::Err(()),
@ -268,19 +268,20 @@ mod tests {
}
}
#[derive(Clone)]
struct Srv2(Rc<Cell<usize>>);
impl Service<Result<&'static str, ()>> for Srv2 {
type Response = (&'static str, &'static str);
type Error = ();
type Future = Ready<Self::Response, ()>;
type Future<'f> = Ready<Self::Response, ()>;
fn poll_ready(&self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.set(self.0.get() + 1);
Poll::Ready(Err(()))
}
fn call(&self, req: Result<&'static str, ()>) -> Self::Future {
fn call(&self, req: Result<&'static str, ()>) -> Self::Future<'_> {
match req {
Ok(msg) => Ready::Ok((msg, "ok")),
Err(()) => Ready::Ok(("srv2", "err")),
@ -295,7 +296,7 @@ mod tests {
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert_eq!(res, Poll::Ready(Err(())));
assert_eq!(cnt.get(), 2);
let res = lazy(|cx| srv.poll_shutdown(cx, false)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
}
@ -321,7 +322,7 @@ mod tests {
let factory = pipeline_factory(blank)
.then(move || Ready::Ok(Srv2(cnt.clone())))
.clone();
let srv = factory.new_service(&()).await.unwrap();
let srv = factory.create(&()).await.unwrap();
let res = srv.call(Ok("srv1")).await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), ("srv1", "ok"));

160
ntex-service/src/transform.rs
View file

@ -1,27 +1,25 @@
use std::{
future::Future, marker::PhantomData, pin::Pin, rc::Rc, task::Context, task::Poll,
};
use std::{future::Future, marker, pin::Pin, rc::Rc, task::Context, task::Poll};
use crate::{IntoServiceFactory, Service, ServiceFactory};
/// Apply transform to a service.
pub fn apply<T, S, R, C, U>(t: T, factory: U) -> ApplyTransform<T, S, R, C>
/// Apply middleware to a service.
pub fn apply<T, S, R, C, U>(t: T, factory: U) -> ApplyMiddleware<T, S, C>
where
S: ServiceFactory<R, C>,
T: Transform<S::Service>,
T: Middleware<S::Service>,
U: IntoServiceFactory<S, R, C>,
{
ApplyTransform::new(t, factory.into_factory())
ApplyMiddleware::new(t, factory.into_factory())
}
/// The `Transform` trait defines the interface of a service factory that wraps inner service
/// The `Middleware` trait defines the interface of a service factory that wraps inner service
/// during construction.
///
/// Transform(middleware) wraps inner service and runs during
/// Middleware wraps inner service and runs during
/// inbound and/or outbound processing in the request/response lifecycle.
/// It may modify request and/or response.
///
/// For example, timeout transform:
/// For example, timeout middleware:
///
/// ```rust,ignore
/// pub struct Timeout<S> {
@ -36,13 +34,13 @@ where
/// type Request = S::Request;
/// type Response = S::Response;
/// type Error = TimeoutError<S::Error>;
/// type Future = TimeoutServiceResponse<S>;
/// type Future = TimeoutResponse<S>;
///
/// fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
/// fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
/// ready!(self.service.poll_ready(cx)).map_err(TimeoutError::Service)
/// }
///
/// fn call(&mut self, req: S::Request) -> Self::Future {
/// fn call(&self, req: S::Request) -> Self::Future {
/// TimeoutServiceResponse {
/// fut: self.service.call(req),
/// sleep: Delay::new(clock::now() + self.timeout),
@ -54,76 +52,71 @@ where
/// Timeout service in above example is decoupled from underlying service implementation
/// and could be applied to any service.
///
/// The `Transform` trait defines the interface of a Service factory. `Transform`
/// is often implemented for middleware, defining how to construct a
/// middleware Service. A Service that is constructed by the factory takes
/// The `Middleware` trait defines the interface of a middleware factory, defining how to
/// construct a middleware Service. A Service that is constructed by the factory takes
/// the Service that follows it during execution as a parameter, assuming
/// ownership of the next Service.
///
/// Factory for `Timeout` middleware from the above example could look like this:
///
/// ```rust,ignore
/// pub struct TimeoutTransform {
/// pub struct TimeoutMiddleware {
/// timeout: Duration,
/// }
///
/// impl<S> Transform<S> for TimeoutTransform<E>
/// impl<S> Middleware<S> for TimeoutMiddleware<E>
/// where
/// S: Service,
/// {
/// type Transform = Timeout<S>;
/// type Service = Timeout<S>;
///
/// fn new_transform(&self, service: S) -> Self::Transform {
/// ok(TimeoutService {
/// fn create(&self, service: S) -> Self::Service {
/// ok(Timeout {
/// service,
/// timeout: self.timeout,
/// })
/// }
/// }
/// ```
pub trait Transform<S> {
/// The `TransformService` value created by this factory
pub trait Middleware<S> {
/// The middleware `Service` value created by this factory
type Service;
/// Creates and returns a new Transform component, asynchronously
fn new_transform(&self, service: S) -> Self::Service;
/// Creates and returns a new middleware Service
fn create(&self, service: S) -> Self::Service;
}
impl<T, S> Transform<S> for Rc<T>
impl<T, S> Middleware<S> for Rc<T>
where
T: Transform<S>,
T: Middleware<S>,
{
type Service = T::Service;
fn new_transform(&self, service: S) -> T::Service {
self.as_ref().new_transform(service)
fn create(&self, service: S) -> T::Service {
self.as_ref().create(service)
}
}
/// `Apply` transform to new service
pub struct ApplyTransform<T, S, R, C>(Rc<(T, S)>, PhantomData<(R, C)>);
/// `Apply` middleware to a service factory.
pub struct ApplyMiddleware<T, S, C>(Rc<(T, S)>, marker::PhantomData<C>);
impl<T, S, R, C> ApplyTransform<T, S, R, C>
where
S: ServiceFactory<R, C>,
T: Transform<S::Service>,
{
/// Create new `ApplyTransform` new service instance
pub(crate) fn new(t: T, service: S) -> Self {
Self(Rc::new((t, service)), PhantomData)
impl<T, S, C> ApplyMiddleware<T, S, C> {
/// Create new `ApplyMiddleware` service factory instance
pub(crate) fn new(mw: T, svc: S) -> Self {
Self(Rc::new((mw, svc)), marker::PhantomData)
}
}
impl<T, S, R, C> Clone for ApplyTransform<T, S, R, C> {
impl<T, S, C> Clone for ApplyMiddleware<T, S, C> {
fn clone(&self) -> Self {
ApplyTransform(self.0.clone(), PhantomData)
Self(self.0.clone(), marker::PhantomData)
}
}
impl<T, S, R, C> ServiceFactory<R, C> for ApplyTransform<T, S, R, C>
impl<T, S, R, C> ServiceFactory<R, C> for ApplyMiddleware<T, S, C>
where
S: ServiceFactory<R, C>,
T: Transform<S::Service>,
T: Middleware<S::Service>,
T::Service: Service<R>,
{
type Response = <T::Service as Service<R>>::Response;
@ -131,34 +124,35 @@ where
type Service = T::Service;
type InitError = S::InitError;
type Future = ApplyTransformFuture<T, S, R, C>;
type Future<'f> = ApplyMiddlewareFuture<'f, T, S, R, C> where Self: 'f, C: 'f;
fn new_service(&self, cfg: C) -> Self::Future {
ApplyTransformFuture {
store: self.0.clone(),
fut: self.0.as_ref().1.new_service(cfg),
_t: PhantomData,
#[inline]
fn create(&self, cfg: C) -> Self::Future<'_> {
ApplyMiddlewareFuture {
slf: self.0.clone(),
fut: self.0 .1.create(cfg),
}
}
}
pin_project_lite::pin_project! {
pub struct ApplyTransformFuture<T, S, R, C>
pub struct ApplyMiddlewareFuture<'f, T, S, R, C>
where
S: ServiceFactory<R, C>,
T: Transform<S::Service>,
S: 'f,
T: Middleware<S::Service>,
C: 'f,
{
store: Rc<(T, S)>,
slf: Rc<(T, S)>,
#[pin]
fut: S::Future,
_t: PhantomData<C>
fut: S::Future<'f>,
}
}
impl<T, S, R, C> Future for ApplyTransformFuture<T, S, R, C>
impl<'f, T, S, R, C> Future for ApplyMiddlewareFuture<'f, T, S, R, C>
where
S: ServiceFactory<R, C>,
T: Transform<S::Service>,
T: Middleware<S::Service>,
{
type Output = Result<T::Service, S::InitError>;
@ -166,59 +160,85 @@ where
let this = self.as_mut().project();
match this.fut.poll(cx)? {
Poll::Ready(srv) => Poll::Ready(Ok(this.store.0.new_transform(srv))),
Poll::Ready(srv) => Poll::Ready(Ok(this.slf.0.create(srv))),
Poll::Pending => Poll::Pending,
}
}
}
/// Identity is a transform.
/// Identity is a middleware.
///
/// It returns service without modifications.
#[derive(Debug, Clone, Copy)]
pub struct Identity;
impl<S> Transform<S> for Identity {
impl<S> Middleware<S> for Identity {
type Service = S;
#[inline]
fn new_transform(&self, service: S) -> Self::Service {
fn create(&self, service: S) -> Self::Service {
service
}
}
/// Stack of middlewares.
#[derive(Debug, Clone)]
pub struct Stack<Inner, Outer> {
inner: Inner,
outer: Outer,
}
impl<Inner, Outer> Stack<Inner, Outer> {
pub fn new(inner: Inner, outer: Outer) -> Self {
Stack { inner, outer }
}
}
impl<S, Inner, Outer> Middleware<S> for Stack<Inner, Outer>
where
Inner: Middleware<S>,
Outer: Middleware<Inner::Service>,
{
type Service = Outer::Service;
fn create(&self, service: S) -> Self::Service {
self.outer.create(self.inner.create(service))
}
}
#[cfg(test)]
#[allow(clippy::redundant_clone)]
mod tests {
use ntex_util::future::{lazy, Ready};
use std::marker;
use super::*;
use crate::{fn_service, Service, ServiceFactory};
#[derive(Clone)]
struct Tr<R>(PhantomData<R>);
struct Tr<R>(marker::PhantomData<R>);
impl<S, R> Transform<S> for Tr<R> {
impl<S, R> Middleware<S> for Tr<R> {
type Service = Srv<S, R>;
fn new_transform(&self, service: S) -> Self::Service {
Srv(service, PhantomData)
fn create(&self, service: S) -> Self::Service {
Srv(service, marker::PhantomData)
}
}
#[derive(Clone)]
struct Srv<S, R>(S, PhantomData<R>);
struct Srv<S, R>(S, marker::PhantomData<R>);
impl<S: Service<R>, R> Service<R> for Srv<S, R> {
type Response = S::Response;
type Error = S::Error;
type Future = S::Future;
type Future<'f> = S::Future<'f> where Self: 'f, R: 'f;
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.poll_ready(cx)
}
fn call(&self, req: R) -> Self::Future {
fn call(&self, req: R) -> Self::Future<'_> {
self.0.call(req)
}
}
@ -226,12 +246,12 @@ mod tests {
#[ntex::test]
async fn transform() {
let factory = apply(
Rc::new(Tr(PhantomData).clone()),
Rc::new(Tr(marker::PhantomData).clone()),
fn_service(|i: usize| Ready::<_, ()>::Ok(i * 2)),
)
.clone();
let srv = factory.new_service(()).await.unwrap();
let srv = factory.create(&()).await.unwrap();
let res = srv.call(10).await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), 20);
@ -239,7 +259,7 @@ mod tests {
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert_eq!(res, Poll::Ready(Ok(())));
let res = lazy(|cx| srv.poll_shutdown(cx, true)).await;
let res = lazy(|cx| srv.poll_shutdown(cx)).await;
assert_eq!(res, Poll::Ready(()));
}
}