vendor/symfony/http-client/Internal/HttplugWaitLoop.php line 49

Open in your IDE?
  1. <?php
  3. /*
  4.  * This file is part of the Symfony package.
  5.  *
  6.  * (c) Fabien Potencier <fabien@symfony.com>
  7.  *
  8.  * For the full copyright and license information, please view the LICENSE
  9.  * file that was distributed with this source code.
  10.  */
  12. namespace Symfony\Component\HttpClient\Internal;
  14. use Http\Client\Exception\NetworkException;
  15. use Http\Promise\Promise;
  16. use Psr\Http\Message\RequestInterface as Psr7RequestInterface;
  17. use Psr\Http\Message\ResponseFactoryInterface;
  18. use Psr\Http\Message\ResponseInterface as Psr7ResponseInterface;
  19. use Psr\Http\Message\StreamFactoryInterface;
  20. use Symfony\Component\HttpClient\Response\StreamableInterface;
  21. use Symfony\Component\HttpClient\Response\StreamWrapper;
  22. use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
  23. use Symfony\Contracts\HttpClient\HttpClientInterface;
  24. use Symfony\Contracts\HttpClient\ResponseInterface;
  26. /**
  27.  * @author Nicolas Grekas <p@tchwork.com>
  28.  *
  29.  * @internal
  30.  */
  31. final class HttplugWaitLoop
  32. {
  33.     private $client;
  34.     private $promisePool;
  35.     private $responseFactory;
  36.     private $streamFactory;
  38.     /**
  39.      * @param \SplObjectStorage<ResponseInterface, array{Psr7RequestInterface, Promise}>|null $promisePool
  40.      */
  41.     public function __construct(HttpClientInterface $client, ?\SplObjectStorage $promisePool, ResponseFactoryInterface $responseFactory, StreamFactoryInterface $streamFactory)
  42.     {
  43.         $this->client = $client;
  44.         $this->promisePool = $promisePool;
  45.         $this->responseFactory = $responseFactory;
  46.         $this->streamFactory = $streamFactory;
  47.     }
  49.     public function wait(?ResponseInterface $pendingResponse, float $maxDuration = null, float $idleTimeout = null): int
  50.     {
  51.         if (!$this->promisePool) {
  52.             return 0;
  53.         }
  55.         $guzzleQueue = \GuzzleHttp\Promise\Utils::queue();
  57.         if (0.0 === $remainingDuration = $maxDuration) {
  58.             $idleTimeout = 0.0;
  59.         } elseif (null !== $maxDuration) {
  60.             $startTime = microtime(true);
  61.             $idleTimeout = max(0.0, min($maxDuration / 5, $idleTimeout ?? $maxDuration));
  62.         }
  64.         do {
  65.             foreach ($this->client->stream($this->promisePool, $idleTimeout) as $response => $chunk) {
  66.                 try {
  67.                     if (null !== $maxDuration && $chunk->isTimeout()) {
  68.                         goto check_duration;
  69.                     }
  71.                     if ($chunk->isFirst()) {
  72.                         // Deactivate throwing on 3/4/5xx
  73.                         $response->getStatusCode();
  74.                     }
  76.                     if (!$chunk->isLast()) {
  77.                         goto check_duration;
  78.                     }
  80.                     if ([, $promise] = $this->promisePool[$response] ?? null) {
  81.                         unset($this->promisePool[$response]);
  82.                         $promise->resolve($this->createPsr7Response($response, true));
  83.                     }
  84.                 } catch (\Exception $e) {
  85.                     if ([$request, $promise] = $this->promisePool[$response] ?? null) {
  86.                         unset($this->promisePool[$response]);
  88.                         if ($e instanceof TransportExceptionInterface) {
  89.                             $e = new NetworkException($e->getMessage(), $request, $e);
  90.                         }
  92.                         $promise->reject($e);
  93.                     }
  94.                 }
  96.                 $guzzleQueue->run();
  98.                 if ($pendingResponse === $response) {
  99.                     return $this->promisePool->count();
  100.                 }
  102.                 check_duration:
  103.                 if (null !== $maxDuration && $idleTimeout && $idleTimeout > $remainingDuration = max(0.0, $maxDuration - microtime(true) + $startTime)) {
  104.                     $idleTimeout = $remainingDuration / 5;
  105.                     break;
  106.                 }
  107.             }
  109.             if (!$count = $this->promisePool->count()) {
  110.                 return 0;
  111.             }
  112.         } while (null === $maxDuration || 0 < $remainingDuration);
  114.         return $count;
  115.     }
  117.     public function createPsr7Response(ResponseInterface $response, bool $buffer = false): Psr7ResponseInterface
  118.     {
  119.         $psrResponse = $this->responseFactory->createResponse($response->getStatusCode());
  121.         foreach ($response->getHeaders(false) as $name => $values) {
  122.             foreach ($values as $value) {
  123.                 $psrResponse = $psrResponse->withAddedHeader($name, $value);
  124.             }
  125.         }
  127.         if ($response instanceof StreamableInterface) {
  128.             $body = $this->streamFactory->createStreamFromResource($response->toStream(false));
  129.         } elseif (!$buffer) {
  130.             $body = $this->streamFactory->createStreamFromResource(StreamWrapper::createResource($response, $this->client));
  131.         } else {
  132.             $body = $this->streamFactory->createStream($response->getContent(false));
  133.         }
  135.         if ($body->isSeekable()) {
  136.             $body->seek(0);
  137.         }
  139.         return $psrResponse->withBody($body);
  140.     }
  141. }