use std::marker::PhantomData; use futures::{try_ready, Async, Future, Poll}; use super::{IntoNewService, NewService, Service}; use crate::cell::Cell; /// 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. pub struct AndThen { a: A, b: Cell, } impl AndThen { /// Create new `AndThen` combinator pub fn new(a: A, b: B) -> Self where A: Service, B: Service, { Self { a, b: Cell::new(b) } } } impl Clone for AndThen where A: Clone, { fn clone(&self) -> Self { AndThen { a: self.a.clone(), b: self.b.clone(), } } } impl Service for AndThen where A: Service, B: Service, { type Request = A::Request; type Response = B::Response; type Error = A::Error; type Future = AndThenFuture; fn poll_ready(&mut self) -> Poll<(), Self::Error> { try_ready!(self.a.poll_ready()); self.b.get_mut().poll_ready() } fn call(&mut self, req: A::Request) -> Self::Future { AndThenFuture::new(self.a.call(req), self.b.clone()) } } pub struct AndThenFuture where A: Service, B: Service, { b: Cell, fut_b: Option, fut_a: Option, } impl AndThenFuture where A: Service, B: Service, { fn new(a: A::Future, b: Cell) -> Self { AndThenFuture { b, fut_a: Some(a), fut_b: None, } } } impl Future for AndThenFuture where A: Service, B: Service, { type Item = B::Response; type Error = A::Error; fn poll(&mut self) -> Poll { if let Some(ref mut fut) = self.fut_b { return fut.poll(); } match self.fut_a.as_mut().expect("Bug in actix-service").poll() { Ok(Async::Ready(resp)) => { let _ = self.fut_a.take(); self.fut_b = Some(self.b.get_mut().call(resp)); self.poll() } Ok(Async::NotReady) => Ok(Async::NotReady), Err(err) => Err(err), } } } /// `AndThenNewService` new service combinator pub struct AndThenNewService where A: NewService, B: NewService, { a: A, b: B, _t: PhantomData, } impl AndThenNewService where A: NewService, B: NewService, { /// Create new `AndThen` combinator pub fn new>(a: A, f: F) -> Self { Self { a, b: f.into_new_service(), _t: PhantomData, } } } impl NewService for AndThenNewService where A: NewService, B: NewService, { type Request = A::Request; type Response = B::Response; type Error = A::Error; type Service = AndThen; type InitError = A::InitError; type Future = AndThenNewServiceFuture; fn new_service(&self, cfg: &C) -> Self::Future { AndThenNewServiceFuture::new(self.a.new_service(cfg), self.b.new_service(cfg)) } } impl Clone for AndThenNewService where A: NewService + Clone, B: NewService + Clone, { fn clone(&self) -> Self { Self { a: self.a.clone(), b: self.b.clone(), _t: PhantomData, } } } pub struct AndThenNewServiceFuture where A: NewService, B: NewService, { fut_b: B::Future, fut_a: A::Future, a: Option, b: Option, } impl AndThenNewServiceFuture where A: NewService, B: NewService, { fn new(fut_a: A::Future, fut_b: B::Future) -> Self { AndThenNewServiceFuture { fut_a, fut_b, a: None, b: None, } } } impl Future for AndThenNewServiceFuture where A: NewService, B: NewService, { type Item = AndThen; type Error = A::InitError; fn poll(&mut self) -> Poll { if self.a.is_none() { if let Async::Ready(service) = self.fut_a.poll()? { self.a = Some(service); } } if self.b.is_none() { if let Async::Ready(service) = self.fut_b.poll()? { self.b = Some(service); } } if self.a.is_some() && self.b.is_some() { Ok(Async::Ready(AndThen::new( self.a.take().unwrap(), self.b.take().unwrap(), ))) } else { Ok(Async::NotReady) } } } #[cfg(test)] mod tests { use futures::future::{ok, FutureResult}; use futures::{Async, Poll}; use std::cell::Cell; use std::rc::Rc; use super::*; use crate::{NewService, Service, ServiceExt}; struct Srv1(Rc>); impl Service for Srv1 { type Request = &'static str; type Response = &'static str; type Error = (); type Future = FutureResult; 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>); impl Service for Srv2 { type Request = &'static str; type Response = (&'static str, &'static str); type Error = (); type Future = FutureResult; 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"))); } #[test] fn test_new_service() { let cnt = Rc::new(Cell::new(0)); let cnt2 = cnt.clone(); let blank = move || Ok::<_, ()>(Srv1(cnt2.clone())); let new_srv = blank .into_new_service() .and_then(move || Ok(Srv2(cnt.clone()))); if let Async::Ready(mut srv) = new_srv.new_service(&()).poll().unwrap() { let res = srv.call("srv1").poll(); assert!(res.is_ok()); assert_eq!(res.unwrap(), Async::Ready(("srv1", "srv2"))); } else { panic!() } } }