Merge branch 'master' into session_special_chars

src/client/connector.rs

@@ -34,6 +34,8 @@ use {HAS_OPENSSL, HAS_TLS};
 pub struct ClientConnectorStats {
     /// Number of waited-on connections
     pub waits: usize,
+    /// Size of the wait queue
+    pub wait_queue: usize,
     /// Number of reused connections
     pub reused: usize,
     /// Number of opened connections
@@ -494,8 +496,13 @@ impl ClientConnector {
         ctx.run_later(Duration::from_secs(1), |act, ctx| act.collect_periodic(ctx));
 
         // send stats
-        let stats = mem::replace(&mut self.stats, ClientConnectorStats::default());
+        let mut stats = mem::replace(&mut self.stats, ClientConnectorStats::default());
         if let Some(ref mut subscr) = self.subscriber {
+            if let Some(ref waiters) = self.waiters {
+                for w in waiters.values() {
+                    stats.wait_queue += w.len();
+                }
+            }
             let _ = subscr.do_send(stats);
         }
     }

src/client/pipeline.rs

@@ -71,7 +71,7 @@ pub struct SendRequest {
     conn: Option<Addr<ClientConnector>>,
     conn_timeout: Duration,
     wait_timeout: Duration,
-    timeout: Option<Delay>,
+    timeout: Option<Duration>,
 }
 
 impl SendRequest {
@@ -115,7 +115,7 @@ impl SendRequest {
     /// Request timeout is the total time before a response must be received.
     /// Default value is 5 seconds.
     pub fn timeout(mut self, timeout: Duration) -> Self {
-        self.timeout = Some(Delay::new(Instant::now() + timeout));
+        self.timeout = Some(timeout);
         self
     }
 
@@ -185,9 +185,10 @@ impl Future for SendRequest {
                         _ => IoBody::Done,
                     };
 
-                    let timeout = self.timeout.take().unwrap_or_else(|| {
+                    let timeout = self
-                        Delay::new(Instant::now() + Duration::from_secs(5))
+                        .timeout
-                    });
+                        .take()
+                        .unwrap_or_else(|| Duration::from_secs(5));
 
                     let pl = Box::new(Pipeline {
                         body,
@@ -201,7 +202,7 @@ impl Future for SendRequest {
                         decompress: None,
                         should_decompress: self.req.response_decompress(),
                         write_state: RunningState::Running,
-                        timeout: Some(timeout),
+                        timeout: Some(Delay::new(Instant::now() + timeout)),
                         close: self.req.method() == &Method::HEAD,
                     });
                     self.state = State::Send(pl);