[WIP] Connection Pool Refactor

composer.json

@@ -25,7 +25,8 @@
         "php": ">=7.1",
         "amphp/amp": "^2",
         "amphp/byte-stream": "^1.6",
-        "amphp/http": "^1.2",
+        "amphp/hpack": "^2",
+        "amphp/http": "^1.3",
         "amphp/socket": "dev-master as 1.0",
         "amphp/file": "^0.2 || ^0.3",
         "kelunik/certificate": "^1.1",

src/Body/StringBody.php

@@ -19,7 +19,7 @@ public function __construct(string $body)
 
     public function createBodyStream(): InputStream
     {
-        return new InMemoryStream($this->body);
+        return new InMemoryStream(\strlen($this->body) ? $this->body : null);
     }
 
     public function getHeaders(): Promise

src/Client.php

@@ -20,7 +20,7 @@ interface Client
      */
     public function request(
         Request $request,
-        CancellationToken $cancellation = null
+        ?CancellationToken $cancellation = null
     ): Promise;
 
     /**

src/ClientBuilder.php

@@ -2,19 +2,15 @@
 
 namespace Amp\Http\Client;
 
-use Amp\Http\Client\Internal\ApplicationInterceptorClient;
-use Amp\Socket\SocketPool;
-use Amp\Socket\UnlimitedSocketPool;
-
 final class ClientBuilder
 {
-    private $socketPool;
+    private $connectionPool;
     private $networkInterceptors = [];
     private $applicationInterceptors = [];
 
-    public function __construct(?SocketPool $socketPool = null)
+    public function __construct(?Connection\ConnectionPool $connectionPool = null)
     {
-        $this->socketPool = $socketPool ?? new UnlimitedSocketPool;
+        $this->connectionPool = $connectionPool ?? new Connection\DefaultConnectionPool;
     }
 
     public function addNetworkInterceptor(NetworkInterceptor $networkInterceptor): self
@@ -33,14 +29,14 @@ public function addApplicationInterceptor(ApplicationInterceptor $applicationInt
 
     public function build(): Client
     {
-        $client = new SocketClient($this->socketPool);
+        $client = new SocketClient($this->connectionPool);
         foreach ($this->networkInterceptors as $networkInterceptor) {
             $client->addNetworkInterceptor($networkInterceptor);
         }
 
         $applicationInterceptors = $this->applicationInterceptors;
         while ($applicationInterceptor = \array_pop($applicationInterceptors)) {
-            $client = new ApplicationInterceptorClient($client, $applicationInterceptor);
+            $client = new Internal\ApplicationInterceptorClient($client, $applicationInterceptor);
         }
 
         return $client;

src/Connection/Connection.php

@@ -0,0 +1,35 @@
+<?php
+
+namespace Amp\Http\Client\Connection;
+
+use Amp\CancellationToken;
+use Amp\Http\Client\Request;
+use Amp\Http\Client\Response;
+use Amp\Promise;
+use Amp\Socket\SocketAddress;
+use Amp\Socket\TlsInfo;
+
+interface Connection
+{
+    public const MAX_KEEP_ALIVE_TIMEOUT = 60;
+
+    /**
+     * @param Request           $request
+     * @param CancellationToken $token
+     *
+     * @return Promise<Response>
+     */
+    public function request(Request $request, CancellationToken $token): Promise;
+
+    public function isBusy(): bool;
+
+    public function close(): Promise;
+
+    public function onClose(callable $onClose): void;
+
+    public function getLocalAddress(): SocketAddress;
+
+    public function getRemoteAddress(): SocketAddress;
+
+    public function getTlsInfo(): ?TlsInfo;
+}

src/Connection/ConnectionPool.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Amp\Http\Client\Connection;
+
+use Amp\CancellationToken;
+use Amp\Http\Client\Request;
+use Amp\Promise;
+
+interface ConnectionPool
+{
+    /**
+     * @param Request           $request
+     * @param CancellationToken $token
+     *
+     * @return Promise<Connection>
+     */
+    public function getConnection(Request $request, CancellationToken $token): Promise;
+}

src/Connection/DefaultConnectionPool.php

@@ -0,0 +1,150 @@
+<?php
+
+namespace Amp\Http\Client\Connection;
+
+use Amp\ByteStream\StreamException;
+use Amp\CancellationToken;
+use Amp\CancelledException;
+use Amp\Http\Client\Internal\CombinedCancellationToken;
+use Amp\Http\Client\Request;
+use Amp\Http\Client\SocketException;
+use Amp\Http\Client\TimeoutException;
+use Amp\Promise;
+use Amp\Socket;
+use Amp\Socket\ClientTlsContext;
+use Amp\Socket\ConnectContext;
+use Amp\Socket\Connector;
+use Amp\Socket\EncryptableSocket;
+use Amp\TimeoutCancellationToken;
+use function Amp\call;
+
+final class DefaultConnectionPool implements ConnectionPool
+{
+    /** @var Connector */
+    private $connector;
+
+    /** @var \SplObjectStorage[] */
+    private $connections = [];
+
+    public function __construct(?Connector $connector = null)
+    {
+        $this->connector = $connector ?? Socket\connector();
+    }
+
+    public function getConnection(Request $request, CancellationToken $cancellation): Promise
+    {
+        return call(function () use ($request, $cancellation) {
+            $uri = $request->getUri();
+            $scheme = \strtolower($uri->getScheme());
+            $isHttps = $scheme === 'https';
+            $defaultPort = $isHttps ? 443 : 80;
+
+            $host = \strtolower($uri->getHost());
+            $port = $uri->getPort() ?? $defaultPort;
+
+            if ($host === '') {
+                throw new \Error('A host must be provided in the URI');
+            }
+
+            $authority = $host . ':' . $port;
+            $key = $scheme . '://' . $authority;
+
+            $this->connections[$key] = $this->connections[$key] ?? new \SplObjectStorage;
+
+            foreach ($this->connections[$key] as $connection) {
+                $connection = yield $connection;
+                \assert($connection instanceof Connection);
+
+                if (!$connection->isBusy()) {
+                    return $connection;
+                }
+            }
+
+            $promise = call(function () use (&$promise, $request, $isHttps, $authority, $cancellation, $key) {
+                $connectContext = new ConnectContext;
+
+                if ($isHttps) {
+                    $tlsContext = ($connectContext->getTlsContext() ?? new ClientTlsContext($request->getUri()->getHost()))
+                        ->withApplicationLayerProtocols(['http/1.1'])
+                        ->withPeerCapturing();
+
+                    if (\in_array('2.0', $request->getProtocolVersions(), true)) {
+                        $tlsContext = $tlsContext->withApplicationLayerProtocols(['h2', 'http/1.1']);
+                    }
+
+                    $connectContext = $connectContext->withTlsContext($tlsContext);
+                }
+
+                try {
+                    $checkoutCancellationToken = new CombinedCancellationToken($cancellation, new TimeoutCancellationToken($request->getTcpConnectTimeout()));
+
+                    /** @var EncryptableSocket $socket */
+                    $socket = yield $this->connector->connect('tcp://' . $authority, $connectContext, $checkoutCancellationToken);
+                } catch (Socket\ConnectException $e) {
+                    throw new SocketException(\sprintf("Connection to '%s' failed", $authority), 0, $e);
+                } catch (CancelledException $e) {
+                    // In case of a user cancellation request, throw the expected exception
+                    $cancellation->throwIfRequested();
+
+                    // Otherwise we ran into a timeout of our TimeoutCancellationToken
+                    throw new TimeoutException(\sprintf("Connection to '%s' timed out, took longer than " . $request->getTcpConnectTimeout() . ' ms', $authority)); // don't pass $e
+                }
+
+                if ($isHttps) {
+                    try {
+                        $tlsState = $socket->getTlsState();
+                        if ($tlsState === EncryptableSocket::TLS_STATE_DISABLED) {
+                            $tlsCancellationToken = new CombinedCancellationToken($cancellation, new TimeoutCancellationToken($request->getTlsHandshakeTimeout()));
+                            yield $socket->setupTls($tlsCancellationToken);
+                        } elseif ($tlsState !== EncryptableSocket::TLS_STATE_ENABLED) {
+                            throw new SocketException('Failed to setup TLS connection, connection was in an unexpected TLS state (' . $tlsState . ')');
+                        }
+                    } catch (StreamException $exception) {
+                        throw new SocketException(\sprintf("Connection to '%s' closed during TLS handshake", $authority), 0, $exception);
+                    } catch (CancelledException $e) {
+                        // In case of a user cancellation request, throw the expected exception
+                        $cancellation->throwIfRequested();
+
+                        // Otherwise we ran into a timeout of our TimeoutCancellationToken
+                        throw new TimeoutException(\sprintf("TLS handshake with '%s' @ '%s' timed out, took longer than " . $request->getTlsHandshakeTimeout() . ' ms', $authority, $socket->getRemoteAddress()->toString())); // don't pass $e
+                    }
+                }
+
+                if ($isHttps && $socket->getTlsInfo()->getApplicationLayerProtocol() === 'h2') {
+                    $connection = new Http2Connection($socket);
+                } else {
+                    $connection = new Http1Connection($socket);
+                }
+
+                \assert($promise instanceof Promise);
+
+                $connection->onClose(function () use ($key, $promise) {
+                    $this->connections[$key]->detach($promise);
+
+                    if (!$this->connections[$key]->count()) {
+                        unset($this->connections[$key]);
+                    }
+                });
+
+                return $connection;
+            });
+
+            $this->connections[$key]->attach($promise);
+
+            $promise->onResolve(function (?\Throwable $exception) use ($key, $promise): void {
+                if (!$exception) {
+                    return;
+                }
+
+                // Connection failed, remove from list of connections.
+                $this->connections[$key]->detach($promise);
+
+                if (!$this->connections[$key]->count()) {
+                    unset($this->connections[$key]);
+                }
+            });
+
+            return $promise;
+        });
+    }
+}