Fiber powered Symfony HttpClient with Revolt (and AMPHP)

composer.json

{
    "require": {
        "php": ">=8.1.2",
        "amphp/amp": "v3.x-dev",
        "symfony/http-client": "^6.0"
    }
}

demo.php

<?php

namespace Demo;

use App\RevoltHttpClient;
use Symfony\Component\HttpClient\CurlHttpClient;
use function Amp\async;
use function Amp\Future\await;

require __DIR__.'/vendor/autoload.php';
require __DIR__.'/RevoltHttpClient.php';

$httpClient = new RevoltHttpClient(new CurlHttpClient());

$request = function (string $version) use ($httpClient) {
    return async(function () use ($version, $httpClient) {
        printf('0-Start %s%s', $version, PHP_EOL);
        $response1 = $httpClient->request('GET', 'https://symfony.com/releases/'.$version.'.json?1');
        printf('1-Continue %s%s', $version, PHP_EOL);
        $response2 = $httpClient->request('GET', 'https://symfony.com/releases/'.$version.'.json?2');
        printf('2-Continue %s%s', $version, PHP_EOL);
        $response3 = $httpClient->request('GET', 'https://symfony.com/releases/'.$version.'.json?3');
        printf('3-End %s%s', $version, PHP_EOL);

        return [
            $response1->toArray(),
            $response2->toArray(),
            $response3->toArray(),
        ];
    });
};

$responses = [
    $request('4.4'),
    $request('5.4'),
    $request('6.0'),
];

$data = await($responses);

printf('THE END%s', PHP_EOL);

//var_dump($data);

output.txt

0-Start 4.4
0-Start 5.4
0-Start 6.0
1-Continue 5.4
1-Continue 4.4
1-Continue 6.0
2-Continue 5.4
2-Continue 4.4
3-End 5.4
2-Continue 6.0
3-End 4.4
3-End 6.0
THE END


========================================================================
# The 3 fibers are run concurrently.
# When any request is completed, the corresponding fiber continues.
# Horizontal timeline. Each | is a new HTTP request that is sent.

4.4 |--------|-------|--------|
5.4  |-----|-------|-----|
6.0   |---------|-----------|-------|

RevoltHttpClient.php

<?php

namespace App;

use Revolt\EventLoop;
use Symfony\Contracts\HttpClient\HttpClientInterface;

class RevoltHttpClient
{
    private int $requests = 0;
    private array $responses = [];
    private array $suspensions = [];
    private bool $registered = false;

    public function __construct(
        private readonly HttpClientInterface $httpClient
    )
    {
    }

    public function request(string $method, string $url, array $options = [])
    {
        $response = $this->httpClient->request($method, $url, $options);
        $id = $this->requests++;
        $this->suspensions[$id] = $suspension = EventLoop::getSuspension();
        $this->responses[$id] = $response;

        $this->register();

        // Here is the magic. The function is interrupted and wait
        // The "race" method will resume when the response is received.
        $suspension->suspend();

        return $response;
    }

    private function register(): void
    {
        if (!$this->registered) {
            EventLoop::defer($this->race(...));
            $this->registered = true;
        }
    }

    /**
     * Resume the suspension for the first received response.
     *
     * @todo Would be more efficient if HttpClient had a "race" function
     */
    private function race(): void
    {
        $stream = $this->httpClient->stream($this->responses);

        foreach ($stream as $response => $chunk) {
            if ($chunk->isLast()) {
                $this->registered = false;
                $id = array_search($response, $this->responses);
                /** @var EventLoop\Suspension $suspension */
                $suspension = $this->suspensions[$id];

                unset($this->responses[$id], $this->suspensions[$id]);
                if (\count($this->responses) > 0) {
                    $this->register();
                }

                $suspension->resume();

                return;
            }
        }

        throw new \RuntimeException('A request should have ended');
    }
}