update to latest tower and tower-hyper updates

Author: jxs

src/http-client.rs

@@ -4,7 +4,11 @@ use hyper::{
     Request, Response, Uri,
 };
 use std::time::Duration;
-use tower::{builder::ServiceBuilder, reconnect::Reconnect, Service, ServiceExt};
+use tower::{MakeService, Service};
+use tower::limit::rate::{Rate, RateLimit};
+use tower::limit::concurrency::ConcurrencyLimit;
+use tower::buffer::Buffer;
+use tower::retry::Retry;
 use tower_hyper::{
     client::{Builder, Connect},
     retry::{Body, RetryPolicy},
@@ -22,38 +26,42 @@ fn main() {
 
 fn request() -> impl Future<Item = Response<hyper::Body>, Error = ()> {
     let connector = Connector::new(HttpConnector::new(1));
-    let hyper = Connect::new(connector, Builder::new());
+    let mut hyper = Connect::new(connector, Builder::new());
 
-    // RetryPolicy is a very simple policy that retries `n` times
-    // if the response has a 500 status code. Here, `n` is 5.
-    let policy = RetryPolicy::new(5);
     // We're calling the tower/examples/server.rs.
     let dst = Destination::try_from_uri(Uri::from_static("http://127.0.0.1:3000")).unwrap();
 
-    // Now, to build the service! We use two BufferLayers in order to:
-    // - provide backpressure for the RateLimitLayer, and ConcurrencyLimitLayer
-    // - meet `RetryLayer`'s requirement that our service implement `Service + Clone`
-    // - ..and to provide cheap clones on the service.
-    let maker = ServiceBuilder::new()
-        .buffer(5)
-        .rate_limit(5, Duration::from_secs(1))
-        .concurrency_limit(5)
-        .retry(policy)
-        .buffer(5)
-        .make_service(hyper);
-
-    // `Reconnect` accepts a destination and a MakeService, creating a new service
-    // any time the connection encounters an error.
-    let client = Reconnect::new(maker, dst);
+    // Now, to build the service!
+    let client = hyper
+        .make_service(dst)
+        .map_err(|err| eprintln!("Connect Error {:?}", err))
+        .map(|conn| {
+            //We use two BufferLayers in order to:
+            // - provide backpressure for the RateLimitLayer, and ConcurrencyLimitLayer
+            // - meet `RetryLayer`'s requirement that our service implement `Service + Clone`
+            // - ..and to provide cheap clones on the service.
+            let buf = Buffer::new(conn, 1);
+
+            // // RetryPolicy is a very simple policy that retries `n` times
+            // // if the response has a 500 status code. Here, `n` is 5.
+            let policy = RetryPolicy::new(5);
+            let retry = Retry::new(policy, buf);
+
+            // - provide backpressure for the RateLimitLayer, and ConcurrencyLimitLayer
+            let rate_limit = RateLimit::new(retry, Rate::new(5, Duration::from_secs(1)));
+            let concurency_limit = ConcurrencyLimit::new(rate_limit, 5);
+
+            Buffer::new(concurency_limit, 1)
+        });
 
     let request = Request::builder()
         .method("GET")
         .body(Body::from(Vec::new()))
         .unwrap();
 
-    // we check to see if the client is ready to accept requests.
+    // let client = Reconnect::new(client, dst);
+
     client
-        .ready()
         .map_err(|e| panic!("Service is not ready: {:?}", e))
         .and_then(|mut c| {
             c.call(request)

src/http-server.rs

@@ -13,7 +13,7 @@ fn main() {
 
         let maker = ServiceBuilder::new()
             .concurrency_limit(5)
-            .make_service(MakeSvc);
+            .service(MakeSvc);
 
         let server = Server::new(maker);