Create tower-test and include tower-mock. (#237)

tower-mock is deleted in favor of having a single test crate. This crate
also includes a new macro: `assert_request_eq!`.

Cargo.toml

@@ -9,10 +9,10 @@ members = [
   "tower-layer",
   "tower-limit",
   "tower-load-shed",
-  "tower-mock",
   "tower-reconnect",
   "tower-retry",
   "tower-service",
+  "tower-test",
   "tower-timeout",
   "tower-util",
 ]

README.md

@@ -45,15 +45,14 @@ crates.
 * [`tower-limit`]: Middleware limiting the number of requests that are
   processed ([docs][tlim-docs]).
 
-* [`tower-mock`]: Testing utility for mocking a `Service`. This is useful for
-  testing middleware implementations ([docs][tm-docs]);
-
 * [`tower-reconnect`]: Middleware that automatically reconnects the inner
   service when it becomes degraded ([docs][tre-docs]).
 
 * [`tower-retry`]: Middleware that retries requests based on a given `Policy`
   ([docs][tretry-docs]).
 
+* [`tower-test`]: Testing utilies ([docs][ttst-docs]).
+
 * [`tower-timeout`]: Middleware that applies a timeout to requests
   ([docs][tt-docs]).
 
@@ -91,13 +90,14 @@ terms or conditions.
 [tf-docs]: https://tower-rs.github.io/tower/doc/tower_filter/index.html
 [`tower-limit`]: tower-limit
 [tlim-docs]: https://tower-rs.github.io/tower/doc/tower_limit/index.html
-[`tower-mock`]: tower-mock
-[tm-docs]: https://tower-rs.github.io/tower/doc/tower_mock/index.html
 [`tower-reconnect`]: tower-reconnect
 [tre-docs]: https://tower-rs.github.io/tower/doc/tower_reconnect/index.html
 [`tower-retry`]: tower-retry
 [tretry-docs]: https://tower-rs.github.io/tower/doc/tower_retry/index.html
 [`tower-timeout`]: tower-timeout
+[`tower-test`]: tower-test
+[ttst-docs]: https://tower-rs.github.io/tower/doc/tower_test/index.html
+[`tower-rate-limit`]: tower-rate-limit
 [tt-docs]: https://tower-rs.github.io/tower/doc/tower_timeout/index.html
 [`tower-util`]: tower-util
 [tu-docs]: https://tower-rs.github.io/tower/doc/tower_util/index.html

azure-pipelines.yml

@@ -20,10 +20,10 @@ jobs:
       - tower-layer
       - tower-limit
       - tower-load-shed
-      - tower-mock
       - tower-reconnect
       - tower-retry
       - tower-service
+      - tower-test
       - tower-timeout
       - tower-util
       - tower

tower-buffer/Cargo.toml

@@ -12,4 +12,4 @@ tokio-executor = "0.1.7"
 tokio-sync = "0.1.0"
 
 [dev-dependencies]
-tower-mock = { version = "0.1", path = "../tower-mock" }
+tower-test = { version = "0.1", path = "../tower-test" }

tower-buffer/tests/buffer.rs

@@ -1,13 +1,15 @@
 extern crate futures;
 extern crate tokio_executor;
 extern crate tower_buffer;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use futures::prelude::*;
 use tokio_executor::{SpawnError, TypedExecutor};
 use tower_buffer::*;
 use tower_service::*;
+use tower_test::mock;
 
 use std::cell::RefCell;
 use std::thread;
@@ -18,9 +20,7 @@ fn req_and_res() {
 
     let response = service.call("hello");
 
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(*request, "hello");
-    request.respond("world");
 
     assert_eq!(response.wait().unwrap(), "world");
 }
@@ -33,8 +33,7 @@ fn clears_canceled_requests() {
 
     let res1 = service.call("hello");
 
-    let req1 = handle.next_request().unwrap();
+    let send_response1 = assert_request_eq!(handle, "hello");
-    assert_eq!(*req1, "hello");
 
     // don't respond yet, new requests will get buffered
 
@@ -47,15 +46,14 @@ fn clears_canceled_requests() {
 
     drop(res2);
 
-    req1.respond("world");
+    send_response1.send_response("world");
     assert_eq!(res1.wait().unwrap(), "world");
 
     // res2 was dropped, so it should have been canceled in the buffer
     handle.allow(1);
 
-    let req3 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello3").send_response("world3");
-    assert_eq!(*req3, "hello3");
+
-    req3.respond("world3");
     assert_eq!(res3.wait().unwrap(), "world3");
 }
 
@@ -77,9 +75,7 @@ fn when_inner_is_not_ready() {
 
     handle.allow(1);
 
-    let req1 = handle.next_request().expect("next_request1");
+    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(*req1, "hello");
-    req1.respond("world");
 
     assert_eq!(res1.wait().expect("res1.wait"), "world");
 }
@@ -92,7 +88,7 @@ fn when_inner_fails() {
 
     // Make the service NotReady
     handle.allow(0);
-    handle.error("foobar");
+    handle.send_error("foobar");
 
     let mut res1 = service.call("hello");
 
@@ -112,7 +108,7 @@ fn when_inner_fails() {
 
 #[test]
 fn poll_ready_when_worker_is_dropped_early() {
-    let (service, _handle) = Mock::new();
+    let (service, _handle) = mock::pair::<(), ()>();
 
     // drop that worker right on the floor!
     let mut exec = ExecFn(drop);
@@ -128,7 +124,7 @@ fn poll_ready_when_worker_is_dropped_early() {
 
 #[test]
 fn response_future_when_worker_is_dropped_early() {
-    let (service, mut handle) = Mock::new();
+    let (service, mut handle) = mock::pair::<_, ()>();
 
     // hold the worker in a cell until we want to drop it later
     let cell = RefCell::new(None);
@@ -146,8 +142,8 @@ fn response_future_when_worker_is_dropped_early() {
     response.wait().expect_err("res.wait");
 }
 
-type Mock = tower_mock::Mock<&'static str, &'static str>;
+type Mock = mock::Mock<&'static str, &'static str>;
-type Handle = tower_mock::Handle<&'static str, &'static str>;
+type Handle = mock::Handle<&'static str, &'static str>;
 
 struct Exec;
 
@@ -177,7 +173,7 @@ where
 }
 
 fn new_service() -> (Buffer<Mock, &'static str>, Handle) {
-    let (service, handle) = Mock::new();
+    let (service, handle) = mock::pair();
     // bound is >0 here because clears_canceled_requests needs multiple outstanding requests
     let service = Buffer::with_executor(service, 10, &mut Exec).unwrap();
     (service, handle)

tower-filter/Cargo.toml

@@ -10,4 +10,4 @@ tower-service = "0.2.0"
 tower-layer = { version = "0.1", path = "../tower-layer" }
 
 [dev-dependencies]
-tower-mock = { version = "0.1", path = "../tower-mock" }
+tower-test = { version = "0.1", path = "../tower-test" }

tower-filter/tests/filter.rs

@@ -1,12 +1,14 @@
 extern crate futures;
 extern crate tower_filter;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use futures::*;
 use tower_filter::error::Error;
 use tower_filter::Filter;
 use tower_service::*;
+use tower_test::mock;
 
 use std::thread;
 
@@ -17,12 +19,7 @@ fn passthrough_sync() {
     let th = thread::spawn(move || {
         // Receive the requests and respond
         for i in 0..10 {
-            let expect = format!("ping-{}", i);
+            assert_request_eq!(handle, format!("ping-{}", i)).send_response(format!("pong-{}", i));
-            let actual = handle.next_request().unwrap();
-
-            assert_eq!(actual.as_str(), expect.as_str());
-
-            actual.respond(format!("pong-{}", i));
         }
     });
 
@@ -53,15 +50,15 @@ fn rejected_sync() {
     assert!(response.is_err());
 }
 
-type Mock = tower_mock::Mock<String, String>;
+type Mock = mock::Mock<String, String>;
-type Handle = tower_mock::Handle<String, String>;
+type Handle = mock::Handle<String, String>;
 
 fn new_service<F, U>(f: F) -> (Filter<Mock, F>, Handle)
 where
     F: Fn(&String) -> U,
     U: IntoFuture<Item = (), Error = Error>,
 {
-    let (service, handle) = Mock::new();
+    let (service, handle) = mock::pair();
     let service = Filter::new(service, f);
     (service, handle)
 }

tower-limit/Cargo.toml

@@ -12,6 +12,6 @@ tokio-sync = "0.1.3"
 tokio-timer = "0.2.6"
 
 [dev-dependencies]
-tower-mock = { version = "0.1", path = "../tower-mock" }
+tower-test = { version = "0.1", path = "../tower-test" }
 tokio = "0.1"
 tokio-mock-task = "0.1.1"

tower-limit/tests/concurrency.rs

@@ -1,31 +1,34 @@
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tower_limit;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use tower_limit::concurrency::ConcurrencyLimit;
 use tower_service::Service;
+use tower_test::mock;
 
 use futures::future::{poll_fn, Future};
 use tokio_mock_task::MockTask;
 
 macro_rules! assert_ready {
     ($e:expr) => {{
+        use futures::Async::*;
         match $e {
-            Ok(futures::Async::Ready(v)) => v,
+            Ok(Ready(v)) => v,
-            Ok(_) => panic!("not ready"),
+            Ok(NotReady) => panic!("not ready"),
-            Err(e) => panic!("error = {:?}", e),
+            Err(e) => panic!("err = {:?}", e),
         }
     }};
 }
 
 macro_rules! assert_not_ready {
     ($e:expr) => {{
+        use futures::Async::*;
         match $e {
-            Ok(futures::Async::NotReady) => {}
+            Ok(NotReady) => {}
-            Ok(futures::Async::Ready(v)) => panic!("ready; value = {:?}", v),
+            r => panic!("unexpected poll status = {:?}", r),
-            Err(e) => panic!("error = {:?}", e),
         }
     }};
 }
@@ -49,14 +52,10 @@ fn basic_service_limit_functionality_with_poll_ready() {
     assert!(!task.is_notified());
 
     // The request gets passed through
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello 1").send_response("world 1");
-    assert_eq!(*request, "hello 1");
-    request.respond("world 1");
 
     // The next request gets passed through
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello 2").send_response("world 2");
-    assert_eq!(*request, "hello 2");
-    request.respond("world 2");
 
     // There are no more requests
     task.enter(|| {
@@ -80,34 +79,36 @@ fn basic_service_limit_functionality_with_poll_ready() {
     assert_eq!(r2.wait().unwrap(), "world 2");
 
     // The request gets passed through
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello 3").send_response("world 3");
-    assert_eq!(*request, "hello 3");
-    request.respond("world 3");
 
     assert_eq!(r3.wait().unwrap(), "world 3");
 }
 
 #[test]
-#[should_panic]
 fn basic_service_limit_functionality_without_poll_ready() {
     let mut task = MockTask::new();
 
     let (mut service, mut handle) = new_service(2);
 
+    assert_ready!(service.poll_ready());
     let r1 = service.call("hello 1");
+
+    assert_ready!(service.poll_ready());
     let r2 = service.call("hello 2");
-    let r3 = service.call("hello 3");
+
-    r3.wait().unwrap_err();
+    task.enter(|| {
+        assert_not_ready!(service.poll_ready());
+    });
 
     // The request gets passed through
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello 1").send_response("world 1");
-    assert_eq!(*request, "hello 1");
+
-    request.respond("world 1");
+    assert!(!task.is_notified());
 
     // The next request gets passed through
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello 2").send_response("world 2");
-    assert_eq!(*request, "hello 2");
+
-    request.respond("world 2");
+    assert!(!task.is_notified());
 
     // There are no more requests
     task.enter(|| {
@@ -116,41 +117,34 @@ fn basic_service_limit_functionality_without_poll_ready() {
 
     assert_eq!(r1.wait().unwrap(), "world 1");
 
+    assert!(task.is_notified());
+
     // One more request can be sent
+    assert_ready!(service.poll_ready());
     let r4 = service.call("hello 4");
 
-    let r5 = service.call("hello 5");
+    task.enter(|| {
-    r5.wait().unwrap_err();
+        assert_not_ready!(service.poll_ready());
+    });
 
     assert_eq!(r2.wait().unwrap(), "world 2");
+    assert!(task.is_notified());
 
     // The request gets passed through
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello 4").send_response("world 4");
-    assert_eq!(*request, "hello 4");
-    request.respond("world 4");
 
     assert_eq!(r4.wait().unwrap(), "world 4");
 }
 
 #[test]
-#[should_panic]
 fn request_without_capacity() {
     let mut task = MockTask::new();
 
-    let (mut service, mut handle) = new_service(0);
+    let (mut service, _) = new_service(0);
 
     task.enter(|| {
-        assert!(service.poll_ready().unwrap().is_not_ready());
+        assert_not_ready!(service.poll_ready());
     });
-
-    let response = service.call("hello");
-
-    // There are no more requests
-    task.enter(|| {
-        assert!(handle.poll_request().unwrap().is_not_ready());
-    });
-
-    response.wait().unwrap_err();
 }
 
 #[test]
@@ -173,8 +167,8 @@ fn reserve_capacity_without_sending_request() {
 
     // s1 sends the request, then s2 is able to get capacity
     let r1 = s1.call("hello");
-    let request = handle.next_request().unwrap();
+
-    request.respond("world");
+    assert_request_eq!(handle, "hello").send_response("world");
 
     task.enter(|| {
         assert!(s2.poll_ready().unwrap().is_not_ready());
@@ -196,20 +190,17 @@ fn service_drop_frees_capacity() {
     let mut s2 = s1.clone();
 
     // Reserve capacity in s1
-    task.enter(|| {
+    assert_ready!(s1.poll_ready());
-        assert!(s1.poll_ready().unwrap().is_ready());
-    });
 
     // Service 2 cannot get capacity
     task.enter(|| {
-        assert!(s2.poll_ready().unwrap().is_not_ready());
+        assert_not_ready!(s2.poll_ready());
     });
 
     drop(s1);
 
-    task.enter(|| {
+    assert!(task.is_notified());
-        assert!(s2.poll_ready().unwrap().is_ready());
+    assert_ready!(s2.poll_ready());
-    });
 }
 
 #[test]
@@ -222,13 +213,13 @@ fn response_error_releases_capacity() {
 
     // Reserve capacity in s1
     task.enter(|| {
-        assert!(s1.poll_ready().unwrap().is_ready());
+        assert_ready!(s1.poll_ready());
     });
 
     // s1 sends the request, then s2 is able to get capacity
     let r1 = s1.call("hello");
-    let request = handle.next_request().unwrap();
+
-    request.error("boom");
+    assert_request_eq!(handle, "hello").send_error("boom");
 
     r1.wait().unwrap_err();
 
@@ -247,14 +238,14 @@ fn response_future_drop_releases_capacity() {
 
     // Reserve capacity in s1
     task.enter(|| {
-        assert!(s1.poll_ready().unwrap().is_ready());
+        assert_ready!(s1.poll_ready());
     });
 
     // s1 sends the request, then s2 is able to get capacity
     let r1 = s1.call("hello");
 
     task.enter(|| {
-        assert!(s2.poll_ready().unwrap().is_not_ready());
+        assert_not_ready!(s2.poll_ready());
     });
 
     drop(r1);
@@ -291,11 +282,11 @@ fn multi_waiters() {
     assert!(task3.is_notified());
 }
 
-type Mock = tower_mock::Mock<&'static str, &'static str>;
+type Mock = mock::Mock<&'static str, &'static str>;
-type Handle = tower_mock::Handle<&'static str, &'static str>;
+type Handle = mock::Handle<&'static str, &'static str>;
 
 fn new_service(max: usize) -> (ConcurrencyLimit<Mock>, Handle) {
-    let (service, handle) = Mock::new();
+    let (service, handle) = mock::pair();
     let service = ConcurrencyLimit::new(service, max);
     (service, handle)
 }

tower-limit/tests/rate.rs

@@ -2,32 +2,53 @@ extern crate futures;
 extern crate tokio;
 extern crate tokio_timer;
 extern crate tower_limit;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use futures::future;
 use tower_limit::rate::*;
 use tower_service::*;
+use tower_test::mock;
 
 use std::time::{Duration, Instant};
 
+macro_rules! assert_ready {
+    ($e:expr) => {{
+        use futures::Async::*;
+        match $e {
+            Ok(Ready(v)) => v,
+            Ok(NotReady) => panic!("not ready"),
+            Err(e) => panic!("err = {:?}", e),
+        }
+    }};
+}
+
+macro_rules! assert_not_ready {
+    ($e:expr) => {{
+        use futures::Async::*;
+        match $e {
+            Ok(NotReady) => {}
+            r => panic!("unexpected poll status = {:?}", r),
+        }
+    }};
+}
+
 #[test]
 fn reaching_capacity() {
     let mut rt = tokio::runtime::current_thread::Runtime::new().unwrap();
     let (mut service, mut handle) = new_service(Rate::new(1, from_millis(100)));
 
-    assert!(service.poll_ready().unwrap().is_ready());
+    assert_ready!(service.poll_ready());
     let response = service.call("hello");
 
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(*request, "hello");
-    request.respond("world");
 
     let response = rt.block_on(response);
     assert_eq!(response.unwrap(), "world");
 
     rt.block_on(future::lazy(|| {
-        assert!(service.poll_ready().unwrap().is_not_ready());
+        assert_not_ready!(service.poll_ready());
         Ok::<_, ()>(())
     }))
     .unwrap();
@@ -42,24 +63,22 @@ fn reaching_capacity() {
     .unwrap();
 
     let poll_ready = rt.block_on(future::lazy(|| service.poll_ready()));
-    assert!(poll_ready.unwrap().is_ready());
+    assert_ready!(poll_ready);
 
     // Send a second request
     let response = service.call("two");
 
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "two").send_response("done");
-    assert_eq!(*request, "two");
-    request.respond("done");
 
     let response = rt.block_on(response);
     assert_eq!(response.unwrap(), "done");
 }
 
-type Mock = tower_mock::Mock<&'static str, &'static str>;
+type Mock = mock::Mock<&'static str, &'static str>;
-type Handle = tower_mock::Handle<&'static str, &'static str>;
+type Handle = mock::Handle<&'static str, &'static str>;
 
 fn new_service(rate: Rate) -> (RateLimit<Mock>, Handle) {
-    let (service, handle) = Mock::new();
+    let (service, handle) = mock::pair();
     let service = RateLimit::new(service, rate);
     (service, handle)
 }

tower-load-shed/Cargo.toml

@@ -11,4 +11,4 @@ tower-layer = { version = "0.1.0", path = "../tower-layer" }
 
 [dev-dependencies]
 tokio-mock-task = "0.1.1"
-tower-mock = { version = "0.1", path = "../tower-mock" }
+tower-test = { version = "0.1", path = "../tower-test" }

tower-load-shed/tests/load-shed.rs

@@ -1,11 +1,13 @@
 extern crate futures;
 extern crate tower_load_shed;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use futures::Future;
 use tower_load_shed::LoadShed;
 use tower_service::Service;
+use tower_test::mock;
 
 #[test]
 fn when_ready() {
@@ -20,10 +22,7 @@ fn when_ready() {
 
     let response = service.call("hello");
 
-    let request = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(*request, "hello");
-    request.respond("world");
-
     assert_eq!(response.wait().unwrap(), "world");
 }
 
@@ -46,11 +45,11 @@ fn when_not_ready() {
     assert!(err.is::<tower_load_shed::error::Overloaded>());
 }
 
-type Mock = tower_mock::Mock<&'static str, &'static str>;
+type Mock = mock::Mock<&'static str, &'static str>;
-type Handle = tower_mock::Handle<&'static str, &'static str>;
+type Handle = mock::Handle<&'static str, &'static str>;
 
 fn new_service() -> (LoadShed<Mock>, Handle) {
-    let (service, handle) = Mock::new();
+    let (service, handle) = mock::pair();
     let service = LoadShed::new(service);
     (service, handle)
 }

tower-retry/Cargo.toml

@@ -11,5 +11,5 @@ tower-layer = { version = "0.1", path = "../tower-layer" }
 tokio-timer = "0.2.4"
 
 [dev-dependencies]
-tower-mock = { version = "0.1", path = "../tower-mock" }
+tower-test = { version = "0.1", path = "../tower-test" }
 tokio-executor = "0.1.2"

tower-retry/tests/retry.rs

@@ -1,11 +1,13 @@
 extern crate futures;
-extern crate tower_mock;
 extern crate tower_retry;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use futures::{future, Future};
 use tower_retry::Policy;
 use tower_service::Service;
+use tower_test::mock;
 
 #[test]
 fn retry_errors() {
@@ -14,15 +16,11 @@ fn retry_errors() {
     assert!(service.poll_ready().unwrap().is_ready());
     let mut fut = service.call("hello");
 
-    let req1 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("retry me");
-    assert_eq!(*req1, "hello");
-    req1.error("retry me");
 
     assert_not_ready(&mut fut);
 
-    let req2 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(*req2, "hello");
-    req2.respond("world");
 
     assert_eq!(fut.wait().unwrap(), "world");
 }
@@ -34,22 +32,13 @@ fn retry_limit() {
     assert!(service.poll_ready().unwrap().is_ready());
     let mut fut = service.call("hello");
 
-    let req1 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("retry 1");
-    assert_eq!(*req1, "hello");
-    req1.error("retry 1");
-
     assert_not_ready(&mut fut);
 
-    let req2 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("retry 2");
-    assert_eq!(*req2, "hello");
-    req2.error("retry 2");
-
     assert_not_ready(&mut fut);
 
-    let req3 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("retry 3");
-    assert_eq!(*req3, "hello");
-    req3.error("retry 3");
-
     assert_eq!(fut.wait().unwrap_err().to_string(), "retry 3");
 }
 
@@ -60,15 +49,10 @@ fn retry_error_inspection() {
     assert!(service.poll_ready().unwrap().is_ready());
     let mut fut = service.call("hello");
 
-    let req1 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("retry 1");
-    assert_eq!(*req1, "hello");
-    req1.error("retry 1");
-
     assert_not_ready(&mut fut);
 
-    let req2 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("reject");
-    assert_eq!(*req2, "hello");
-    req2.error("reject");
     assert_eq!(fut.wait().unwrap_err().to_string(), "reject");
 }
 
@@ -79,10 +63,7 @@ fn retry_cannot_clone_request() {
     assert!(service.poll_ready().unwrap().is_ready());
     let fut = service.call("hello");
 
-    let req1 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_error("retry 1");
-    assert_eq!(*req1, "hello");
-    req1.error("retry 1");
-
     assert_eq!(fut.wait().unwrap_err().to_string(), "retry 1");
 }
 
@@ -95,10 +76,7 @@ fn success_with_cannot_clone() {
     assert!(service.poll_ready().unwrap().is_ready());
     let fut = service.call("hello");
 
-    let req1 = handle.next_request().unwrap();
+    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(*req1, "hello");
-    req1.respond("world");
-
     assert_eq!(fut.wait().unwrap(), "world");
 }
 
@@ -106,8 +84,8 @@ type Req = &'static str;
 type Res = &'static str;
 type InnerError = &'static str;
 type Error = Box<::std::error::Error + Send + Sync>;
-type Mock = tower_mock::Mock<Req, Res>;
+type Mock = mock::Mock<Req, Res>;
-type Handle = tower_mock::Handle<Req, Res>;
+type Handle = mock::Handle<Req, Res>;
 
 #[derive(Clone)]
 struct RetryErrors;
@@ -182,7 +160,7 @@ impl Policy<Req, Res, Error> for CannotClone {
 fn new_service<P: Policy<Req, Res, Error> + Clone>(
     policy: P,
 ) -> (tower_retry::Retry<P, Mock>, Handle) {
-    let (service, handle) = Mock::new();
+    let (service, handle) = mock::pair();
     let service = tower_retry::Retry::new(policy, service);
     (service, handle)
 }

tower-test/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "tower-mock"
+name = "tower-test"
 version = "0.1.0"
 authors = ["Carl Lerche <me@carllerche.com>"]
 publish = false
@@ -8,3 +8,6 @@ publish = false
 futures = "0.1"
 tokio-sync = "0.1.3"
 tower-service = "0.2.0"
+
+[dev-dependencies]
+tower = { version = "0.1.0", path = "../tower" }

tower-test/src/lib.rs

@@ -0,0 +1,8 @@
+//! Mock `Service` that can be used in tests.
+
+extern crate futures;
+extern crate tokio_sync;
+extern crate tower_service;
+
+mod macros;
+pub mod mock;

tower-test/src/macros.rs

@@ -0,0 +1,41 @@
+/// Asserts that the mock handle receives a new request equal to the given
+/// value.
+///
+/// On success, the [`SendResponse`] handle for the matched request is returned,
+/// allowing the caller to respond to the request. On failure, the macro panics.
+///
+/// # Examples
+///
+/// ```rust
+/// extern crate tower;
+/// #[macro_use]
+/// extern crate tower_test;
+///
+/// use tower::Service;
+/// use tower_test::mock;
+///
+/// # fn main() {
+/// let (mut mock, mut handle) = mock::pair();
+///
+/// assert!(mock.poll_ready().is_ok());
+/// let _response = mock.call("hello");
+///
+/// assert_request_eq!(handle, "hello")
+///     .send_response("world");
+/// # }
+/// ```
+#[macro_export]
+macro_rules! assert_request_eq {
+    ($mock_handle:expr, $expect:expr) => {
+        assert_request_eq!($mock_handle, $expect,)
+    };
+    ($mock_handle:expr, $expect:expr, $($arg:tt)*) => {{
+        let (actual, send_response) = match $mock_handle.next_request() {
+            Some(r) => r,
+            None => panic!("expected a request but none was received."),
+        };
+
+        assert_eq!(actual, $expect, $($arg)*);
+        send_response
+    }};
+}

tower-test/src/mock/future.rs

@@ -1,6 +1,6 @@
 //! Future types
 
-use error::{self, Error};
+use super::error::{self, Error};
 use futures::{Async, Future, Poll};
 use tokio_sync::oneshot;
 

tower-test/src/mock/mod.rs

@@ -1,14 +1,10 @@
 //! Mock `Service` that can be used in tests.
 
-extern crate futures;
-extern crate tokio_sync;
-extern crate tower_service;
-
 pub mod error;
 pub mod future;
 
-use error::Error;
+use self::error::Error;
-use future::ResponseFuture;
+use self::future::ResponseFuture;
 use futures::task::{self, Task};
 use futures::{Async, Future, Poll, Stream};
 use tokio_sync::{mpsc, oneshot};
@@ -16,7 +12,7 @@ use tower_service::Service;
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
-use std::{ops, u64};
+use std::u64;
 
 /// A mock service
 #[derive(Debug)]
@@ -34,15 +30,11 @@ pub struct Handle<T, U> {
     state: Arc<Mutex<State>>,
 }
 
-#[derive(Debug)]
+type Request<T, U> = (T, SendResponse<U>);
-pub struct Request<T, U> {
-    request: T,
-    respond: Respond<U>,
-}
 
-/// Respond to a request received by `Mock`.
+/// Send a response in reply to a received request.
 #[derive(Debug)]
-pub struct Respond<T> {
+pub struct SendResponse<T> {
     tx: oneshot::Sender<Result<T, Error>>,
 }
 
@@ -67,11 +59,8 @@ struct State {
 type Tx<T, U> = mpsc::UnboundedSender<Request<T, U>>;
 type Rx<T, U> = mpsc::UnboundedReceiver<Request<T, U>>;
 
-// ===== impl Mock =====
+/// Create a new `Mock` and `Handle` pair.
-
+pub fn pair<T, U>() -> (Mock<T, U>, Handle<T, U>) {
-impl<T, U> Mock<T, U> {
-    /// Create a new `Mock` and `Handle` pair.
-    pub fn new() -> (Self, Handle<T, U>) {
     let (tx, rx) = mpsc::unbounded_channel();
     let tx = Mutex::new(tx);
 
@@ -87,7 +76,6 @@ impl<T, U> Mock<T, U> {
     let handle = Handle { rx, state };
 
     (mock, handle)
-    }
 }
 
 impl<T, U> Service<T> for Mock<T, U> {
@@ -148,13 +136,9 @@ impl<T, U> Service<T> for Mock<T, U> {
         }
 
         let (tx, rx) = oneshot::channel();
+        let send_response = SendResponse { tx };
 
-        let request = Request {
+        match self.tx.lock().unwrap().try_send((request, send_response)) {
-            request,
-            respond: Respond { tx },
-        };
-
-        match self.tx.lock().unwrap().try_send(request) {
             Ok(_) => {}
             Err(_) => {
                 // TODO: Can this be reached
@@ -234,7 +218,7 @@ impl<T, U> Handle<T, U> {
     }
 
     /// Make the next poll_ method error with the given error.
-    pub fn error<E: Into<Error>>(&mut self, e: E) {
+    pub fn send_error<E: Into<Error>>(&mut self, e: E) {
         let mut state = self.state.lock().unwrap();
         state.err_with = Some(e.into());
 
@@ -265,40 +249,15 @@ impl<T, U> Drop for Handle<T, U> {
     }
 }
 
-// ===== impl Request =====
+// ===== impl SendResponse =====
 
-impl<T, U> Request<T, U> {
+impl<T> SendResponse<T> {
-    /// Split the request and respond handle
+    pub fn send_response(self, response: T) {
-    pub fn into_parts(self) -> (T, Respond<U>) {
-        (self.request, self.respond)
-    }
-
-    pub fn respond(self, response: U) {
-        self.respond.respond(response)
-    }
-
-    pub fn error<E: Into<Error>>(self, err: E) {
-        self.respond.error(err)
-    }
-}
-
-impl<T, U> ops::Deref for Request<T, U> {
-    type Target = T;
-
-    fn deref(&self) -> &T {
-        &self.request
-    }
-}
-
-// ===== impl Respond =====
-
-impl<T> Respond<T> {
-    pub fn respond(self, response: T) {
         // TODO: Should the result be dropped?
         let _ = self.tx.send(Ok(response));
     }
 
-    pub fn error<E: Into<Error>>(self, err: E) {
+    pub fn send_error<E: Into<Error>>(self, err: E) {
         // TODO: Should the result be dropped?
         let _ = self.tx.send(Err(err.into()));
     }

tower-test/tests/mock.rs

@@ -1,8 +1,10 @@
 extern crate futures;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 
 use tower_service::Service;
+use tower_test::mock;
 
 use futures::Future;
 
@@ -20,9 +22,7 @@ fn single_request_ready() {
     let mut response = mock.call("hello?".into());
 
     // Get the request from the handle
-    let request = handle.next_request().unwrap();
+    let send_response = assert_request_eq!(handle, "hello?");
-
-    assert_eq!(request.as_str(), "hello?");
 
     // Response is not ready
     with_task(|| {
@@ -30,7 +30,7 @@ fn single_request_ready() {
     });
 
     // Send the response
-    request.respond("yes?".into());
+    send_response.send_response("yes?".into());
 
     assert_eq!(response.wait().unwrap().as_str(), "yes?");
 }
@@ -51,11 +51,11 @@ fn backpressure() {
     mock.call("hello?".into());
 }
 
-type Mock = tower_mock::Mock<String, String>;
+type Mock = mock::Mock<String, String>;
-type Handle = tower_mock::Handle<String, String>;
+type Handle = mock::Handle<String, String>;
 
 fn new_mock() -> (Mock, Handle) {
-    Mock::new()
+    mock::pair()
 }
 
 // Helper to run some code within context of a task

tower-util/Cargo.toml

@@ -30,4 +30,4 @@ tower-layer = { version = "0.1.0", path = "../tower-layer" }
 [dev-dependencies]
 tokio-mock-task = "0.1"
 tower = { version = "0.1.0", path = "../tower" }
-tower-mock = { version = "0.1.0", path = "../tower-mock" }
+tower-test = { version = "0.1.0", path = "../tower-test" }

tower-util/tests/call_all.rs

@@ -1,8 +1,9 @@
 extern crate futures;
 extern crate tokio_mock_task;
 extern crate tower;
-extern crate tower_mock;
 extern crate tower_service;
+#[macro_use]
+extern crate tower_test;
 extern crate tower_util;
 
 use futures::future::{ok, FutureResult};
@@ -11,8 +12,8 @@ use futures::{Async, Poll, Stream};
 use std::cell::Cell;
 use std::rc::Rc;
 use tower::ServiceExt;
-use tower_mock::*;
 use tower_service::*;
+use tower_test::mock;
 
 type Error = Box<::std::error::Error + Send + Sync>;
 
@@ -123,26 +124,23 @@ fn ordered() {
 
 #[test]
 fn unordered() {
-    let (mock, mut handle) = Mock::<_, &'static str>::new();
+    let (mock, mut handle) = mock::pair::<_, &'static str>();
     let mut task = tokio_mock_task::MockTask::new();
     let requests = stream::iter_ok::<_, Error>(&["one", "two"]);
 
     let mut svc = mock.call_all(requests).unordered();
     assert_not_ready!(task.enter(|| svc.poll()));
 
-    let (req1, resp1) = handle.next_request().unwrap().into_parts();
+    let resp1 = assert_request_eq!(handle, &"one");
-    let (req2, resp2) = handle.next_request().unwrap().into_parts();
+    let resp2 = assert_request_eq!(handle, &"two");
 
-    assert_eq!(req1, &"one");
+    resp2.send_response("resp 1");
-    assert_eq!(req2, &"two");
-
-    resp2.respond("resp 1");
 
     let v = assert_ready!(task.enter(|| svc.poll()));
     assert_eq!(v, Some("resp 1"));
     assert_not_ready!(task.enter(|| svc.poll()));
 
-    resp1.respond("resp 2");
+    resp1.send_response("resp 2");
 
     let v = assert_ready!(task.enter(|| svc.poll()));
     assert_eq!(v, Some("resp 2"));