Remove combinators (#214)
This commit is contained in:
Carl Lerche
GitHub
parent
3a407c2775
commit
f6dcbb6ca1
|
|
@ -0,0 +1,52 @@
|
|||
//! Combinators for working with `Service`s
|
||||
|
||||
pub use tower_util::BoxService;
|
||||
pub use tower_util::CallAll;
|
||||
pub use tower_util::CallAllUnordered;
|
||||
pub use tower_util::EitherService;
|
||||
pub use tower_util::Oneshot;
|
||||
pub use tower_util::OptionService;
|
||||
pub use tower_util::Ready;
|
||||
pub use tower_util::ServiceFn;
|
||||
pub use tower_util::UnsyncBoxService;
|
||||
|
||||
use futures::Stream;
|
||||
use tower_service::Service;
|
||||
|
||||
impl<T: ?Sized, Request> ServiceExt<Request> for T where T: Service<Request> {}
|
||||
|
||||
type Error = Box<::std::error::Error + Send + Sync>;
|
||||
|
||||
/// An extension trait for `Service`s that provides a variety of convenient
|
||||
/// adapters
|
||||
pub trait ServiceExt<Request>: Service<Request> {
|
||||
/// A future yielding the service when it is ready to accept a request.
|
||||
fn ready(self) -> Ready<Self, Request>
|
||||
where
|
||||
Self: Sized,
|
||||
{
|
||||
Ready::new(self)
|
||||
}
|
||||
|
||||
/// Consume this `Service`, calling with the providing request once it is ready.
|
||||
fn oneshot(self, req: Request) -> Oneshot<Self, Request>
|
||||
where
|
||||
Self: Sized,
|
||||
{
|
||||
Oneshot::new(self, req)
|
||||
}
|
||||
|
||||
/// Process all requests from the given `Stream`, and produce a `Stream` of their responses.
|
||||
///
|
||||
/// This is essentially `Stream<Item = Request>` + `Self` => `Stream<Item = Response>`. See the
|
||||
/// documentation for [`CallAll`](struct.CallAll.html) for details.
|
||||
fn call_all<S>(self, reqs: S) -> CallAll<Self, S>
|
||||
where
|
||||
Self: Sized,
|
||||
Self::Error: Into<Error>,
|
||||
S: Stream<Item = Request>,
|
||||
S::Error: Into<Error>,
|
||||
{
|
||||
CallAll::new(self, reqs)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,139 +0,0 @@
|
|||
use futures::{Future, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
/// Service for the `and_then` combinator, chaining a computation onto the end of
|
||||
/// another service which completes successfully.
|
||||
///
|
||||
/// This is created by the `ServiceExt::and_then` method.
|
||||
#[derive(Clone)]
|
||||
pub struct AndThen<A, B> {
|
||||
a: A,
|
||||
b: B,
|
||||
}
|
||||
|
||||
impl<A, B> AndThen<A, B> {
|
||||
/// Create new `AndThen` combinator
|
||||
pub fn new<Request>(a: A, b: B) -> AndThen<A, B>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<A::Response, Error = A::Error> + Clone,
|
||||
{
|
||||
AndThen { a, b }
|
||||
}
|
||||
}
|
||||
|
||||
impl<A, B, Request> Service<Request> for AndThen<A, B>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<A::Response, Error = A::Error> + Clone,
|
||||
{
|
||||
type Response = B::Response;
|
||||
type Error = B::Error;
|
||||
type Future = AndThenFuture<A, B, Request>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
let _ = try_ready!(self.a.poll_ready());
|
||||
self.b.poll_ready()
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
AndThenFuture {
|
||||
fut_a: self.a.call(req),
|
||||
b: self.b.clone(),
|
||||
fut_b: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct AndThenFuture<A, B, Request>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<A::Response, Error = A::Error>,
|
||||
{
|
||||
b: B,
|
||||
fut_b: Option<B::Future>,
|
||||
fut_a: A::Future,
|
||||
}
|
||||
|
||||
impl<A, B, Request> Future for AndThenFuture<A, B, Request>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<A::Response, Error = A::Error>,
|
||||
{
|
||||
type Item = B::Response;
|
||||
type Error = B::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
if let Some(ref mut fut) = self.fut_b {
|
||||
return fut.poll();
|
||||
}
|
||||
|
||||
let resp = try_ready!(self.fut_a.poll());
|
||||
self.fut_b = Some(self.b.call(resp));
|
||||
self.poll()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use futures::future::{ok, FutureResult};
|
||||
use futures::{Async, Poll};
|
||||
use std::cell::Cell;
|
||||
use std::rc::Rc;
|
||||
|
||||
use super::*;
|
||||
use ServiceExt;
|
||||
|
||||
struct Srv1(Rc<Cell<usize>>);
|
||||
impl Service<&'static str> for Srv1 {
|
||||
type Response = &'static str;
|
||||
type Error = ();
|
||||
type Future = FutureResult<Self::Response, ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.0.set(self.0.get() + 1);
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, req: &'static str) -> Self::Future {
|
||||
ok(req)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Srv2(Rc<Cell<usize>>);
|
||||
|
||||
impl Service<&'static str> for Srv2 {
|
||||
type Response = (&'static str, &'static str);
|
||||
type Error = ();
|
||||
type Future = FutureResult<Self::Response, ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.0.set(self.0.get() + 1);
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, req: &'static str) -> Self::Future {
|
||||
ok((req, "srv2"))
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_poll_ready() {
|
||||
let cnt = Rc::new(Cell::new(0));
|
||||
let mut srv = Srv1(cnt.clone()).and_then(Srv2(cnt.clone()));
|
||||
let res = srv.poll_ready();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(()));
|
||||
assert_eq!(cnt.get(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_call() {
|
||||
let cnt = Rc::new(Cell::new(0));
|
||||
let mut srv = Srv1(cnt.clone()).and_then(Srv2(cnt));
|
||||
let res = srv.call("srv1").poll();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(("srv1", "srv2")));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,98 +0,0 @@
|
|||
use std::marker::PhantomData;
|
||||
|
||||
use futures::{Future, IntoFuture, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
/// `Apply` service combinator
|
||||
pub struct Apply<T, F, In, Out, Request>
|
||||
where
|
||||
T: Service<Request>,
|
||||
{
|
||||
service: T,
|
||||
f: F,
|
||||
_r: PhantomData<Fn((In, Request)) -> Out>,
|
||||
}
|
||||
|
||||
impl<T, F, In, Out, Request> Apply<T, F, In, Out, Request>
|
||||
where
|
||||
T: Service<Request>,
|
||||
F: Fn(In, T) -> Out,
|
||||
Out: IntoFuture,
|
||||
{
|
||||
/// Create new `Apply` combinator
|
||||
pub(crate) fn new(f: F, service: T) -> Self {
|
||||
Self {
|
||||
service,
|
||||
f,
|
||||
_r: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, In, Out, Request> Clone for Apply<T, F, In, Out, Request>
|
||||
where
|
||||
T: Service<Request> + Clone,
|
||||
F: Clone,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
Apply {
|
||||
service: self.service.clone(),
|
||||
f: self.f.clone(),
|
||||
_r: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, In, Out, Request> Service<In> for Apply<T, F, In, Out, Request>
|
||||
where
|
||||
T: Service<Request, Error = Out::Error> + Clone,
|
||||
F: Fn(In, T) -> Out,
|
||||
Out: IntoFuture,
|
||||
{
|
||||
type Response = <Out::Future as Future>::Item;
|
||||
type Error = <Out::Future as Future>::Error;
|
||||
type Future = Out::Future;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.service.poll_ready()
|
||||
}
|
||||
|
||||
fn call(&mut self, req: In) -> Self::Future {
|
||||
let service = self.service.clone();
|
||||
(self.f)(req, service).into_future()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use futures::future::{ok, FutureResult};
|
||||
use futures::{Async, Future, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
use crate::ServiceExt;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Srv;
|
||||
impl Service<()> for Srv {
|
||||
type Response = ();
|
||||
type Error = ();
|
||||
type Future = FutureResult<(), ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, _: ()) -> Self::Future {
|
||||
ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_call() {
|
||||
let mut srv =
|
||||
Srv.apply(|req: &'static str, mut srv| srv.call(()).map(move |res| (req, res)));
|
||||
let res = srv.call("srv").poll();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(("srv", ())));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,125 +0,0 @@
|
|||
use std::marker::PhantomData;
|
||||
|
||||
use futures::{Future, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
/// Service for the `from_err` combinator, changing the error type of a service.
|
||||
///
|
||||
/// This is created by the `ServiceExt::from_err` method.
|
||||
pub struct FromErr<A, E> {
|
||||
service: A,
|
||||
_e: PhantomData<E>,
|
||||
}
|
||||
|
||||
impl<A, E> FromErr<A, E> {
|
||||
pub(crate) fn new<Request>(service: A) -> Self
|
||||
where
|
||||
A: Service<Request>,
|
||||
E: From<A::Error>,
|
||||
{
|
||||
FromErr {
|
||||
service,
|
||||
_e: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<A, E> Clone for FromErr<A, E>
|
||||
where
|
||||
A: Clone,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
FromErr {
|
||||
service: self.service.clone(),
|
||||
_e: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<A, E, Request> Service<Request> for FromErr<A, E>
|
||||
where
|
||||
A: Service<Request>,
|
||||
E: From<A::Error>,
|
||||
{
|
||||
type Response = A::Response;
|
||||
type Error = E;
|
||||
type Future = FromErrFuture<A::Future, E>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), E> {
|
||||
Ok(self.service.poll_ready().map_err(E::from)?)
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
FromErrFuture {
|
||||
fut: self.service.call(req),
|
||||
f: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct FromErrFuture<A, E> {
|
||||
fut: A,
|
||||
f: PhantomData<E>,
|
||||
}
|
||||
|
||||
impl<A, E> Future for FromErrFuture<A, E>
|
||||
where
|
||||
A: Future,
|
||||
E: From<A::Error>,
|
||||
{
|
||||
type Item = A::Item;
|
||||
type Error = E;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
self.fut.poll().map_err(E::from)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use futures::future::{err, FutureResult};
|
||||
|
||||
use super::*;
|
||||
use ServiceExt;
|
||||
|
||||
struct Srv;
|
||||
|
||||
impl Service<()> for Srv {
|
||||
type Response = ();
|
||||
type Error = ();
|
||||
type Future = FutureResult<(), ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Err(())
|
||||
}
|
||||
|
||||
fn call(&mut self, _: ()) -> Self::Future {
|
||||
err(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
struct Error;
|
||||
|
||||
impl From<()> for Error {
|
||||
fn from(_: ()) -> Self {
|
||||
Error
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_poll_ready() {
|
||||
let mut srv = Srv.from_err::<Error>();
|
||||
let res = srv.poll_ready();
|
||||
assert!(res.is_err());
|
||||
assert_eq!(res.err().unwrap(), Error);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_call() {
|
||||
let mut srv = Srv.from_err::<Error>();
|
||||
let res = srv.call(()).poll();
|
||||
assert!(res.is_err());
|
||||
assert_eq!(res.err().unwrap(), Error);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,133 +0,0 @@
|
|||
use futures::{Async, Future, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
use std::marker::PhantomData;
|
||||
|
||||
/// Service for the `map` combinator, changing the type of a service's response.
|
||||
///
|
||||
/// This is created by the `ServiceExt::map` method.
|
||||
pub struct Map<T, F, R> {
|
||||
service: T,
|
||||
f: F,
|
||||
_p: PhantomData<fn() -> R>,
|
||||
}
|
||||
|
||||
impl<T, F, R> Map<T, F, R> {
|
||||
/// Create new `Map` combinator
|
||||
pub fn new<Request>(service: T, f: F) -> Self
|
||||
where
|
||||
T: Service<Request>,
|
||||
F: Fn(T::Response) -> R + Clone,
|
||||
{
|
||||
Map {
|
||||
service,
|
||||
f,
|
||||
_p: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, R> Clone for Map<T, F, R>
|
||||
where
|
||||
T: Clone,
|
||||
F: Clone,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
Map {
|
||||
service: self.service.clone(),
|
||||
f: self.f.clone(),
|
||||
_p: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, R, Request> Service<Request> for Map<T, F, R>
|
||||
where
|
||||
T: Service<Request>,
|
||||
F: Fn(T::Response) -> R + Clone,
|
||||
{
|
||||
type Response = R;
|
||||
type Error = T::Error;
|
||||
type Future = MapFuture<T::Future, F, R>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.service.poll_ready()
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
MapFuture::new(self.service.call(req), self.f.clone())
|
||||
}
|
||||
}
|
||||
|
||||
pub struct MapFuture<T, F, R>
|
||||
where
|
||||
T: Future,
|
||||
F: Fn(T::Item) -> R,
|
||||
{
|
||||
f: F,
|
||||
fut: T,
|
||||
}
|
||||
|
||||
impl<T, F, R> MapFuture<T, F, R>
|
||||
where
|
||||
T: Future,
|
||||
F: Fn(T::Item) -> R,
|
||||
{
|
||||
fn new(fut: T, f: F) -> Self {
|
||||
MapFuture { f, fut }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, R> Future for MapFuture<T, F, R>
|
||||
where
|
||||
T: Future,
|
||||
F: Fn(T::Item) -> R,
|
||||
{
|
||||
type Item = R;
|
||||
type Error = T::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
let resp = try_ready!(self.fut.poll());
|
||||
Ok(Async::Ready((self.f)(resp)))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use futures::future::{ok, FutureResult};
|
||||
|
||||
use super::*;
|
||||
use ServiceExt;
|
||||
|
||||
struct Srv;
|
||||
|
||||
impl Service<()> for Srv {
|
||||
type Response = ();
|
||||
type Error = ();
|
||||
type Future = FutureResult<(), ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, _: ()) -> Self::Future {
|
||||
ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_poll_ready() {
|
||||
let mut srv = Srv.map(|_| "ok");
|
||||
let res = srv.poll_ready();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_call() {
|
||||
let mut srv = Srv.map(|_| "ok");
|
||||
let res = srv.call(()).poll();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready("ok"));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,132 +0,0 @@
|
|||
use futures::{Future, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
use std::marker::PhantomData;
|
||||
|
||||
/// Service for the `map_err` combinator, changing the type of a service's error.
|
||||
///
|
||||
/// This is created by the `ServiceExt::map_err` method.
|
||||
pub struct MapErr<T, F, E> {
|
||||
service: T,
|
||||
f: F,
|
||||
_p: PhantomData<E>,
|
||||
}
|
||||
|
||||
impl<T, F, E> MapErr<T, F, E> {
|
||||
/// Create new `MapErr` combinator
|
||||
pub fn new<Request>(service: T, f: F) -> Self
|
||||
where
|
||||
T: Service<Request>,
|
||||
F: Fn(T::Error) -> E + Clone,
|
||||
{
|
||||
MapErr {
|
||||
service,
|
||||
f,
|
||||
_p: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, E> Clone for MapErr<T, F, E>
|
||||
where
|
||||
T: Clone,
|
||||
F: Clone,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
MapErr {
|
||||
service: self.service.clone(),
|
||||
f: self.f.clone(),
|
||||
_p: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, E, Request> Service<Request> for MapErr<T, F, E>
|
||||
where
|
||||
T: Service<Request>,
|
||||
F: Fn(T::Error) -> E + Clone,
|
||||
{
|
||||
type Response = T::Response;
|
||||
type Error = E;
|
||||
type Future = MapErrFuture<T::Future, F, E>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.service.poll_ready().map_err(&self.f)
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
MapErrFuture::new(self.service.call(req), self.f.clone())
|
||||
}
|
||||
}
|
||||
|
||||
pub struct MapErrFuture<T, F, E>
|
||||
where
|
||||
T: Future,
|
||||
F: Fn(T::Error) -> E,
|
||||
{
|
||||
f: F,
|
||||
fut: T,
|
||||
}
|
||||
|
||||
impl<T, F, E> MapErrFuture<T, F, E>
|
||||
where
|
||||
T: Future,
|
||||
F: Fn(T::Error) -> E,
|
||||
{
|
||||
fn new(fut: T, f: F) -> Self {
|
||||
MapErrFuture { f, fut }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, F, E> Future for MapErrFuture<T, F, E>
|
||||
where
|
||||
T: Future,
|
||||
F: Fn(T::Error) -> E,
|
||||
{
|
||||
type Item = T::Item;
|
||||
type Error = E;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
self.fut.poll().map_err(&self.f)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use futures::future::{err, FutureResult};
|
||||
|
||||
use super::*;
|
||||
use ServiceExt;
|
||||
|
||||
struct Srv;
|
||||
|
||||
impl Service<()> for Srv {
|
||||
type Response = ();
|
||||
type Error = ();
|
||||
type Future = FutureResult<(), ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Err(())
|
||||
}
|
||||
|
||||
fn call(&mut self, _: ()) -> Self::Future {
|
||||
err(())
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_poll_ready() {
|
||||
let mut srv = Srv.map_err(|_| "error");
|
||||
let res = srv.poll_ready();
|
||||
assert!(res.is_err());
|
||||
assert_eq!(res.err().unwrap(), "error");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_call() {
|
||||
let mut srv = Srv.map_err(|_| "error");
|
||||
let res = srv.call(()).poll();
|
||||
assert!(res.is_err());
|
||||
assert_eq!(res.err().unwrap(), "error");
|
||||
}
|
||||
}
|
||||
|
|
@ -1,152 +0,0 @@
|
|||
//! Combinators for working with `Service`s
|
||||
|
||||
mod and_then;
|
||||
mod apply;
|
||||
mod from_err;
|
||||
mod map;
|
||||
mod map_err;
|
||||
mod then;
|
||||
|
||||
pub use self::and_then::AndThen;
|
||||
pub use self::apply::Apply;
|
||||
pub use self::from_err::FromErr;
|
||||
pub use self::map::Map;
|
||||
pub use self::map_err::MapErr;
|
||||
pub use self::then::Then;
|
||||
|
||||
// `tower-util` re-exports
|
||||
pub use tower_util::BoxService;
|
||||
pub use tower_util::CallAll;
|
||||
pub use tower_util::CallAllUnordered;
|
||||
pub use tower_util::EitherService;
|
||||
pub use tower_util::Oneshot;
|
||||
pub use tower_util::OptionService;
|
||||
pub use tower_util::Ready;
|
||||
pub use tower_util::ServiceFn;
|
||||
pub use tower_util::UnsyncBoxService;
|
||||
|
||||
use futures::{IntoFuture, Stream};
|
||||
use tower_service::Service;
|
||||
|
||||
impl<T: ?Sized, Request> ServiceExt<Request> for T where T: Service<Request> {}
|
||||
|
||||
type Error = Box<::std::error::Error + Send + Sync>;
|
||||
|
||||
/// An extension trait for `Service`s that provides a variety of convenient
|
||||
/// adapters
|
||||
pub trait ServiceExt<Request>: Service<Request> {
|
||||
/// A future yielding the service when it is ready to accept a request.
|
||||
fn ready(self) -> Ready<Self, Request>
|
||||
where
|
||||
Self: Sized,
|
||||
{
|
||||
Ready::new(self)
|
||||
}
|
||||
|
||||
fn apply<F, In, Out>(self, f: F) -> Apply<Self, F, In, Out, Request>
|
||||
where
|
||||
Self: Service<Request> + Clone + Sized,
|
||||
F: Fn(In, Self) -> Out,
|
||||
Out: IntoFuture<Error = Self::Error>,
|
||||
{
|
||||
Apply::new(f, self)
|
||||
}
|
||||
|
||||
/// Call another service after call to this one has resolved successfully.
|
||||
///
|
||||
/// This function can be used to chain two services together and ensure that
|
||||
/// the second service isn't called until call to the fist service have finished.
|
||||
/// Result of the call to the first service is used as an input parameter
|
||||
/// for the second service's call.
|
||||
///
|
||||
/// Note that this function consumes the receiving service and returns a
|
||||
/// wrapped version of it.
|
||||
fn and_then<B>(self, service: B) -> AndThen<Self, B>
|
||||
where
|
||||
Self: Sized,
|
||||
B: Service<Self::Response, Error = Self::Error> + Clone,
|
||||
{
|
||||
AndThen::new(self, service)
|
||||
}
|
||||
|
||||
/// Map this service's error to any error implementing `From` for
|
||||
/// this service`s `Error`.
|
||||
///
|
||||
/// Note that this function consumes the receiving service and returns a
|
||||
/// wrapped version of it.
|
||||
fn from_err<E>(self) -> FromErr<Self, E>
|
||||
where
|
||||
Self: Sized,
|
||||
E: From<Self::Error>,
|
||||
{
|
||||
FromErr::new(self)
|
||||
}
|
||||
|
||||
/// Chain on a computation for when a call to the service finished,
|
||||
/// passing the result of the call to the next service `B`.
|
||||
///
|
||||
/// Note that this function consumes the receiving future and returns a
|
||||
/// wrapped version of it.
|
||||
fn then<B>(self, service: B) -> Then<Self, B>
|
||||
where
|
||||
Self: Sized,
|
||||
B: Service<Result<Self::Response, Self::Error>, Error = Self::Error> + Clone,
|
||||
{
|
||||
Then::new(self, service)
|
||||
}
|
||||
|
||||
/// Map this service's output to a different type, returning a new service of
|
||||
/// the resulting type.
|
||||
///
|
||||
/// This function is similar to the `Option::map` or `Iterator::map` where
|
||||
/// it will change the type of the underlying service.
|
||||
///
|
||||
/// 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.
|
||||
fn map<F, R>(self, f: F) -> Map<Self, F, R>
|
||||
where
|
||||
Self: Sized,
|
||||
F: Fn(Self::Response) -> R + Clone,
|
||||
{
|
||||
Map::new(self, f)
|
||||
}
|
||||
|
||||
/// Map this service's error to a different error, returning a new service.
|
||||
///
|
||||
/// This function is similar to the `Result::map_err` where it will change
|
||||
/// the error type of the underlying service. This is useful for example to
|
||||
/// ensure that services have the same error type.
|
||||
///
|
||||
/// Note that this function consumes the receiving service and returns a
|
||||
/// wrapped version of it.
|
||||
fn map_err<F, E>(self, f: F) -> MapErr<Self, F, E>
|
||||
where
|
||||
Self: Sized,
|
||||
F: Fn(Self::Error) -> E + Clone,
|
||||
{
|
||||
MapErr::new(self, f)
|
||||
}
|
||||
|
||||
/// Consume this `Service`, calling with the providing request once it is ready.
|
||||
fn oneshot(self, req: Request) -> Oneshot<Self, Request>
|
||||
where
|
||||
Self: Sized,
|
||||
{
|
||||
Oneshot::new(self, req)
|
||||
}
|
||||
|
||||
/// Process all requests from the given `Stream`, and produce a `Stream` of their responses.
|
||||
///
|
||||
/// This is essentially `Stream<Item = Request>` + `Self` => `Stream<Item = Response>`. See the
|
||||
/// documentation for [`CallAll`](struct.CallAll.html) for details.
|
||||
fn call_all<S>(self, reqs: S) -> CallAll<Self, S>
|
||||
where
|
||||
Self: Sized,
|
||||
Self::Error: Into<Error>,
|
||||
S: Stream<Item = Request>,
|
||||
S::Error: Into<Error>,
|
||||
{
|
||||
CallAll::new(self, reqs)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,170 +0,0 @@
|
|||
use futures::{Async, Future, Poll};
|
||||
use tower_service::Service;
|
||||
|
||||
/// Service for the `then` combinator, chaining a computation onto the end of
|
||||
/// another service.
|
||||
///
|
||||
/// This is created by the `ServiceExt::then` method.
|
||||
pub struct Then<A, B> {
|
||||
a: A,
|
||||
b: B,
|
||||
}
|
||||
|
||||
impl<A, B> Then<A, B> {
|
||||
/// Create new `Then` combinator
|
||||
pub fn new<Request>(a: A, b: B) -> Then<A, B>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<Result<A::Response, A::Error>, Error = A::Error> + Clone,
|
||||
{
|
||||
Then { a, b }
|
||||
}
|
||||
}
|
||||
|
||||
impl<A, B> Clone for Then<A, B>
|
||||
where
|
||||
A: Clone,
|
||||
B: Clone,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
Then {
|
||||
a: self.a.clone(),
|
||||
b: self.b.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<A, B, Request> Service<Request> for Then<A, B>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<Result<A::Response, A::Error>, Error = A::Error> + Clone,
|
||||
{
|
||||
type Response = B::Response;
|
||||
type Error = B::Error;
|
||||
type Future = ThenFuture<A, B, Request>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
let _ = try_ready!(self.a.poll_ready());
|
||||
self.b.poll_ready()
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
ThenFuture {
|
||||
fut_a: self.a.call(req),
|
||||
b: self.b.clone(),
|
||||
fut_b: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ThenFuture<A, B, Request>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<Result<A::Response, A::Error>>,
|
||||
{
|
||||
b: B,
|
||||
fut_b: Option<B::Future>,
|
||||
fut_a: A::Future,
|
||||
}
|
||||
|
||||
impl<A, B, Request> Future for ThenFuture<A, B, Request>
|
||||
where
|
||||
A: Service<Request>,
|
||||
B: Service<Result<A::Response, A::Error>>,
|
||||
{
|
||||
type Item = B::Response;
|
||||
type Error = B::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
if let Some(ref mut fut) = self.fut_b {
|
||||
return fut.poll();
|
||||
}
|
||||
|
||||
match self.fut_a.poll() {
|
||||
Ok(Async::Ready(resp)) => {
|
||||
self.fut_b = Some(self.b.call(Ok(resp)));
|
||||
self.poll()
|
||||
}
|
||||
Err(err) => {
|
||||
self.fut_b = Some(self.b.call(Err(err)));
|
||||
self.poll()
|
||||
}
|
||||
Ok(Async::NotReady) => Ok(Async::NotReady),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use futures::future::{err, ok, FutureResult};
|
||||
use futures::{Async, Poll};
|
||||
use std::cell::Cell;
|
||||
use std::rc::Rc;
|
||||
|
||||
use super::*;
|
||||
use ServiceExt;
|
||||
|
||||
struct Srv1(Rc<Cell<usize>>);
|
||||
impl Service<Result<&'static str, &'static str>> for Srv1 {
|
||||
type Response = &'static str;
|
||||
type Error = ();
|
||||
type Future = FutureResult<Self::Response, Self::Error>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.0.set(self.0.get() + 1);
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Result<&'static str, &'static str>) -> Self::Future {
|
||||
match req {
|
||||
Ok(msg) => ok(msg),
|
||||
Err(_) => err(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Srv2(Rc<Cell<usize>>);
|
||||
|
||||
impl Service<Result<&'static str, ()>> for Srv2 {
|
||||
type Response = (&'static str, &'static str);
|
||||
type Error = ();
|
||||
type Future = FutureResult<Self::Response, ()>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.0.set(self.0.get() + 1);
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Result<&'static str, ()>) -> Self::Future {
|
||||
match req {
|
||||
Ok(msg) => ok((msg, "ok")),
|
||||
Err(()) => ok(("srv2", "err")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_poll_ready() {
|
||||
let cnt = Rc::new(Cell::new(0));
|
||||
let mut srv = Srv1(cnt.clone()).then(Srv2(cnt.clone()));
|
||||
let res = srv.poll_ready();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(()));
|
||||
assert_eq!(cnt.get(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_call() {
|
||||
let cnt = Rc::new(Cell::new(0));
|
||||
let mut srv = Srv1(cnt.clone()).then(Srv2(cnt));
|
||||
|
||||
let res = srv.call(Ok("srv1")).poll();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(("srv1", "ok")));
|
||||
|
||||
let res = srv.call(Err("srv")).poll();
|
||||
assert!(res.is_ok());
|
||||
assert_eq!(res.unwrap(), Async::Ready(("srv2", "err")));
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue