add client_timeout option

tests/test_pool.py

@@ -86,3 +86,17 @@ def test_preconnect2(self):
         for i in range(0, 5):
             pool.append(client)
         c.destroy()
+
+    @tornado.testing.gen_test
+    def test_timeout(self):
+        c = ClientPool(max_size=5, client_timeout=1)
+        client1 = yield c.get_connected_client()
+        c.release_client(client1)
+        client2 = yield c.get_connected_client()
+        c.release_client(client2)
+        self.assertTrue(client1 == client2)
+        yield tornado.gen.sleep(1)
+        client3 = yield c.get_connected_client()
+        self.assertFalse(client1 == client3)
+        c.release_client(client3)
+        c.destroy()

tornadis/pool.py

@@ -19,17 +19,22 @@ class ClientPool(object):
 
     Attributes:
         max_size (int): max size of the pool (-1 means "no limit").
+        client_timeout (int): timeout in seconds of a connection released
+            to the pool (-1 means "no timeout").
         client_kwargs (dict): Client constructor arguments
     """
 
-    def __init__(self, max_size=-1, **client_kwargs):
+    def __init__(self, max_size=-1, client_timeout=-1, **client_kwargs):
         """Constructor.
 
         Args:
             max_size (int): max size of the pool (-1 means "no limit").
+            client_timeout (int): timeout in seconds of a connection released
+                to the pool (-1 means "no timeout").
             client_kwargs (dict): Client constructor arguments.
         """
         self.max_size = max_size
+        self.client_timeout = client_timeout
         self.client_kwargs = client_kwargs
         self.__pool = deque()
         if self.max_size != -1:
@@ -57,12 +62,22 @@ def get_connected_client(self):
             while True:
                 client = self.__pool.popleft()
                 if client.is_connected():
+                    if self._is_expired_client(client):
+                        client.disconnect()
+                        continue
                     break
         except IndexError:
             client = self._make_client()
             yield client.connect()
         raise tornado.gen.Return(client)
 
+    def _is_expired_client(self, client):
+        if self.client_timeout != -1 and client.is_connected():
+            delta = client.get_last_state_change_timedelta()
+            if delta.total_seconds() >= self.client_timeout:
+                return True
+        return False
+
     def connected_client(self):
         """Returns a ContextManagerFuture to be yielded in a with statement.
 
@@ -90,7 +105,8 @@ def release_client(self, client):
         Args:
             client: Client object.
         """
-        self.__pool.append(client)
+        if not self._is_expired_client(client):
+            self.__pool.append(client)
         if self.__sem is not None:
             self.__sem.release()