[HttpClient] resolve promise chains on HttplugClient::wait()
This commit is contained in:
parent
6e7f3257ce
commit
ea0be07a33
@ -32,9 +32,8 @@ use Psr\Http\Message\StreamFactoryInterface;
|
|||||||
use Psr\Http\Message\StreamInterface;
|
use Psr\Http\Message\StreamInterface;
|
||||||
use Psr\Http\Message\UriFactoryInterface;
|
use Psr\Http\Message\UriFactoryInterface;
|
||||||
use Psr\Http\Message\UriInterface;
|
use Psr\Http\Message\UriInterface;
|
||||||
|
use Symfony\Component\HttpClient\Internal\HttplugWaitLoop;
|
||||||
use Symfony\Component\HttpClient\Response\HttplugPromise;
|
use Symfony\Component\HttpClient\Response\HttplugPromise;
|
||||||
use Symfony\Component\HttpClient\Response\ResponseTrait;
|
|
||||||
use Symfony\Component\HttpClient\Response\StreamWrapper;
|
|
||||||
use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
|
use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
|
||||||
use Symfony\Contracts\HttpClient\HttpClientInterface;
|
use Symfony\Contracts\HttpClient\HttpClientInterface;
|
||||||
use Symfony\Contracts\HttpClient\ResponseInterface;
|
use Symfony\Contracts\HttpClient\ResponseInterface;
|
||||||
@ -60,27 +59,27 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
private $client;
|
private $client;
|
||||||
private $responseFactory;
|
private $responseFactory;
|
||||||
private $streamFactory;
|
private $streamFactory;
|
||||||
private $promisePool = [];
|
private $promisePool;
|
||||||
private $pendingResponse;
|
private $waitLoop;
|
||||||
|
|
||||||
public function __construct(HttpClientInterface $client = null, ResponseFactoryInterface $responseFactory = null, StreamFactoryInterface $streamFactory = null)
|
public function __construct(HttpClientInterface $client = null, ResponseFactoryInterface $responseFactory = null, StreamFactoryInterface $streamFactory = null)
|
||||||
{
|
{
|
||||||
$this->client = $client ?? HttpClient::create();
|
$this->client = $client ?? HttpClient::create();
|
||||||
$this->responseFactory = $responseFactory;
|
$this->responseFactory = $responseFactory;
|
||||||
$this->streamFactory = $streamFactory ?? ($responseFactory instanceof StreamFactoryInterface ? $responseFactory : null);
|
$this->streamFactory = $streamFactory ?? ($responseFactory instanceof StreamFactoryInterface ? $responseFactory : null);
|
||||||
$this->promisePool = new \SplObjectStorage();
|
$this->promisePool = \function_exists('GuzzleHttp\Promise\queue') ? new \SplObjectStorage() : null;
|
||||||
|
|
||||||
if (null !== $this->responseFactory && null !== $this->streamFactory) {
|
if (null === $this->responseFactory || null === $this->streamFactory) {
|
||||||
return;
|
if (!class_exists(Psr17Factory::class)) {
|
||||||
|
throw new \LogicException('You cannot use the "Symfony\Component\HttpClient\HttplugClient" as no PSR-17 factories have been provided. Try running "composer require nyholm/psr7".');
|
||||||
|
}
|
||||||
|
|
||||||
|
$psr17Factory = new Psr17Factory();
|
||||||
|
$this->responseFactory = $this->responseFactory ?? $psr17Factory;
|
||||||
|
$this->streamFactory = $this->streamFactory ?? $psr17Factory;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!class_exists(Psr17Factory::class)) {
|
$this->waitLoop = new HttplugWaitLoop($this->client, $this->promisePool, $this->responseFactory, $this->streamFactory);
|
||||||
throw new \LogicException('You cannot use the "Symfony\Component\HttpClient\HttplugClient" as no PSR-17 factories have been provided. Try running "composer require nyholm/psr7".');
|
|
||||||
}
|
|
||||||
|
|
||||||
$psr17Factory = new Psr17Factory();
|
|
||||||
$this->responseFactory = $this->responseFactory ?? $psr17Factory;
|
|
||||||
$this->streamFactory = $this->streamFactory ?? $psr17Factory;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -89,7 +88,7 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
public function sendRequest(RequestInterface $request): Psr7ResponseInterface
|
public function sendRequest(RequestInterface $request): Psr7ResponseInterface
|
||||||
{
|
{
|
||||||
try {
|
try {
|
||||||
return $this->createPsr7Response($this->sendPsr7Request($request));
|
return $this->waitLoop->createPsr7Response($this->sendPsr7Request($request));
|
||||||
} catch (TransportExceptionInterface $e) {
|
} catch (TransportExceptionInterface $e) {
|
||||||
throw new NetworkException($e->getMessage(), $request, $e);
|
throw new NetworkException($e->getMessage(), $request, $e);
|
||||||
}
|
}
|
||||||
@ -102,7 +101,7 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
*/
|
*/
|
||||||
public function sendAsyncRequest(RequestInterface $request): Promise
|
public function sendAsyncRequest(RequestInterface $request): Promise
|
||||||
{
|
{
|
||||||
if (!class_exists(GuzzlePromise::class)) {
|
if (!$promisePool = $this->promisePool) {
|
||||||
throw new \LogicException(sprintf('You cannot use "%s()" as the "guzzlehttp/promises" package is not installed. Try running "composer require guzzlehttp/promises".', __METHOD__));
|
throw new \LogicException(sprintf('You cannot use "%s()" as the "guzzlehttp/promises" package is not installed. Try running "composer require guzzlehttp/promises".', __METHOD__));
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -112,23 +111,22 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
return new RejectedPromise($e);
|
return new RejectedPromise($e);
|
||||||
}
|
}
|
||||||
|
|
||||||
$cancel = function () use ($response) {
|
$waitLoop = $this->waitLoop;
|
||||||
|
|
||||||
|
$promise = new GuzzlePromise(static function () use ($response, $waitLoop) {
|
||||||
|
$waitLoop->wait($response);
|
||||||
|
}, static function () use ($response, $promisePool) {
|
||||||
$response->cancel();
|
$response->cancel();
|
||||||
unset($this->promisePool[$response]);
|
unset($promisePool[$response]);
|
||||||
};
|
});
|
||||||
|
|
||||||
$promise = new GuzzlePromise(function () use ($response) {
|
$promisePool[$response] = [$request, $promise];
|
||||||
$this->pendingResponse = $response;
|
|
||||||
$this->wait();
|
|
||||||
}, $cancel);
|
|
||||||
|
|
||||||
$this->promisePool[$response] = [$request, $promise];
|
return new HttplugPromise($promise);
|
||||||
|
|
||||||
return new HttplugPromise($promise, $cancel);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Resolve pending promises that complete before the timeouts are reached.
|
* Resolves pending promises that complete before the timeouts are reached.
|
||||||
*
|
*
|
||||||
* When $maxDuration is null and $idleTimeout is reached, promises are rejected.
|
* When $maxDuration is null and $idleTimeout is reached, promises are rejected.
|
||||||
*
|
*
|
||||||
@ -136,64 +134,7 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
*/
|
*/
|
||||||
public function wait(float $maxDuration = null, float $idleTimeout = null): int
|
public function wait(float $maxDuration = null, float $idleTimeout = null): int
|
||||||
{
|
{
|
||||||
$pendingResponse = $this->pendingResponse;
|
return $this->waitLoop->wait(null, $maxDuration, $idleTimeout);
|
||||||
$this->pendingResponse = null;
|
|
||||||
|
|
||||||
if (null !== $maxDuration) {
|
|
||||||
$startTime = microtime(true);
|
|
||||||
$idleTimeout = max(0.0, min($maxDuration / 5, $idleTimeout ?? $maxDuration));
|
|
||||||
$remainingDuration = $maxDuration;
|
|
||||||
}
|
|
||||||
|
|
||||||
do {
|
|
||||||
foreach ($this->client->stream($this->promisePool, $idleTimeout) as $response => $chunk) {
|
|
||||||
try {
|
|
||||||
if (null !== $maxDuration && $chunk->isTimeout()) {
|
|
||||||
goto check_duration;
|
|
||||||
}
|
|
||||||
|
|
||||||
if ($chunk->isFirst()) {
|
|
||||||
// Deactivate throwing on 3/4/5xx
|
|
||||||
$response->getStatusCode();
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!$chunk->isLast()) {
|
|
||||||
goto check_duration;
|
|
||||||
}
|
|
||||||
|
|
||||||
if ([$request, $promise] = $this->promisePool[$response] ?? null) {
|
|
||||||
unset($this->promisePool[$response]);
|
|
||||||
$promise->resolve($this->createPsr7Response($response, true));
|
|
||||||
}
|
|
||||||
} catch (\Exception $e) {
|
|
||||||
if ([$request, $promise] = $this->promisePool[$response] ?? null) {
|
|
||||||
unset($this->promisePool[$response]);
|
|
||||||
|
|
||||||
if ($e instanceof TransportExceptionInterface) {
|
|
||||||
$e = new NetworkException($e->getMessage(), $request, $e);
|
|
||||||
}
|
|
||||||
|
|
||||||
$promise->reject($e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if ($pendingResponse === $response) {
|
|
||||||
return \count($this->promisePool);
|
|
||||||
}
|
|
||||||
|
|
||||||
check_duration:
|
|
||||||
if (null !== $maxDuration && $idleTimeout && $idleTimeout > $remainingDuration = max(0.0, $maxDuration - microtime(true) + $startTime)) {
|
|
||||||
$idleTimeout = $remainingDuration / 5;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!$count = \count($this->promisePool)) {
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
} while (null !== $maxDuration && 0 < $remainingDuration);
|
|
||||||
|
|
||||||
return $count;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -265,6 +206,11 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
return new Uri($uri);
|
return new Uri($uri);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public function __destruct()
|
||||||
|
{
|
||||||
|
$this->wait();
|
||||||
|
}
|
||||||
|
|
||||||
private function sendPsr7Request(RequestInterface $request, bool $buffer = null): ResponseInterface
|
private function sendPsr7Request(RequestInterface $request, bool $buffer = null): ResponseInterface
|
||||||
{
|
{
|
||||||
try {
|
try {
|
||||||
@ -286,29 +232,4 @@ final class HttplugClient implements HttplugInterface, HttpAsyncClient, RequestF
|
|||||||
throw new NetworkException($e->getMessage(), $request, $e);
|
throw new NetworkException($e->getMessage(), $request, $e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private function createPsr7Response(ResponseInterface $response, bool $buffer = false): Psr7ResponseInterface
|
|
||||||
{
|
|
||||||
$psrResponse = $this->responseFactory->createResponse($response->getStatusCode());
|
|
||||||
|
|
||||||
foreach ($response->getHeaders(false) as $name => $values) {
|
|
||||||
foreach ($values as $value) {
|
|
||||||
$psrResponse = $psrResponse->withAddedHeader($name, $value);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (isset(class_uses($response)[ResponseTrait::class])) {
|
|
||||||
$body = $this->streamFactory->createStreamFromResource($response->toStream(false));
|
|
||||||
} elseif (!$buffer) {
|
|
||||||
$body = $this->streamFactory->createStreamFromResource(StreamWrapper::createResource($response, $this->client));
|
|
||||||
} else {
|
|
||||||
$body = $this->streamFactory->createStream($response->getContent(false));
|
|
||||||
}
|
|
||||||
|
|
||||||
if ($body->isSeekable()) {
|
|
||||||
$body->seek(0);
|
|
||||||
}
|
|
||||||
|
|
||||||
return $psrResponse->withBody($body);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
136
src/Symfony/Component/HttpClient/Internal/HttplugWaitLoop.php
Normal file
136
src/Symfony/Component/HttpClient/Internal/HttplugWaitLoop.php
Normal file
@ -0,0 +1,136 @@
|
|||||||
|
<?php
|
||||||
|
|
||||||
|
/*
|
||||||
|
* This file is part of the Symfony package.
|
||||||
|
*
|
||||||
|
* (c) Fabien Potencier <fabien@symfony.com>
|
||||||
|
*
|
||||||
|
* 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 Http\Client\Exception\NetworkException;
|
||||||
|
use Psr\Http\Message\ResponseFactoryInterface;
|
||||||
|
use Psr\Http\Message\ResponseInterface as Psr7ResponseInterface;
|
||||||
|
use Psr\Http\Message\StreamFactoryInterface;
|
||||||
|
use Symfony\Component\HttpClient\Response\ResponseTrait;
|
||||||
|
use Symfony\Component\HttpClient\Response\StreamWrapper;
|
||||||
|
use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
|
||||||
|
use Symfony\Contracts\HttpClient\HttpClientInterface;
|
||||||
|
use Symfony\Contracts\HttpClient\ResponseInterface;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author Nicolas Grekas <p@tchwork.com>
|
||||||
|
*
|
||||||
|
* @internal
|
||||||
|
*/
|
||||||
|
final class HttplugWaitLoop
|
||||||
|
{
|
||||||
|
private $client;
|
||||||
|
private $promisePool;
|
||||||
|
private $responseFactory;
|
||||||
|
private $streamFactory;
|
||||||
|
|
||||||
|
public function __construct(HttpClientInterface $client, ?\SplObjectStorage $promisePool, ResponseFactoryInterface $responseFactory, StreamFactoryInterface $streamFactory)
|
||||||
|
{
|
||||||
|
$this->client = $client;
|
||||||
|
$this->promisePool = $promisePool;
|
||||||
|
$this->responseFactory = $responseFactory;
|
||||||
|
$this->streamFactory = $streamFactory;
|
||||||
|
}
|
||||||
|
|
||||||
|
public function wait(?ResponseInterface $pendingResponse, float $maxDuration = null, float $idleTimeout = null): int
|
||||||
|
{
|
||||||
|
if (!$this->promisePool) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
$guzzleQueue = \GuzzleHttp\Promise\queue();
|
||||||
|
|
||||||
|
if (0.0 === $remainingDuration = $maxDuration) {
|
||||||
|
$idleTimeout = 0.0;
|
||||||
|
} elseif (null !== $maxDuration) {
|
||||||
|
$startTime = microtime(true);
|
||||||
|
$idleTimeout = max(0.0, min($maxDuration / 5, $idleTimeout ?? $maxDuration));
|
||||||
|
}
|
||||||
|
|
||||||
|
do {
|
||||||
|
foreach ($this->client->stream($this->promisePool, $idleTimeout) as $response => $chunk) {
|
||||||
|
try {
|
||||||
|
if (null !== $maxDuration && $chunk->isTimeout()) {
|
||||||
|
goto check_duration;
|
||||||
|
}
|
||||||
|
|
||||||
|
if ($chunk->isFirst()) {
|
||||||
|
// Deactivate throwing on 3/4/5xx
|
||||||
|
$response->getStatusCode();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!$chunk->isLast()) {
|
||||||
|
goto check_duration;
|
||||||
|
}
|
||||||
|
|
||||||
|
if ([$request, $promise] = $this->promisePool[$response] ?? null) {
|
||||||
|
unset($this->promisePool[$response]);
|
||||||
|
$promise->resolve($this->createPsr7Response($response, true));
|
||||||
|
}
|
||||||
|
} catch (\Exception $e) {
|
||||||
|
if ([$request, $promise] = $this->promisePool[$response] ?? null) {
|
||||||
|
unset($this->promisePool[$response]);
|
||||||
|
|
||||||
|
if ($e instanceof TransportExceptionInterface) {
|
||||||
|
$e = new NetworkException($e->getMessage(), $request, $e);
|
||||||
|
}
|
||||||
|
|
||||||
|
$promise->reject($e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
$guzzleQueue->run();
|
||||||
|
|
||||||
|
if ($pendingResponse === $response) {
|
||||||
|
return $this->promisePool->count();
|
||||||
|
}
|
||||||
|
|
||||||
|
check_duration:
|
||||||
|
if (null !== $maxDuration && $idleTimeout && $idleTimeout > $remainingDuration = max(0.0, $maxDuration - microtime(true) + $startTime)) {
|
||||||
|
$idleTimeout = $remainingDuration / 5;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!$count = $this->promisePool->count()) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
} while (null === $maxDuration || 0 < $remainingDuration);
|
||||||
|
|
||||||
|
return $count;
|
||||||
|
}
|
||||||
|
|
||||||
|
public function createPsr7Response(ResponseInterface $response, bool $buffer = false): Psr7ResponseInterface
|
||||||
|
{
|
||||||
|
$psrResponse = $this->responseFactory->createResponse($response->getStatusCode());
|
||||||
|
|
||||||
|
foreach ($response->getHeaders(false) as $name => $values) {
|
||||||
|
foreach ($values as $value) {
|
||||||
|
$psrResponse = $psrResponse->withAddedHeader($name, $value);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (isset(class_uses($response)[ResponseTrait::class])) {
|
||||||
|
$body = $this->streamFactory->createStreamFromResource($response->toStream(false));
|
||||||
|
} elseif (!$buffer) {
|
||||||
|
$body = $this->streamFactory->createStreamFromResource(StreamWrapper::createResource($response, $this->client));
|
||||||
|
} else {
|
||||||
|
$body = $this->streamFactory->createStream($response->getContent(false));
|
||||||
|
}
|
||||||
|
|
||||||
|
if ($body->isSeekable()) {
|
||||||
|
$body->seek(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
return $psrResponse->withBody($body);
|
||||||
|
}
|
||||||
|
}
|
@ -23,12 +23,10 @@ use Psr\Http\Message\ResponseInterface as Psr7ResponseInterface;
|
|||||||
final class HttplugPromise implements HttplugPromiseInterface
|
final class HttplugPromise implements HttplugPromiseInterface
|
||||||
{
|
{
|
||||||
private $promise;
|
private $promise;
|
||||||
private $cancel;
|
|
||||||
|
|
||||||
public function __construct(GuzzlePromiseInterface $promise, callable $cancel = null)
|
public function __construct(GuzzlePromiseInterface $promise)
|
||||||
{
|
{
|
||||||
$this->promise = $promise;
|
$this->promise = $promise;
|
||||||
$this->cancel = $cancel;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public function then(callable $onFulfilled = null, callable $onRejected = null): self
|
public function then(callable $onFulfilled = null, callable $onRejected = null): self
|
||||||
@ -58,16 +56,4 @@ final class HttplugPromise implements HttplugPromiseInterface
|
|||||||
{
|
{
|
||||||
return $this->promise->wait($unwrap);
|
return $this->promise->wait($unwrap);
|
||||||
}
|
}
|
||||||
|
|
||||||
public function __destruct()
|
|
||||||
{
|
|
||||||
if ($this->cancel) {
|
|
||||||
($this->cancel)();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public function __wakeup()
|
|
||||||
{
|
|
||||||
throw new \BadMethodCallException('Cannot unserialize '.__CLASS__);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
@ -75,6 +75,31 @@ class HttplugClientTest extends TestCase
|
|||||||
$this->assertSame('HTTP/1.1', $body['SERVER_PROTOCOL']);
|
$this->assertSame('HTTP/1.1', $body['SERVER_PROTOCOL']);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public function testWait()
|
||||||
|
{
|
||||||
|
$client = new HttplugClient(new NativeHttpClient());
|
||||||
|
|
||||||
|
$successCallableCalled = false;
|
||||||
|
$failureCallableCalled = false;
|
||||||
|
$client->sendAsyncRequest($client->createRequest('GET', 'http://localhost:8057/timeout-body'))
|
||||||
|
->then(function (ResponseInterface $response) use (&$successCallableCalled) {
|
||||||
|
$successCallableCalled = true;
|
||||||
|
|
||||||
|
return $response;
|
||||||
|
}, function (\Exception $exception) use (&$failureCallableCalled) {
|
||||||
|
$failureCallableCalled = true;
|
||||||
|
|
||||||
|
throw $exception;
|
||||||
|
});
|
||||||
|
|
||||||
|
$client->wait(0);
|
||||||
|
$this->assertFalse($successCallableCalled, '$promise->then() should not be called yet.');
|
||||||
|
|
||||||
|
$client->wait();
|
||||||
|
$this->assertTrue($successCallableCalled, '$promise->then() should have been called.');
|
||||||
|
$this->assertFalse($failureCallableCalled, 'Failure callable should not be called when request is successful.');
|
||||||
|
}
|
||||||
|
|
||||||
public function testPostRequest()
|
public function testPostRequest()
|
||||||
{
|
{
|
||||||
$client = new HttplugClient(new NativeHttpClient());
|
$client = new HttplugClient(new NativeHttpClient());
|
||||||
|
Reference in New Issue
Block a user