Better encapsulation for connection instances (#353)

* Push Connection interface around

* Tighter encapsulation of connection interface

* Move ssl_context into start_tls method

* Pass keepalive_expiry to connection

* Push keepalive logic inside connection classes

* Fix ._state in HTTP2Connection

* may_close -> is_idle

* Docstring

* Update test cases

Author: lovelydinosaur

httpcore/_async/connection.py

@@ -5,12 +5,7 @@
 from .._exceptions import ConnectError, ConnectTimeout
 from .._types import URL, Headers, Origin, TimeoutDict
 from .._utils import exponential_backoff, get_logger, url_to_origin
-from .base import (
-    AsyncByteStream,
-    AsyncHTTPTransport,
-    ConnectionState,
-    NewConnectionRequired,
-)
+from .base import AsyncByteStream, AsyncHTTPTransport, NewConnectionRequired
 from .http import AsyncBaseHTTPConnection
 from .http11 import AsyncHTTP11Connection
 
@@ -25,6 +20,7 @@ def __init__(
         origin: Origin,
         http1: bool = True,
         http2: bool = False,
+        keepalive_expiry: float = None,
         uds: str = None,
         ssl_context: SSLContext = None,
         socket: AsyncSocketStream = None,
@@ -35,6 +31,7 @@ def __init__(
         self.origin = origin
         self.http1 = http1
         self.http2 = http2
+        self.keepalive_expiry = keepalive_expiry
         self.uds = uds
         self.ssl_context = SSLContext() if ssl_context is None else ssl_context
         self.socket = socket
@@ -57,20 +54,58 @@ def __init__(
         self.backend = AutoBackend() if backend is None else backend
 
     def __repr__(self) -> str:
-        http_version = "UNKNOWN"
-        if self.is_http11:
-            http_version = "HTTP/1.1"
-        elif self.is_http2:
-            http_version = "HTTP/2"
-        return f"<AsyncHTTPConnection http_version={http_version} state={self.state}>"
+        return f"<AsyncHTTPConnection [{self.info()}]>"
 
     def info(self) -> str:
         if self.connection is None:
-            return "Not connected"
-        elif self.state == ConnectionState.PENDING:
-            return "Connecting"
+            return "Connection failed" if self.connect_failed else "Connecting"
         return self.connection.info()
 
+    def should_close(self) -> bool:
+        """
+        Return `True` if the connection is in a state where it should be closed.
+        This occurs when any of the following occur:
+
+        * There are no active requests on an HTTP/1.1 connection, and the underlying
+          socket is readable. The only valid state the socket can be readable in
+          if this occurs is when the b"" EOF marker is about to be returned,
+          indicating a server disconnect.
+        * There are no active requests being made and the keepalive timeout has passed.
+        """
+        if self.connection is None:
+            return False
+        return self.connection.should_close()
+
+    def is_idle(self) -> bool:
+        """
+        Return `True` if the connection is currently idle.
+        """
+        if self.connection is None:
+            return False
+        return self.connection.is_idle()
+
+    def is_closed(self) -> bool:
+        if self.connection is None:
+            return self.connect_failed
+        return self.connection.is_closed()
+
+    def is_available(self) -> bool:
+        """
+        Return `True` if the connection is currently able to accept an outgoing request.
+        This occurs when any of the following occur:
+
+        * The connection has not yet been opened, and HTTP/2 support is enabled.
+          We don't *know* at this point if we'll end up on an HTTP/2 connection or
+          not, but we *might* do, so we indicate availability.
+        * The connection has been opened, and is currently idle.
+        * The connection is open, and is an HTTP/2 connection. The connection must
+          also not currently be exceeding the maximum number of allowable concurrent
+          streams and must not have exhausted the maximum total number of stream IDs.
+        """
+        if self.connection is None:
+            return self.http2 and not self.is_closed
+        return self.connection.is_available()
+
     @property
     def request_lock(self) -> AsyncLock:
         # We do this lazily, to make sure backend autodetection always
@@ -91,18 +126,16 @@ async def handle_async_request(
         timeout = cast(TimeoutDict, extensions.get("timeout", {}))
 
         async with self.request_lock:
-            if self.state == ConnectionState.PENDING:
+            if self.connection is None:
+                if self.connect_failed:
+                    raise NewConnectionRequired()
                 if not self.socket:
                     logger.trace(
                         "open_socket origin=%r timeout=%r", self.origin, timeout
                     )
                     self.socket = await self._open_socket(timeout)
                 self._create_connection(self.socket)
-            elif self.state in (ConnectionState.READY, ConnectionState.IDLE):
-                pass
-            elif self.state == ConnectionState.ACTIVE and self.is_http2:
-                pass
-            else:
+            elif not self.connection.is_available():
                 raise NewConnectionRequired()
 
         assert self.connection is not None
@@ -159,33 +192,24 @@ def _create_connection(self, socket: AsyncSocketStream) -> None:
 
             self.is_http2 = True
             self.connection = AsyncHTTP2Connection(
-                socket=socket, backend=self.backend, ssl_context=self.ssl_context
+                socket=socket,
+                keepalive_expiry=self.keepalive_expiry,
+                backend=self.backend,
             )
         else:
             self.is_http11 = True
             self.connection = AsyncHTTP11Connection(
-                socket=socket, ssl_context=self.ssl_context
+                socket=socket, keepalive_expiry=self.keepalive_expiry
             )
 
-    @property
-    def state(self) -> ConnectionState:
-        if self.connect_failed:
-            return ConnectionState.CLOSED
-        elif self.connection is None:
-            return ConnectionState.PENDING
-        return self.connection.get_state()
-
-    def is_socket_readable(self) -> bool:
-        return self.connection is not None and self.connection.is_socket_readable()
-
-    def mark_as_ready(self) -> None:
-        if self.connection is not None:
-            self.connection.mark_as_ready()
-
-    async def start_tls(self, hostname: bytes, timeout: TimeoutDict = None) -> None:
+    async def start_tls(
+        self, hostname: bytes, ssl_context: SSLContext, timeout: TimeoutDict = None
+    ) -> None:
         if self.connection is not None:
             logger.trace("start_tls hostname=%r timeout=%r", hostname, timeout)
-            self.socket = await self.connection.start_tls(hostname, timeout)
+            self.socket = await self.connection.start_tls(
+                hostname, ssl_context, timeout
+            )
             logger.trace("start_tls complete hostname=%r timeout=%r", hostname, timeout)
 
     async def aclose(self) -> None:

httpcore/_async/connection_pool.py

@@ -18,12 +18,7 @@
 from .._threadlock import ThreadLock
 from .._types import URL, Headers, Origin, TimeoutDict
 from .._utils import get_logger, origin_to_url_string, url_to_origin
-from .base import (
-    AsyncByteStream,
-    AsyncHTTPTransport,
-    ConnectionState,
-    NewConnectionRequired,
-)
+from .base import AsyncByteStream, AsyncHTTPTransport, NewConnectionRequired
 from .connection import AsyncHTTPConnection
 
 logger = get_logger(__name__)
@@ -184,6 +179,7 @@ def _create_connection(
             origin=origin,
             http1=self._http1,
             http2=self._http2,
+            keepalive_expiry=self._keepalive_expiry,
             uds=self._uds,
             ssl_context=self._ssl_context,
             local_address=self._local_address,
@@ -260,53 +256,15 @@ async def _get_connection_from_pool(
         self, origin: Origin
     ) -> Optional[AsyncHTTPConnection]:
         # Determine expired keep alive connections on this origin.
-        seen_http11 = False
-        pending_connection = None
         reuse_connection = None
         connections_to_close = set()
 
         for connection in self._connections_for_origin(origin):
-            if connection.is_http11:
-                seen_http11 = True
-
-            if connection.state == ConnectionState.IDLE:
-                if connection.is_socket_readable():
-                    # If the socket is readable while the connection is idle (meaning
-                    # we don't expect the server to send any data), then the only valid
-                    # reason is that the other end has disconnected, and is readable
-                    # because it is ready to return the b"" disconnect indicator, which
-                    # means we should drop the connection too.
-                    # (For a detailed run-through of what a "readable" socket is, and
-                    # why this is the best thing for us to do here, see:
-                    # https://github.com/encode/httpx/pull/143#issuecomment-515181778)
-                    logger.trace("removing dropped idle connection=%r", connection)
-                    # IDLE connections that have been dropped should be
-                    # removed from the pool.
-                    connections_to_close.add(connection)
-                    await self._remove_from_pool(connection)
-                else:
-                    # IDLE connections that are still maintained may
-                    # be reused.
-                    logger.trace("reusing idle http11 connection=%r", connection)
-                    reuse_connection = connection
-            elif connection.state == ConnectionState.ACTIVE and connection.is_http2:
-                # HTTP/2 connections may be reused.
-                logger.trace("reusing active http2 connection=%r", connection)
+            if connection.should_close():
+                connections_to_close.add(connection)
+                await self._remove_from_pool(connection)
+            elif connection.is_available():
                 reuse_connection = connection
-            elif connection.state == ConnectionState.PENDING:
-                # Pending connections may potentially be reused.
-                pending_connection = connection
-
-        if reuse_connection is not None:
-            # Mark the connection as READY before we return it, to indicate
-            # that if it is HTTP/1.1 then it should not be re-acquired.
-            reuse_connection.mark_as_ready()
-            reuse_connection.expires_at = None
-        elif self._http2 and pending_connection is not None and not seen_http11:
-            # If we have a PENDING connection, and no HTTP/1.1 connections
-            # on this origin, then we can attempt to share the connection.
-            logger.trace("reusing pending connection=%r", connection)
-            reuse_connection = pending_connection
 
         # Close any dropped connections.
         for connection in connections_to_close:
@@ -318,19 +276,16 @@ async def _response_closed(self, connection: AsyncHTTPConnection) -> None:
         remove_from_pool = False
         close_connection = False
 
-        if connection.state == ConnectionState.CLOSED:
+        if connection.is_closed():
             remove_from_pool = True
-        elif connection.state == ConnectionState.IDLE:
+        elif connection.is_idle():
             num_connections = len(self._get_all_connections())
             if (
                 self._max_keepalive_connections is not None
                 and num_connections > self._max_keepalive_connections
             ):
                 remove_from_pool = True
                 close_connection = True
-            elif self._keepalive_expiry is not None:
-                now = await self._backend.time()
-                connection.expires_at = now + self._keepalive_expiry
 
         if remove_from_pool:
             await self._remove_from_pool(connection)
@@ -353,11 +308,7 @@ async def _keepalive_sweep(self) -> None:
         connections_to_close = set()
 
         for connection in self._get_all_connections():
-            if (
-                connection.state == ConnectionState.IDLE
-                and connection.expires_at is not None
-                and now >= connection.expires_at
-            ):
+            if connection.should_close():
                 connections_to_close.add(connection)
                 await self._remove_from_pool(connection)
 

httpcore/_async/http.py

@@ -1,33 +1,40 @@
+from ssl import SSLContext
+
 from .._backends.auto import AsyncSocketStream
 from .._types import TimeoutDict
-from .base import AsyncHTTPTransport, ConnectionState
+from .base import AsyncHTTPTransport
 
 
 class AsyncBaseHTTPConnection(AsyncHTTPTransport):
     def info(self) -> str:
         raise NotImplementedError()  # pragma: nocover
 
-    def get_state(self) -> ConnectionState:
+    def should_close(self) -> bool:
+        """
+        Return `True` if the connection is in a state where it should be closed.
+        """
+        raise NotImplementedError()  # pragma: nocover
+
+    def is_idle(self) -> bool:
         """
-        Return the current state.
+        Return `True` if the connection is currently idle.
         """
         raise NotImplementedError()  # pragma: nocover
 
-    def mark_as_ready(self) -> None:
+    def is_closed(self) -> bool:
         """
-        The connection has been acquired from the pool, and the state
-        should reflect that.
+        Return `True` if the connection has been closed.
         """
         raise NotImplementedError()  # pragma: nocover
 
-    def is_socket_readable(self) -> bool:
+    def is_available(self) -> bool:
         """
-        Return 'True' if the underlying network socket is readable.
+        Return `True` if the connection is currently able to accept an outgoing request.
         """
         raise NotImplementedError()  # pragma: nocover
 
     async def start_tls(
-        self, hostname: bytes, timeout: TimeoutDict = None
+        self, hostname: bytes, ssl_context: SSLContext, timeout: TimeoutDict = None
     ) -> AsyncSocketStream:
         """
         Upgrade the underlying socket to TLS.