Keep-alive timeouts.

httpx/concurrency/asyncio.py

@@ -252,6 +252,10 @@ async def open_uds_stream(
 
         return SocketStream(stream_reader=stream_reader, stream_writer=stream_writer)
 
+    def time(self) -> float:
+        loop = asyncio.get_event_loop()
+        return loop.time()
+
     async def run_in_threadpool(
         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:

httpx/concurrency/auto.py

@@ -41,6 +41,9 @@ async def open_uds_stream(
     ) -> BaseSocketStream:
         return await self.backend.open_uds_stream(path, hostname, ssl_context, timeout)
 
+    def time(self) -> float:
+        return self.backend.time()
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         return self.backend.get_semaphore(limits)
 

httpx/concurrency/base.py

@@ -105,6 +105,9 @@ async def open_uds_stream(
     ) -> BaseSocketStream:
         raise NotImplementedError()  # pragma: no cover
 
+    def time(self) -> float:
+        raise NotImplementedError()  # pragma: no cover
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         raise NotImplementedError()  # pragma: no cover
 

httpx/concurrency/trio.py

@@ -168,6 +168,9 @@ def run(
             functools.partial(coroutine, **kwargs) if kwargs else coroutine, *args
         )
 
+    def time(self) -> float:
+        return trio.current_time()
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         return PoolSemaphore(limits)
 

httpx/dispatch/connection.py

@@ -38,6 +38,7 @@ def __init__(
         self.release_func = release_func
         self.uds = uds
         self.open_connection: typing.Optional[OpenConnection] = None
+        self.timeout_at: typing.Optional[float] = None
 
     async def send(
         self,

httpx/dispatch/connection_pool.py

@@ -70,6 +70,8 @@ def __len__(self) -> int:
 
 
 class ConnectionPool(Dispatcher):
+    KEEP_ALIVE_TIMEOUT = 5.0
+
     def __init__(
         self,
         *,
@@ -93,6 +95,7 @@ def __init__(
         self.active_connections = ConnectionStore()
 
         self.backend = lookup_backend(backend)
+        self.next_keepalive_check = 0.0
 
     @property
     def max_connections(self) -> BasePoolSemaphore:
@@ -106,13 +109,26 @@ def max_connections(self) -> BasePoolSemaphore:
     def num_connections(self) -> int:
         return len(self.keepalive_connections) + len(self.active_connections)
 
+    async def keepalive_timeouts(self) -> None:
+        now = self.backend.time()
+        if now < self.next_keepalive_check:
+            return
+        self.next_keepalive_check = now + 1.0
+        keepalives = list(self.keepalive_connections.all.keys())
+        for connection in keepalives:
+            if connection.timeout_at is not None and now > connection.timeout_at:
+                self.keepalive_connections.remove(connection)
+                self.max_connections.release()
+                await connection.close()
+
     async def send(
         self,
         request: Request,
         verify: VerifyTypes = None,
         cert: CertTypes = None,
         timeout: Timeout = None,
     ) -> Response:
+        await self.keepalive_timeouts()
         connection = await self.acquire_connection(
             origin=request.url.origin, timeout=timeout
         )
@@ -168,6 +184,7 @@ async def release_connection(self, connection: HTTPConnection) -> None:
             self.max_connections.release()
             await connection.close()
         else:
+            connection.timeout_at = self.backend.time() + self.KEEP_ALIVE_TIMEOUT
             self.active_connections.remove(connection)
             self.keepalive_connections.add(connection)