[HttpClient] fix monitoring timeouts when other streams are active
This commit is contained in:
parent
d87b6665dd
commit
d2a53f0bda
@ -28,8 +28,6 @@ final class NativeClientState extends ClientState
|
||||
public $responseCount = 0;
|
||||
/** @var string[] */
|
||||
public $dnsCache = [];
|
||||
/** @var resource[] */
|
||||
public $handles = [];
|
||||
/** @var bool */
|
||||
public $sleep = false;
|
||||
|
||||
|
@ -220,11 +220,6 @@ final class NativeResponse implements ResponseInterface
|
||||
*/
|
||||
private static function perform(ClientState $multi, array &$responses = null): void
|
||||
{
|
||||
// List of native handles for stream_select()
|
||||
if (null !== $responses) {
|
||||
$multi->handles = [];
|
||||
}
|
||||
|
||||
foreach ($multi->openHandles as $i => [$h, $buffer, $onProgress]) {
|
||||
$hasActivity = false;
|
||||
$remaining = &$multi->openHandles[$i][3];
|
||||
@ -291,8 +286,6 @@ final class NativeResponse implements ResponseInterface
|
||||
$multi->handlesActivity[$i][] = $e;
|
||||
unset($multi->openHandles[$i]);
|
||||
$multi->sleep = false;
|
||||
} elseif (null !== $responses) {
|
||||
$multi->handles[] = $h;
|
||||
}
|
||||
}
|
||||
|
||||
@ -307,7 +300,7 @@ final class NativeResponse implements ResponseInterface
|
||||
}
|
||||
}
|
||||
|
||||
if (\count($multi->handles) >= $multi->maxHostConnections) {
|
||||
if (\count($multi->openHandles) >= $multi->maxHostConnections) {
|
||||
return;
|
||||
}
|
||||
|
||||
@ -318,10 +311,6 @@ final class NativeResponse implements ResponseInterface
|
||||
$multi->sleep = false;
|
||||
self::perform($multi);
|
||||
|
||||
if (null !== $response->handle) {
|
||||
$multi->handles[] = $response->handle;
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
@ -335,7 +324,8 @@ final class NativeResponse implements ResponseInterface
|
||||
private static function select(ClientState $multi, float $timeout): int
|
||||
{
|
||||
$_ = [];
|
||||
$handles = array_column($multi->openHandles, 0);
|
||||
|
||||
return (!$multi->sleep = !$multi->sleep) ? -1 : stream_select($multi->handles, $_, $_, (int) $timeout, (int) (1E6 * ($timeout - (int) $timeout)));
|
||||
return (!$multi->sleep = !$multi->sleep) ? -1 : stream_select($handles, $_, $_, (int) $timeout, (int) (1E6 * ($timeout - (int) $timeout)));
|
||||
}
|
||||
}
|
||||
|
@ -316,7 +316,7 @@ trait ResponseTrait
|
||||
}
|
||||
|
||||
$lastActivity = microtime(true);
|
||||
$isTimeout = false;
|
||||
$enlapsedTimeout = 0;
|
||||
|
||||
while (true) {
|
||||
$hasActivity = false;
|
||||
@ -338,7 +338,7 @@ trait ResponseTrait
|
||||
} elseif (!isset($multi->openHandles[$j])) {
|
||||
unset($responses[$j]);
|
||||
continue;
|
||||
} elseif ($isTimeout) {
|
||||
} elseif ($enlapsedTimeout >= $timeoutMax) {
|
||||
$multi->handlesActivity[$j] = [new ErrorChunk($response->offset, sprintf('Idle timeout reached for "%s".', $response->getInfo('url')))];
|
||||
} else {
|
||||
continue;
|
||||
@ -346,7 +346,7 @@ trait ResponseTrait
|
||||
|
||||
while ($multi->handlesActivity[$j] ?? false) {
|
||||
$hasActivity = true;
|
||||
$isTimeout = false;
|
||||
$enlapsedTimeout = 0;
|
||||
|
||||
if (\is_string($chunk = array_shift($multi->handlesActivity[$j]))) {
|
||||
if (null !== $response->inflate && false === $chunk = @inflate_add($response->inflate, $chunk)) {
|
||||
@ -379,7 +379,7 @@ trait ResponseTrait
|
||||
}
|
||||
} elseif ($chunk instanceof ErrorChunk) {
|
||||
unset($responses[$j]);
|
||||
$isTimeout = true;
|
||||
$enlapsedTimeout = $timeoutMax;
|
||||
} elseif ($chunk instanceof FirstChunk) {
|
||||
if ($response->logger) {
|
||||
$info = $response->getInfo();
|
||||
@ -447,10 +447,11 @@ trait ResponseTrait
|
||||
continue;
|
||||
}
|
||||
|
||||
switch (self::select($multi, $timeoutMin)) {
|
||||
case -1: usleep(min(500, 1E6 * $timeoutMin)); break;
|
||||
case 0: $isTimeout = microtime(true) - $lastActivity > $timeoutMax; break;
|
||||
}
|
||||
if (-1 === self::select($multi, min($timeoutMin, $timeoutMax - $enlapsedTimeout))) {
|
||||
usleep(min(500, 1E6 * $timeoutMin));
|
||||
}
|
||||
|
||||
$enlapsedTimeout = microtime(true) - $lastActivity;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -112,6 +112,15 @@ class CurlHttpClientTest extends HttpClientTestCase
|
||||
$this->assertSame($expected, $logger->logs);
|
||||
}
|
||||
|
||||
public function testTimeoutIsNotAFatalError()
|
||||
{
|
||||
if ('\\' === \DIRECTORY_SEPARATOR) {
|
||||
$this->markTestSkipped('Too transient on Windows');
|
||||
}
|
||||
|
||||
parent::testTimeoutIsNotAFatalError();
|
||||
}
|
||||
|
||||
private function getVulcainClient(): CurlHttpClient
|
||||
{
|
||||
if (\PHP_VERSION_ID >= 70300 && \PHP_VERSION_ID < 70304) {
|
||||
|
@ -72,9 +72,9 @@ abstract class HttpClientTestCase extends BaseHttpClientTestCase
|
||||
$this->assertSame($response, stream_get_meta_data($stream)['wrapper_data']->getResponse());
|
||||
$this->assertSame(404, $response->getStatusCode());
|
||||
|
||||
$this->expectException(ClientException::class);
|
||||
$response = $client->request('GET', 'http://localhost:8057/404');
|
||||
$stream = $response->toStream();
|
||||
$this->expectException(ClientException::class);
|
||||
$response->toStream();
|
||||
}
|
||||
|
||||
public function testNonBlockingStream()
|
||||
@ -82,6 +82,7 @@ abstract class HttpClientTestCase extends BaseHttpClientTestCase
|
||||
$client = $this->getHttpClient(__FUNCTION__);
|
||||
$response = $client->request('GET', 'http://localhost:8057/timeout-body');
|
||||
$stream = $response->toStream();
|
||||
usleep(10000);
|
||||
|
||||
$this->assertTrue(stream_set_blocking($stream, false));
|
||||
$this->assertSame('<1>', fread($stream, 8192));
|
||||
@ -99,6 +100,7 @@ abstract class HttpClientTestCase extends BaseHttpClientTestCase
|
||||
$response = $client->request('GET', 'http://localhost:8057/timeout-body', [
|
||||
'timeout' => 0.25,
|
||||
]);
|
||||
$this->assertSame(200, $response->getStatusCode());
|
||||
|
||||
try {
|
||||
$response->getContent();
|
||||
|
@ -69,6 +69,10 @@ class MockHttpClientTest extends HttpClientTestCase
|
||||
$this->markTestSkipped("MockHttpClient doesn't unzip");
|
||||
break;
|
||||
|
||||
case 'testTimeoutWithActiveConcurrentStream':
|
||||
$this->markTestSkipped('Real transport required');
|
||||
break;
|
||||
|
||||
case 'testDestruct':
|
||||
$this->markTestSkipped("MockHttpClient doesn't timeout on destruct");
|
||||
break;
|
||||
|
@ -786,6 +786,30 @@ abstract class HttpClientTestCase extends TestCase
|
||||
}
|
||||
}
|
||||
|
||||
public function testTimeoutWithActiveConcurrentStream()
|
||||
{
|
||||
$p1 = TestHttpServer::start(8067);
|
||||
$p2 = TestHttpServer::start(8077);
|
||||
|
||||
$client = $this->getHttpClient(__FUNCTION__);
|
||||
$streamingResponse = $client->request('GET', 'http://localhost:8067/max-duration');
|
||||
$blockingResponse = $client->request('GET', 'http://localhost:8077/timeout-body', [
|
||||
'timeout' => 0.25,
|
||||
]);
|
||||
|
||||
$this->assertSame(200, $streamingResponse->getStatusCode());
|
||||
$this->assertSame(200, $blockingResponse->getStatusCode());
|
||||
|
||||
$this->expectException(TransportExceptionInterface::class);
|
||||
|
||||
try {
|
||||
$blockingResponse->getContent();
|
||||
} finally {
|
||||
$p1->stop();
|
||||
$p2->stop();
|
||||
}
|
||||
}
|
||||
|
||||
public function testDestruct()
|
||||
{
|
||||
$client = $this->getHttpClient(__FUNCTION__);
|
||||
|
@ -19,23 +19,28 @@ use Symfony\Component\Process\Process;
|
||||
*/
|
||||
class TestHttpServer
|
||||
{
|
||||
private static $process;
|
||||
private static $process = [];
|
||||
|
||||
public static function start()
|
||||
public static function start(int $port = 8057)
|
||||
{
|
||||
if (self::$process) {
|
||||
self::$process->stop();
|
||||
if (isset(self::$process[$port])) {
|
||||
self::$process[$port]->stop();
|
||||
} else {
|
||||
register_shutdown_function(static function () use ($port) {
|
||||
self::$process[$port]->stop();
|
||||
});
|
||||
}
|
||||
|
||||
$finder = new PhpExecutableFinder();
|
||||
$process = new Process(array_merge([$finder->find(false)], $finder->findArguments(), ['-dopcache.enable=0', '-dvariables_order=EGPCS', '-S', '127.0.0.1:8057']));
|
||||
$process = new Process(array_merge([$finder->find(false)], $finder->findArguments(), ['-dopcache.enable=0', '-dvariables_order=EGPCS', '-S', '127.0.0.1:'.$port]));
|
||||
$process->setWorkingDirectory(__DIR__.'/Fixtures/web');
|
||||
$process->start();
|
||||
self::$process[$port] = $process;
|
||||
|
||||
do {
|
||||
usleep(50000);
|
||||
} while (!@fopen('http://127.0.0.1:8057/', 'r'));
|
||||
} while (!@fopen('http://127.0.0.1:'.$port, 'r'));
|
||||
|
||||
self::$process = $process;
|
||||
return $process;
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user