symfony · nicolas-grekas · Sep 3, 2021
diff --git a/src/Symfony/Component/HttpClient/Internal/ClientState.php b/src/Symfony/Component/HttpClient/Internal/ClientState.php
@@ -22,4 +22,5 @@ class ClientState
 {
     public $handlesActivity = [];
     public $openHandles = [];
+    public $lastTimeout;
 }
diff --git a/src/Symfony/Component/HttpClient/Response/AmpResponse.php b/src/Symfony/Component/HttpClient/Response/AmpResponse.php
@@ -45,7 +45,6 @@ final class AmpResponse implements ResponseInterface, StreamableInterface
 
     private static $nextId = 'a';
 
-    private $multi;
     private $options;
     private $canceller;
     private $onProgress;

diff --git a/src/Symfony/Component/HttpClient/Response/AsyncResponse.php b/src/Symfony/Component/HttpClient/Response/AsyncResponse.php
@@ -57,13 +57,13 @@ public function __construct(HttpClientInterface $client, string $method, string
         }
         $this->response = $client->request($method, $url, ['buffer' => false] + $options);
         $this->passthru = $passthru;
-        $this->initializer = static function (self $response) {
+        $this->initializer = static function (self $response, float $timeout = null) {
             if (null === $response->shouldBuffer) {
                 return false;
             }
 
             while (true) {
-                foreach (self::stream([$response]) as $chunk) {
+                foreach (self::stream([$response], $timeout) as $chunk) {
                     if ($chunk->isTimeout() && $response->passthru) {
                         foreach (self::passthru($response->client, $response, new ErrorChunk($response->offset, new TransportException($chunk->getError()))) as $chunk) {
                             if ($chunk->isFirst()) {
@@ -179,6 +179,7 @@ public function __destruct()
 
         if ($this->initializer && null === $this->getInfo('error')) {
             try {
+                self::initialize($this, -0.0);
                 $this->getHeaders(true);
             } catch (HttpExceptionInterface $httpException) {
                 // no-op

diff --git a/src/Symfony/Component/HttpClient/Response/CommonResponseTrait.php b/src/Symfony/Component/HttpClient/Response/CommonResponseTrait.php
@@ -145,15 +145,15 @@ public function __wakeup()
      */
     abstract protected function close(): void;
 
-    private static function initialize(self $response): void
+    private static function initialize(self $response, float $timeout = null): void
     {
         if (null !== $response->getInfo('error')) {
             throw new TransportException($response->getInfo('error'));
         }
 
         try {
-            if (($response->initializer)($response)) {
-                foreach (self::stream([$response]) as $chunk) {
+            if (($response->initializer)($response, $timeout)) {
+                foreach (self::stream([$response], $timeout) as $chunk) {
                     if ($chunk->isFirst()) {
                         break;
                     }

diff --git a/src/Symfony/Component/HttpClient/Response/CurlResponse.php b/src/Symfony/Component/HttpClient/Response/CurlResponse.php
@@ -33,7 +33,6 @@ final class CurlResponse implements ResponseInterface, StreamableInterface
     use TransportResponseTrait;
 
     private static $performing = false;
-    private $multi;
     private $debugBuffer;
 
     /**

diff --git a/src/Symfony/Component/HttpClient/Response/NativeResponse.php b/src/Symfony/Component/HttpClient/Response/NativeResponse.php
@@ -35,7 +35,6 @@ final class NativeResponse implements ResponseInterface, StreamableInterface
     private $onProgress;
     private $remaining;
     private $buffer;
-    private $multi;
     private $debugBuffer;
     private $shouldBuffer;
     private $pauseExpiry = 0;

diff --git a/src/Symfony/Component/HttpClient/Response/TransportResponseTrait.php b/src/Symfony/Component/HttpClient/Response/TransportResponseTrait.php
@@ -43,6 +43,8 @@ trait TransportResponseTrait
     private $finalInfo;
     private $canary;
     private $logger;
+    /** @var ClientState|null */
+    private $multi = null;
 
     /**
      * {@inheritdoc}
@@ -138,7 +140,7 @@ private function doDestruct()
         $this->shouldBuffer = true;
 
         if ($this->initializer && null === $this->info['error']) {
-            self::initialize($this);
+            self::initialize($this, -0.0);
             $this->checkStatusCode();
         }
     }
@@ -159,6 +161,12 @@ public static function stream(iterable $responses, float $timeout = null): \Gene
         $lastActivity = microtime(true);
         $elapsedTimeout = 0;
 
+        if ($fromLastTimeout = 0.0 === $timeout && '-0' === (string) $timeout) {
+            $timeout = null;
+        } elseif ($fromLastTimeout = 0 > $timeout) {
+            $timeout = -$timeout;
+        }
+
         while (true) {
             $hasActivity = false;
             $timeoutMax = 0;
@@ -172,15 +180,21 @@ public static function stream(iterable $responses, float $timeout = null): \Gene
                 foreach ($responses as $j => $response) {
                     $timeoutMax = $timeout ?? max($timeoutMax, $response->timeout);
                     $timeoutMin = min($timeoutMin, $response->timeout, 1);
+
+                    if ($fromLastTimeout && null !== $multi->lastTimeout) {
+                        $elapsedTimeout = microtime(true) - $multi->lastTimeout;
+                    }
+
                     $chunk = false;
 
                     if (isset($multi->handlesActivity[$j])) {
-                        // no-op
+                        $multi->lastTimeout = null;
                     } elseif (!isset($multi->openHandles[$j])) {
                         unset($responses[$j]);
                         continue;
                     } elseif ($elapsedTimeout >= $timeoutMax) {
                         $multi->handlesActivity[$j] = [new ErrorChunk($response->offset, sprintf('Idle timeout reached for "%s".', $response->getInfo('url')))];
+                        $multi->lastTimeout ?? $multi->lastTimeout = $lastActivity;
                     } else {
                         continue;
                     }

diff --git a/src/Symfony/Component/HttpClient/Tests/AsyncDecoratorTraitTest.php b/src/Symfony/Component/HttpClient/Tests/AsyncDecoratorTraitTest.php
@@ -12,7 +12,9 @@
 namespace Symfony\Component\HttpClient\Tests;
 
 use Symfony\Component\HttpClient\AsyncDecoratorTrait;
+use Symfony\Component\HttpClient\CurlHttpClient;
 use Symfony\Component\HttpClient\DecoratorTrait;
+use Symfony\Component\HttpClient\HttpClient;
 use Symfony\Component\HttpClient\Response\AsyncContext;
 use Symfony\Component\HttpClient\Response\AsyncResponse;
 use Symfony\Contracts\HttpClient\ChunkInterface;
@@ -29,6 +31,10 @@ protected function getHttpClient(string $testCase, \Closure $chunkFilter = null,
             $this->markTestSkipped("AsyncDecoratorTrait doesn't cache handles");
         }
 
+        if ('testTimeoutOnDestruct' === $testCase) {
+            return new CurlHttpClient();
+        }
+
         $chunkFilter = $chunkFilter ?? static function (ChunkInterface $chunk, AsyncContext $context) { yield $chunk; };
 
         return new class($decoratedClient ?? parent::getHttpClient($testCase), $chunkFilter) implements HttpClientInterface {
@@ -49,6 +55,15 @@ public function request(string $method, string $url, array $options = []): Respo
         };
     }
 
+    public function testTimeoutOnDestruct()
+    {
+        if (HttpClient::create() instanceof NativeHttpClient) {
+            parent::testTimeoutOnDestruct();
+        } else {
+            HttpClientTestCase::testTimeoutOnDestruct();
+        }
+    }
+
     public function testRetry404()
     {
         $client = $this->getHttpClient(__FUNCTION__, function (ChunkInterface $chunk, AsyncContext $context) {

diff --git a/src/Symfony/Component/HttpClient/Tests/HttpClientTestCase.php b/src/Symfony/Component/HttpClient/Tests/HttpClientTestCase.php
@@ -32,6 +32,15 @@ abstract class HttpClientTestCase extends BaseHttpClientTestCase
 {
     private static $vulcainStarted = false;
 
+    public function testTimeoutOnDestruct()
+    {
+        if (!method_exists(parent::class, 'testTimeoutOnDestruct')) {
+            $this->markTestSkipped('BaseHttpClientTestCase doesn\'t have testTimeoutOnDestruct().');
+        }
+
+        parent::testTimeoutOnDestruct();
+    }
+
     public function testAcceptHeader()
     {
         $client = $this->getHttpClient(__FUNCTION__);

diff --git a/src/Symfony/Component/HttpClient/Tests/MockHttpClientTest.php b/src/Symfony/Component/HttpClient/Tests/MockHttpClientTest.php
@@ -236,6 +236,10 @@ protected function getHttpClient(string $testCase): HttpClientInterface
                 $this->markTestSkipped('Real transport required');
                 break;
 
+            case 'testTimeoutOnDestruct':
+                $this->markTestSkipped('Real transport required');
+                break;
+
             case 'testDestruct':
                 $this->markTestSkipped("MockHttpClient doesn't timeout on destruct");
                 break;

diff --git a/src/Symfony/Component/HttpClient/Tests/NativeHttpClientTest.php b/src/Symfony/Component/HttpClient/Tests/NativeHttpClientTest.php
@@ -26,6 +26,11 @@ public function testInformationalResponseStream()
         $this->markTestSkipped('NativeHttpClient doesn\'t support informational status codes.');
     }
 
+    public function testTimeoutOnDestruct()
+    {
+        $this->markTestSkipped('NativeHttpClient doesn\'t support opening concurrent requests.');
+    }
+
     public function testHttp2PushVulcain()
     {
         $this->markTestSkipped('NativeHttpClient doesn\'t support HTTP/2.');

diff --git a/src/Symfony/Contracts/HttpClient/Test/HttpClientTestCase.php b/src/Symfony/Contracts/HttpClient/Test/HttpClientTestCase.php
@@ -835,6 +835,38 @@ public function testTimeoutWithActiveConcurrentStream()
         }
     }
 
+    public function testTimeoutOnDestruct()
+    {
+        $p1 = TestHttpServer::start(8067);
+        $p2 = TestHttpServer::start(8077);
+
+        $client = $this->getHttpClient(__FUNCTION__);
+        $start = microtime(true);
+        $responses = [];
+
+        $responses[] = $client->request('GET', 'http://localhost:8067/timeout-header', ['timeout' => 0.25]);
+        $responses[] = $client->request('GET', 'http://localhost:8077/timeout-header', ['timeout' => 0.25]);
+        $responses[] = $client->request('GET', 'http://localhost:8067/timeout-header', ['timeout' => 0.25]);
+        $responses[] = $client->request('GET', 'http://localhost:8077/timeout-header', ['timeout' => 0.25]);
+
+        try {
+            while ($response = array_shift($responses)) {
+                try {
+                    unset($response);
+                    $this->fail(TransportExceptionInterface::class.' expected');
+                } catch (TransportExceptionInterface $e) {
+                }
+            }
+
+            $duration = microtime(true) - $start;
+
+            $this->assertLessThan(0.75, $duration);
+        } finally {
+            $p1->stop();
+            $p2->stop();
+        }
+    }
+
     public function testDestruct()
     {
         $client = $this->getHttpClient(__FUNCTION__);