[HttpClient] Add support for amphp/http-client v5

Author: nicolas-grekas

AmpHttpClient.php

@@ -12,17 +12,20 @@
 namespace Symfony\Component\HttpClient;
 
 use Amp\CancelledException;
+use Amp\DeferredFuture;
 use Amp\Http\Client\DelegateHttpClient;
 use Amp\Http\Client\InterceptedHttpClient;
 use Amp\Http\Client\PooledHttpClient;
 use Amp\Http\Client\Request;
+use Amp\Http\HttpMessage;
 use Amp\Http\Tunnel\Http1TunnelConnector;
-use Amp\Promise;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
 use Symfony\Component\HttpClient\Exception\TransportException;
-use Symfony\Component\HttpClient\Internal\AmpClientState;
-use Symfony\Component\HttpClient\Response\AmpResponse;
+use Symfony\Component\HttpClient\Internal\AmpClientStateV4;
+use Symfony\Component\HttpClient\Internal\AmpClientStateV5;
+use Symfony\Component\HttpClient\Response\AmpResponseV4;
+use Symfony\Component\HttpClient\Response\AmpResponseV5;
 use Symfony\Component\HttpClient\Response\ResponseStream;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
 use Symfony\Contracts\HttpClient\ResponseInterface;
@@ -33,8 +36,8 @@
     throw new \LogicException('You cannot use "Symfony\Component\HttpClient\AmpHttpClient" as the "amphp/http-client" package is not installed. Try running "composer require amphp/http-client:^4.2.1".');
 }
 
-if (!interface_exists(Promise::class)) {
-    throw new \LogicException('You cannot use "Symfony\Component\HttpClient\AmpHttpClient" as the installed "amphp/http-client" is not compatible with this version of "symfony/http-client". Try downgrading "amphp/http-client" to "^4.2.1".');
+if (\PHP_VERSION_ID < 80400 && is_subclass_of(Request::class, HttpMessage::class)) {
+    throw new \LogicException('Using "Symfony\Component\HttpClient\AmpHttpClient" with amphp/http-client >= 5 requires PHP >= 8.4. Try running "composer require amphp/http-client:^4.2.1" or upgrade to PHP >= 8.4.');
 }
 
 /**
@@ -53,7 +56,7 @@ final class AmpHttpClient implements HttpClientInterface, LoggerAwareInterface,
 
     private array $defaultOptions = self::OPTIONS_DEFAULTS;
     private static array $emptyDefaults = self::OPTIONS_DEFAULTS;
-    private AmpClientState $multi;
+    private AmpClientStateV4|AmpClientStateV5 $multi;
 
     /**
      * @param array         $defaultOptions     Default requests' options
@@ -72,7 +75,11 @@ public function __construct(array $defaultOptions = [], ?callable $clientConfigu
             [, $this->defaultOptions] = self::prepareRequest(null, null, $defaultOptions, $this->defaultOptions);
         }
 
-        $this->multi = new AmpClientState($clientConfigurator, $maxHostConnections, $maxPendingPushes, $this->logger);
+        if (is_subclass_of(Request::class, HttpMessage::class)) {
+            $this->multi = new AmpClientStateV5($clientConfigurator, $maxHostConnections, $maxPendingPushes, $this->logger);
+        } else {
+            $this->multi = new AmpClientStateV4($clientConfigurator, $maxHostConnections, $maxPendingPushes, $this->logger);
+        }
     }
 
     /**
@@ -132,9 +139,10 @@ public function request(string $method, string $url, array $options = []): Respo
             $request->addHeader($h[0], $h[1]);
         }
 
-        $request->setTcpConnectTimeout(1000 * $options['timeout']);
-        $request->setTlsHandshakeTimeout(1000 * $options['timeout']);
-        $request->setTransferTimeout(1000 * $options['max_duration']);
+        $coef = $request instanceof HttpMessage ? 1 : 1000;
+        $request->setTcpConnectTimeout($coef * $options['timeout']);
+        $request->setTlsHandshakeTimeout($coef * $options['timeout']);
+        $request->setTransferTimeout($coef * $options['max_duration']);
         if (method_exists($request, 'setInactivityTimeout')) {
             $request->setInactivityTimeout(0);
         }
@@ -145,25 +153,37 @@ public function request(string $method, string $url, array $options = []): Respo
             $request->setHeader('Authorization', 'Basic '.base64_encode(implode(':', $auth)));
         }
 
-        return new AmpResponse($this->multi, $request, $options, $this->logger);
+        if ($request instanceof HttpMessage) {
+            return new AmpResponseV5($this->multi, $request, $options, $this->logger);
+        }
+
+        return new AmpResponseV4($this->multi, $request, $options, $this->logger);
     }
 
     public function stream(ResponseInterface|iterable $responses, ?float $timeout = null): ResponseStreamInterface
     {
-        if ($responses instanceof AmpResponse) {
+        if ($responses instanceof AmpResponseV4 || $responses instanceof AmpResponseV5) {
             $responses = [$responses];
         }
 
-        return new ResponseStream(AmpResponse::stream($responses, $timeout));
+        if ($this->multi instanceof AmpClientStateV5) {
+            return new ResponseStream(AmpResponseV5::stream($responses, $timeout));
+        }
+
+        return new ResponseStream(AmpResponseV4::stream($responses, $timeout));
     }
 
     public function reset(): void
     {
         $this->multi->dnsCache = [];
 
-        foreach ($this->multi->pushedResponses as $authority => $pushedResponses) {
+        foreach ($this->multi->pushedResponses as $pushedResponses) {
             foreach ($pushedResponses as [$pushedUrl, $pushDeferred]) {
-                $pushDeferred->fail(new CancelledException());
+                if ($pushDeferred instanceof DeferredFuture) {
+                    $pushDeferred->error(new CancelledException());
+                } else {
+                    $pushDeferred->fail(new CancelledException());
+                }
 
                 $this->logger?->debug(\sprintf('Unused pushed response: "%s"', $pushedUrl));
             }

CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.2
+---
+
+ * Add support for amphp/http-client v5 on PHP 8.4+
+
 7.1
 ---
 

HttpClient.php

@@ -11,8 +11,8 @@
 
 namespace Symfony\Component\HttpClient;
 
-use Amp\Http\Client\Connection\ConnectionLimitingPool;
-use Amp\Promise;
+use Amp\Http\Client\Request as AmpRequest;
+use Amp\Http\HttpMessage;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
 
 /**
@@ -31,7 +31,7 @@ final class HttpClient
      */
     public static function create(array $defaultOptions = [], int $maxHostConnections = 6, int $maxPendingPushes = 50): HttpClientInterface
     {
-        if ($amp = class_exists(ConnectionLimitingPool::class) && interface_exists(Promise::class)) {
+        if ($amp = class_exists(AmpRequest::class) && (\PHP_VERSION_ID >= 80400 || is_subclass_of(AmpRequest::class, HttpMessage::class))) {
             if (!\extension_loaded('curl')) {
                 return new AmpHttpClient($defaultOptions, null, $maxHostConnections, $maxPendingPushes);
             }

Internal/AmpBody.php renamed to Internal/AmpBodyV4.php

@@ -23,7 +23,7 @@
  *
  * @internal
  */
-class AmpBody implements RequestBody, InputStream
+class AmpBodyV4 implements RequestBody, InputStream
 {
     private ResourceInputStream|\Closure|string $body;
     private array $info;

Internal/AmpBodyV5.php

@@ -0,0 +1,150 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\HttpClient\Internal;
+
+use Amp\ByteStream\ReadableBuffer;
+use Amp\ByteStream\ReadableIterableStream;
+use Amp\ByteStream\ReadableResourceStream;
+use Amp\ByteStream\ReadableStream;
+use Amp\Cancellation;
+use Amp\Http\Client\HttpContent;
+use Symfony\Component\HttpClient\Exception\TransportException;
+
+/**
+ * @author Nicolas Grekas <[email protected]>
+ *
+ * @internal
+ */
+class AmpBodyV5 implements HttpContent, ReadableStream, \IteratorAggregate
+{
+    private ReadableStream $body;
+    private ?string $content;
+    private array $info;
+    private ?int $offset = 0;
+    private int $length = -1;
+    private ?int $uploaded = null;
+
+    /**
+     * @param \Closure|resource|string $body
+     */
+    public function __construct(
+        $body,
+        &$info,
+        private \Closure $onProgress,
+    ) {
+        $this->info = &$info;
+
+        if (\is_resource($body)) {
+            $this->offset = ftell($body);
+            $this->length = fstat($body)['size'];
+            $this->body = new ReadableResourceStream($body);
+        } elseif (\is_string($body)) {
+            $this->length = \strlen($body);
+            $this->body = new ReadableBuffer($body);
+            $this->content = $body;
+        } else {
+            $this->body = new ReadableIterableStream((static function () use ($body) {
+                while ('' !== $data = ($body)(16372)) {
+                    if (!\is_string($data)) {
+                        throw new TransportException(\sprintf('Return value of the "body" option callback must be string, "%s" returned.', get_debug_type($data)));
+                    }
+
+                    yield $data;
+                }
+            })());
+        }
+    }
+
+    public function getContent(): ReadableStream
+    {
+        if (null !== $this->uploaded) {
+            $this->uploaded = null;
+
+            if (\is_string($this->body)) {
+                $this->offset = 0;
+            } elseif ($this->body instanceof ReadableResourceStream) {
+                fseek($this->body->getResource(), $this->offset);
+            }
+        }
+
+        return $this;
+    }
+
+    public function getContentType(): ?string
+    {
+        return null;
+    }
+
+    public function getContentLength(): ?int
+    {
+        return 0 <= $this->length ? $this->length - $this->offset : null;
+    }
+
+    public function read(?Cancellation $cancellation = null): ?string
+    {
+        $this->info['size_upload'] += $this->uploaded;
+        $this->uploaded = 0;
+        ($this->onProgress)();
+
+        if (null !== $data = $this->body->read($cancellation)) {
+            $this->uploaded = \strlen($data);
+        } else {
+            $this->info['upload_content_length'] = $this->info['size_upload'];
+        }
+
+        return $data;
+    }
+
+    public function isReadable(): bool
+    {
+        return $this->body->isReadable();
+    }
+
+    public function close(): void
+    {
+        $this->body->close();
+    }
+
+    public function isClosed(): bool
+    {
+        return $this->body->isClosed();
+    }
+
+    public function onClose(\Closure $onClose): void
+    {
+        $this->body->onClose($onClose);
+    }
+
+    public function getIterator(): \Traversable
+    {
+        return $this->body;
+    }
+
+    public static function rewind(HttpContent $body): HttpContent
+    {
+        if (!$body instanceof self) {
+            return $body;
+        }
+
+        $body->uploaded = null;
+
+        if ($body->body instanceof ReadableResourceStream && !$body->body->isClosed()) {
+            fseek($body->body->getResource(), $body->offset);
+        }
+
+        if ($body->body instanceof ReadableBuffer) {
+            return new $body($body->content, $body->info, $body->onProgress);
+        }
+
+        return $body;
+    }
+}

Internal/AmpClientState.php renamed to Internal/AmpClientStateV4.php

@@ -39,7 +39,7 @@
  *
  * @internal
  */
-final class AmpClientState extends ClientState
+final class AmpClientStateV4 extends ClientState
 {
     public array $dnsCache = [];
     public int $responseCount = 0;
@@ -90,7 +90,7 @@ public function request(array $options, Request $request, CancellationToken $can
             $info['peer_certificate_chain'] = [];
         }
 
-        $request->addEventListener(new AmpListener($info, $options['peer_fingerprint']['pin-sha256'] ?? [], $onProgress, $handle));
+        $request->addEventListener(new AmpListenerV4($info, $options['peer_fingerprint']['pin-sha256'] ?? [], $onProgress, $handle));
         $request->setPushHandler(fn ($request, $response): Promise => $this->handlePush($request, $response, $options));
 
         ($request->hasHeader('content-length') ? new Success((int) $request->getHeader('content-length')) : $request->getBody()->getBodyLength())
@@ -157,7 +157,7 @@ public function connect(string $uri, ?ConnectContext $context = null, ?Cancellat
                 return $result;
             }
         };
-        $connector->connector = new DnsConnector(new AmpResolver($this->dnsCache));
+        $connector->connector = new DnsConnector(new AmpResolverV4($this->dnsCache));
 
         $context = (new ConnectContext())
             ->withTcpNoDelay()