diff --git a/src/client/mod.rs b/src/client/mod.rs index 2c71c29489..8a0928e4eb 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -7,7 +7,8 @@ use std::marker::PhantomData; use std::rc::Rc; use std::time::Duration; -use futures::{future, Poll, Future, Stream}; +use futures::{Future, Poll, Stream}; +use futures::future::{self, Executor}; #[cfg(feature = "compat")] use http; use tokio::reactor::Handle; @@ -25,6 +26,8 @@ pub use proto::response::Response; pub use proto::request::Request; pub use self::connect::{HttpConnector, Connect}; +use self::background::{bg, Background}; + mod connect; mod dns; mod pool; @@ -37,7 +40,7 @@ pub mod compat; // If the Connector is clone, then the Client can be clone easily. pub struct Client { connector: C, - handle: Handle, + executor: Exec, pool: Pool>, } @@ -74,18 +77,24 @@ impl Client { } impl Client { - /// Return a reference to a handle to the event loop this Client is associated with. - #[inline] + // Eventually, a Client won't really care about a tokio Handle, and only + // the executor used to spawn background tasks. Removing this method is + // a breaking change, so for now, it's just deprecated. + #[doc(hidden)] + #[deprecated] pub fn handle(&self) -> &Handle { - &self.handle + match self.executor { + Exec::Handle(ref h) => h, + Exec::Executor(..) => panic!("Client not built with a Handle"), + } } /// Create a new client with a specific connector. #[inline] - fn configured(config: Config, handle: &Handle) -> Client { + fn configured(config: Config, exec: Exec) -> Client { Client { connector: config.connector, - handle: handle.clone(), + executor: exec, pool: Pool::new(config.keep_alive, config.keep_alive_timeout) } } @@ -185,11 +194,11 @@ where C: Connect, let checkout = self.pool.checkout(domain.as_ref()); let connect = { - let handle = self.handle.clone(); + let executor = self.executor.clone(); let pool = self.pool.clone(); let pool_key = Rc::new(domain.to_string()); self.connector.connect(url) - .map(move |io| { + .and_then(move |io| { let (tx, rx) = mpsc::channel(0); let tx = HyperClient { tx: RefCell::new(tx), @@ -198,8 +207,8 @@ where C: Connect, let pooled = pool.pooled(pool_key, tx); let conn = proto::Conn::<_, _, proto::ClientTransaction, _>::new(io, pooled.clone()); let dispatch = proto::dispatch::Dispatcher::new(proto::dispatch::Client::new(rx), conn); - handle.spawn(dispatch.map_err(|err| debug!("client connection error: {}", err))); - pooled + executor.execute(dispatch.map_err(|e| debug!("client connection error: {}", e)))?; + Ok(pooled) }) }; @@ -236,7 +245,7 @@ impl Clone for Client { fn clone(&self) -> Client { Client { connector: self.connector.clone(), - handle: self.handle.clone(), + executor: self.executor.clone(), pool: self.pool.clone(), } } @@ -384,7 +393,18 @@ where C: Connect, /// Construct the Client with this configuration. #[inline] pub fn build(self, handle: &Handle) -> Client { - Client::configured(self, handle) + Client::configured(self, Exec::Handle(handle.clone())) + } + + /// Construct a Client with this configuration and an executor. + /// + /// The executor will be used to spawn "background" connection tasks + /// to drive requests and responses. + pub fn executor(self, executor: E) -> Client + where + E: Executor + 'static, + { + Client::configured(self, Exec::Executor(Rc::new(executor))) } } @@ -417,3 +437,65 @@ impl Clone for Config { } } } + + +// ===== impl Exec ===== + +#[derive(Clone)] +enum Exec { + Handle(Handle), + Executor(Rc>), +} + + +impl Exec { + fn execute(&self, fut: F) -> io::Result<()> + where + F: Future + 'static, + { + match *self { + Exec::Handle(ref h) => h.spawn(fut), + Exec::Executor(ref e) => { + e.execute(bg(Box::new(fut))) + .map_err(|err| { + debug!("executor error: {:?}", err.kind()); + io::Error::new( + io::ErrorKind::Other, + "executor error", + ) + })? + }, + } + Ok(()) + } +} + +// ===== impl Background ===== + +// The types inside this module are not exported out of the crate, +// so they are in essence un-nameable. +mod background { + use futures::{Future, Poll}; + + // This is basically `impl Future`, since the type is un-nameable, + // and only implementeds `Future`. + #[allow(missing_debug_implementations)] + pub struct Background { + inner: Box>, + } + + pub fn bg(fut: Box>) -> Background { + Background { + inner: fut, + } + } + + impl Future for Background { + type Item = (); + type Error = (); + + fn poll(&mut self) -> Poll { + self.inner.poll() + } + } +} diff --git a/tests/client.rs b/tests/client.rs index 4e05742acf..d77cba32f5 100644 --- a/tests/client.rs +++ b/tests/client.rs @@ -888,6 +888,44 @@ mod dispatch_impl { assert_eq!(closes.load(Ordering::Relaxed), 1); } + #[test] + fn client_custom_executor() { + let server = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = server.local_addr().unwrap(); + let mut core = Core::new().unwrap(); + let handle = core.handle(); + let closes = Arc::new(AtomicUsize::new(0)); + + let (tx1, rx1) = oneshot::channel(); + + thread::spawn(move || { + let mut sock = server.accept().unwrap().0; + sock.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + sock.set_write_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut buf = [0; 4096]; + sock.read(&mut buf).expect("read 1"); + sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n").unwrap(); + let _ = tx1.send(()); + }); + + let uri = format!("http://{}/a", addr).parse().unwrap(); + + let client = Client::configure() + .connector(DebugConnector(HttpConnector::new(1, &handle), closes.clone())) + .executor(handle.clone()); + let res = client.get(uri).and_then(move |res| { + assert_eq!(res.status(), hyper::StatusCode::Ok); + res.body().concat2() + }); + let rx = rx1.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked"))); + + let timeout = Timeout::new(Duration::from_millis(200), &handle).unwrap(); + let rx = rx.and_then(move |_| timeout.map_err(|e| e.into())); + core.run(res.join(rx).map(|r| r.0)).unwrap(); + + assert_eq!(closes.load(Ordering::Relaxed), 1); + } + struct DebugConnector(HttpConnector, Arc); impl Service for DebugConnector {