Merge 0f018b0 into 45074e8

src/ClientBuilder.php

@@ -3,18 +3,18 @@
 namespace Amp\Http\Client;
 
 use Amp\Http\Client\Internal\ApplicationInterceptorClient;
-use Amp\Socket\SocketPool;
-use Amp\Socket\UnlimitedSocketPool;
+use Amp\Socket\Connector;
+use Amp\Socket\DnsConnector;
 
 final class ClientBuilder
 {
-    private $socketPool;
+    private $connector;
     private $networkInterceptors = [];
     private $applicationInterceptors = [];
 
-    public function __construct(?SocketPool $socketPool = null)
+    public function __construct(?Connector $connector = null)
     {
-        $this->socketPool = $socketPool ?? new UnlimitedSocketPool;
+        $this->connector = $connector ?? new DnsConnector;
     }
 
     public function addNetworkInterceptor(NetworkInterceptor $networkInterceptor): self
@@ -33,7 +33,7 @@ public function addApplicationInterceptor(ApplicationInterceptor $applicationInt
 
     public function build(): Client
     {
-        $client = new SocketClient($this->socketPool);
+        $client = new SocketClient($this->connector);
         foreach ($this->networkInterceptors as $networkInterceptor) {
             $client->addNetworkInterceptor($networkInterceptor);
         }

src/Connection/Connection.php

@@ -0,0 +1,29 @@
+<?php
+
+namespace Amp\Http\Client\Connection;
+
+use Amp\CancellationToken;
+use Amp\Http\Client\ConnectionInfo;
+use Amp\Http\Client\Request;
+use Amp\Http\Client\Response;
+use Amp\Promise;
+
+interface Connection
+{
+    /**
+     * @param Request                $request
+     * @param CancellationToken|null $token
+     *
+     * @return Promise<Response>
+     */
+    public function request(Request $request, ?CancellationToken $token = null): Promise;
+
+    /**
+     * @return ConnectionInfo
+     */
+    public function getConnectionInfo(): ConnectionInfo;
+
+    public function isClosed(): bool;
+
+    public function close(): Promise;
+}

src/Connection/ConnectionPool.php

@@ -0,0 +1,28 @@
+<?php
+
+namespace Amp\Http\Client\Connection;
+
+use Amp\Http\Client\Request;
+use Amp\Socket\Socket;
+
+interface ConnectionPool
+{
+    /**
+     * @param Socket  $socket
+     *
+     * @return Connection
+     */
+    public function createConnection(Socket $socket): Connection;
+
+    /**
+     * @param Request $request
+     *
+     * @return Connection|null
+     */
+    public function getConnection(Request $request): ?Connection;
+
+    /**
+     * @return string[] A list of supported application-layer protocols (ALPNs).
+     */
+    public function getApplicationLayerProtocols(): array;
+}

src/Connection/DefaultConnectionPool.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace Amp\Http\Client\Connection;
+
+use Amp\Http\Client\Request;
+use Amp\Socket\Socket;
+
+final class DefaultConnectionPool implements ConnectionPool
+{
+    private $connections = [];
+
+    public function createConnection(Socket $socket): Connection
+    {
+        $connection = new Http1Connection($socket);
+        $this->connections[$socket->getRemoteAddress()->toString()] = $connection;
+        return $connection;
+    }
+
+    public function getConnection(Request $request): ?Connection
+    {
+        $uri = $request->getUri();
+
+        $address = $uri->getHost();
+        $port = $uri->getPort();
+
+        if ($port !== null) {
+            $address .= ':' . $port;
+        }
+
+        if (isset($this->connections[$address])) {
+            return $this->connections[$address];
+        }
+
+        return null;
+    }
+
+    public function getApplicationLayerProtocols(): array
+    {
+        return ['http/1.1'];
+    }
+}

src/Driver/Http1Driver.php → src/Connection/Http1Connection.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Amp\Http\Client\Driver;
+namespace Amp\Http\Client\Connection;
 
 use Amp\ByteStream\IteratorStream;
 use Amp\ByteStream\StreamException;
@@ -33,16 +33,36 @@
  *
  * @see Client
  */
-final class Http1Driver implements HttpDriver
+final class Http1Connection implements Connection
 {
+    private $socket;
+    private $connectionInfo;
+
+    public function __construct(Socket $socket)
+    {
+        $this->socket = $socket;
+        $this->connectionInfo = ConnectionInfo::fromSocket($socket);
+    }
+
+    public function getConnectionInfo(): ConnectionInfo
+    {
+        return $this->connectionInfo;
+    }
+
+    public function isClosed(): bool
+    {
+        return $this->socket->isClosed();
+    }
+
+    public function close(): Promise
+    {
+        $this->socket->close();
+        return new Success;
+    }
+
     /** @inheritdoc */
-    public function request(
-        Socket $socket,
-        ConnectionInfo $connectionInfo,
-        Request $request,
-        ?CancellationToken $cancellation = null
-    ): Promise {
-        return call(function () use ($request, $socket, $connectionInfo, $cancellation) {
+    public function request(Request $request, ?CancellationToken $cancellation = null): Promise {
+        return call(function () use ($request, $cancellation) {
             $cancellation = $cancellation ?? new NullCancellationToken;
 
             /** @var Request $request */
@@ -58,18 +78,16 @@ public function request(
                 $readingCancellation = $cancellation;
             }
 
-            $cancellationId = $readingCancellation->subscribe(static function () use ($socket) {
-                $socket->close();
-            });
+            $cancellationId = $readingCancellation->subscribe([$this->socket, 'close']);
 
             $completionDeferred->promise()->onResolve(static function () use ($readingCancellation, $cancellationId) {
                 $readingCancellation->unsubscribe($cancellationId);
             });
 
             try {
-                yield RequestWriter::writeRequest($socket, $request, $protocolVersion);
+                yield RequestWriter::writeRequest($this->socket, $request, $protocolVersion);
 
-                return yield from $this->doRead($socket, $request, $connectionInfo, $cancellation, $readingCancellation, $completionDeferred);
+                return yield from $this->doRead($request, $cancellation, $readingCancellation, $completionDeferred);
             } catch (HttpException $e) {
                 $cancellation->throwIfRequested();
 
@@ -98,9 +116,7 @@ private function buildRequest(Request $request): \Generator
     }
 
     /**
-     * @param Socket $socket
      * @param Request           $request
-     * @param ConnectionInfo    $connectionInfo
      * @param CancellationToken $originalCancellation
      * @param CancellationToken $readingCancellation
      * @param Deferred          $completionDeferred
@@ -111,9 +127,7 @@ private function buildRequest(Request $request): \Generator
      * @throws CancelledException
      */
     private function doRead(
-        Socket $socket,
         Request $request,
-        ConnectionInfo $connectionInfo,
         CancellationToken $originalCancellation,
         CancellationToken $readingCancellation,
         Deferred $completionDeferred
@@ -127,20 +141,18 @@ private function doRead(
                 $backpressure = $bodyEmitter->emit($data);
             };
 
-        $parser = new Http1Parser($request, $connectionInfo, $bodyCallback);
+        $parser = new Http1Parser($request, $this->connectionInfo, $bodyCallback);
 
         try {
-            while (null !== $chunk = yield $socket->read()) {
+            while (null !== $chunk = yield $this->socket->read()) {
                 $response = $parser->parse($chunk);
                 if ($response === null) {
                     continue;
                 }
 
                 $bodyCancellationSource = new CancellationTokenSource;
                 $bodyCancellationToken = new CombinedCancellationToken($readingCancellation, $bodyCancellationSource->getToken());
-                $bodyCancellationToken->subscribe(static function () use ($socket) {
-                    $socket->close();
-                });
+                $bodyCancellationToken->subscribe([$this->socket, 'close']);
 
                 $response = $response
                     ->withBody(new ResponseBodyStream(new IteratorStream($bodyEmitter->iterate()), $bodyCancellationSource))
@@ -156,7 +168,6 @@ private function doRead(
                     $originalCancellation,
                     $readingCancellation,
                     $bodyCancellationToken,
-                    $socket,
                     &$backpressure
                 ) {
                     try {
@@ -177,7 +188,7 @@ private function doRead(
                                 if (!$backpressure instanceof Success) {
                                     yield $this->withCancellation($backpressure, $bodyCancellationToken);
                                 }
-                            } while (null !== $chunk = yield $socket->read());
+                            } while (null !== $chunk = yield $this->socket->read());
 
                             $originalCancellation->throwIfRequested();
 
@@ -194,13 +205,13 @@ private function doRead(
                         }
 
                         if ($this->shouldCloseSocketAfterResponse($response) || $parser->getState() === Http1Parser::BODY_IDENTITY_EOF) {
-                            $socket->close();
+                            $this->socket->close();
                         }
 
                         $bodyEmitter->complete();
                         $completionDeferred->resolve();
                     } catch (\Throwable $e) {
-                        $socket->close();
+                        $this->socket->close();
 
                         $bodyEmitter->fail($e);
                         $completionDeferred->fail($e);
@@ -300,7 +311,6 @@ private function normalizeTraceRequest(Request $request): Request
         return $request;
     }
 
-
     private function shouldCloseSocketAfterResponse(Response $response): bool
     {
         $request = $response->getRequest();

src/Driver/Http1Parser.php → src/Connection/Http1Parser.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Amp\Http\Client\Driver;
+namespace Amp\Http\Client\Connection;
 
 use Amp\ByteStream\InMemoryStream;
 use Amp\Http\Client\ConnectionInfo;

src/ConnectionInfo.php

@@ -2,6 +2,8 @@
 
 namespace Amp\Http\Client;
 
+use Amp\Socket\EncryptableSocket;
+use Amp\Socket\Socket;
 use Amp\Socket\SocketAddress;
 use Amp\Socket\TlsInfo;
 
@@ -11,6 +13,15 @@ final class ConnectionInfo
     private $remoteAddress;
     private $tlsInfo;
 
+    public static function fromSocket(Socket $socket): self
+    {
+        return new self(
+            $socket->getLocalAddress(),
+            $socket->getRemoteAddress(),
+            $socket instanceof EncryptableSocket ? $socket->getTlsInfo() : null
+        );
+    }
+
     public function __construct(SocketAddress $localAddress, SocketAddress $remoteAddress, ?TlsInfo $tlsInfo = null)
     {
         $this->localAddress = $localAddress;