feat(client): allow to configure keep alive duration for http1 exclus…

…ive pool

client/src/builder.rs

@@ -24,6 +24,8 @@ pub struct ClientBuilder {
     connector: Connector,
     resolver: ResolverService,
     pool_capacity: usize,
+    keep_alive_idle: Duration,
+    keep_alive_born: Duration,
     timeout_config: TimeoutConfig,
     local_addr: Option<SocketAddr>,
     max_http_version: Version,
@@ -42,6 +44,8 @@ impl ClientBuilder {
             connector: connector::nop(),
             resolver: base_resolver(),
             pool_capacity: 2,
+            keep_alive_idle: Duration::from_secs(60),
+            keep_alive_born: Duration::from_secs(3600),
             timeout_config: TimeoutConfig::new(),
             local_addr: None,
             max_http_version: max_http_version(),
@@ -296,6 +300,28 @@ impl ClientBuilder {
         self
     }
 
+    /// Set duration of keep alive idle connection.
+    ///
+    /// This duration force a connection to be closed if it's idle for this long.
+    ///
+    /// Default to 10 minutes.
+    ///
+    pub fn set_keep_alive_idle(mut self, duration: Duration) -> Self {
+        self.keep_alive_idle = duration;
+        self
+    }
+
+    /// Set duration of keep alive born connection.
+    ///
+    /// This duration force a connection to be closed if it has been created for this long.
+    ///
+    /// Default to 1 hour.
+    ///
+    pub fn set_keep_alive_born(mut self, duration: Duration) -> Self {
+        self.keep_alive_born = duration;
+        self
+    }
+
     /// Set max http version client would be used.
     ///
     /// Default to the max version of http feature enabled within Cargo.toml
@@ -425,7 +451,7 @@ impl ClientBuilder {
         };
 
         Client {
-            exclusive_pool: pool::exclusive::Pool::with_capacity(self.pool_capacity),
+            exclusive_pool: pool::exclusive::Pool::new(self.pool_capacity, self.keep_alive_idle, self.keep_alive_born),
             shared_pool: pool::shared::Pool::with_capacity(self.pool_capacity),
             connector: self.connector,
             resolver: self.resolver,

client/src/pool/exclusive.rs

@@ -21,13 +21,17 @@ pub struct Pool<K, C> {
     // the pool can have unbounded entries with different keys but a single
     // entry can only have up to cap size of C inside it.
     cap: usize,
+    keep_alive_idle: Duration,
+    keep_alive_born: Duration,
 }
 
 impl<K, C> Clone for Pool<K, C> {
     fn clone(&self) -> Self {
         Self {
             conns: self.conns.clone(),
             cap: self.cap,
+            keep_alive_idle: self.keep_alive_idle,
+            keep_alive_born: self.keep_alive_born,
         }
     }
 }
@@ -36,10 +40,12 @@ impl<K, C> Pool<K, C>
 where
     K: Eq + Hash + Clone,
 {
-    pub(crate) fn with_capacity(cap: usize) -> Self {
+    pub(crate) fn new(cap: usize, keep_alive_idle: Duration, keep_alive_born: Duration) -> Self {
         Self {
             conns: Arc::new(Mutex::new(HashMap::new())),
             cap,
+            keep_alive_idle,
+            keep_alive_born,
         }
     }
 
@@ -114,7 +120,7 @@ where
                 if res.is_ok() {
                     queue.push_back(PooledConn {
                         conn,
-                        state: ConnState::new(),
+                        state: ConnState::new(self.keep_alive_idle, self.keep_alive_born),
                     });
                 }
             }
@@ -123,7 +129,7 @@ where
                 let mut queue = VecDeque::with_capacity(self.cap);
                 queue.push_back(PooledConn {
                     conn,
-                    state: ConnState::new(),
+                    state: ConnState::new(self.keep_alive_idle, self.keep_alive_born),
                 });
                 conns.insert(key, (permits, queue));
             }
@@ -257,15 +263,19 @@ impl<C> DerefMut for PooledConn<C> {
 struct ConnState {
     born: Instant,
     idle_since: Instant,
+    keep_alive_idle: Duration,
+    keep_alive_born: Duration,
 }
 
 impl ConnState {
-    fn new() -> Self {
+    fn new(keep_alive_idle: Duration, keep_alive_born: Duration) -> Self {
         let now = Instant::now();
 
         Self {
             born: now,
             idle_since: now,
+            keep_alive_idle,
+            keep_alive_born,
         }
     }
 
@@ -274,7 +284,7 @@ impl ConnState {
     }
 
     fn is_expired(&self) -> bool {
-        self.born.elapsed() > Duration::from_secs(3600) || self.idle_since.elapsed() > Duration::from_secs(600)
+        self.born.elapsed() > self.keep_alive_born || self.idle_since.elapsed() > self.keep_alive_idle
     }
 }
 
@@ -299,7 +309,7 @@ where
         if let Some((_, queue)) = self.pool.conns.lock().unwrap().get_mut(&self.key) {
             queue.push_back(PooledConn {
                 conn,
-                state: ConnState::new(),
+                state: ConnState::new(self.pool.keep_alive_idle, self.pool.keep_alive_born),
             });
         }
     }