PHP code example of aubes / http-pool-bundle

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

aubes / http-pool-bundle example snippets

$responses = [];
$responses['user'] = $httpClient->request('GET', "https://api.example.com/users/{$userId}");
$responses['orders'] = $httpClient->request('GET', "https://api.example.com/orders?user={$userId}");

$results = [];
foreach ($httpClient->stream($responses) as $response => $chunk) {
    if ($chunk->isLast()) {
        $key = array_search($response, $responses, true);
        $results[$key] = $response->toArray();
    }
}
// No concurrency limit, no rate limiting, no retry, no fan-out,
// no error handling per request, and it gets worse with each new API.

$pool = $this->httpPool->create(concurrency: 10);

$pool->add('user', 'GET', "https://api.example.com/users/{$userId}");
$pool->add('orders', 'GET', "https://api.example.com/orders?user={$userId}");

$results = $pool->flush();
$user = $results->get('user')->toArray();
$orders = $results->get('orders')->toArray();

use Aubes\HttpPoolBundle\Pool\PoolFactoryInterface;
use Aubes\HttpPoolBundle\Pool\PoolInterface;
use Symfony\Contracts\HttpClient\ResponseInterface;

class MyService
{
    public function __construct(
        private readonly PoolFactoryInterface $httpPool,
    ) {}

    public function fetchUserData(int $userId): array
    {
        $pool = $this->httpPool->create(concurrency: 10);

        $pool->add('user', 'GET', "https://api.example.com/users/{$userId}")
            ->then(function (ResponseInterface $response, PoolInterface $pool) {
                $user = $response->toArray();

                // Fan-out: callbacks can add requests to the pool
                $pool->add('orders', 'GET', "https://api.example.com/orders?user={$user['id']}");
                $pool->add('avatar', 'GET', $user['avatar_url']);
            });

        $results = $pool->flush();

        return [
            'user' => $results->get('user')->toArray(),
            'orders' => $results->get('orders')->toArray(),
            'avatar' => $results->get('avatar')->getContent(),
        ];
    }
}

$pool = $this->httpPool->create(concurrency: 5);

for ($i = 0; $i < 100; $i++) {
    $pool->add("item_{$i}", 'GET', "https://api.example.com/items/{$i}");
}

// 100 requests executed in batches of 5
$results = $pool->flush();

$pool->add('user', 'GET', 'https://api.example.com/users/42')
    ->then(function (ResponseInterface $response, PoolInterface $pool) {
        $user = $response->toArray();

        // Level 2: requests triggered by the response
        $pool->add('orders', 'GET', "https://api.example.com/orders?user={$user['id']}")
            ->then(function (ResponseInterface $response, PoolInterface $pool) {
                // Level 3: nest as deep as needed
                foreach ($response->toArray() as $order) {
                    $pool->add(
                        "invoice_{$order['id']}",
                        'GET',
                        "https://api.example.com/invoices/{$order['invoiceId']}",
                    );
                }
            });
    });

$results = $pool->flush();

// All responses are accessible in a flat structure
$user = $results->get('user')->toArray();
$orders = $results->get('orders')->toArray();
$invoice1 = $results->get('invoice_1')->toArray();

// Two products share the same brand: only one HTTP request
$pool->add('product_1', 'GET', 'https://api.example.com/products/1')
    ->then(function (ResponseInterface $response, PoolInterface $pool) use (&$product1) {
        $data = $response->toArray();
        $pool->addOnce("brand_{$data['brandId']}", 'GET', "https://api.example.com/brands/{$data['brandId']}")
            ->then(function (ResponseInterface $response) use (&$product1) {
                $product1['brand'] = $response->toArray();
            });
    });

$pool->add('product_2', 'GET', 'https://api.example.com/products/2')
    ->then(function (ResponseInterface $response, PoolInterface $pool) use (&$product2) {
        $data = $response->toArray();
        // Same brandId: no new request, the then() receives the cached response
        $pool->addOnce("brand_{$data['brandId']}", 'GET', "https://api.example.com/brands/{$data['brandId']}")
            ->then(function (ResponseInterface $response) use (&$product2) {
                $product2['brand'] = $response->toArray();
            });
    });

$pool->add('user', 'GET', 'https://api.example.com/users/42');
$pool->fire('POST', 'https://analytics.example.com/events', [
    'json' => ['event' => 'user_viewed', 'user_id' => 42],
]);

$results = $pool->flush();
// $results contains 'user' but not the fire-and-forget request

use Aubes\HttpPoolBundle\ErrorStrategy;

// Default: collect errors, flush() continues
$pool = $this->httpPool->create(errorStrategy: ErrorStrategy::Collect);

// Stop on the first unhandled error, cancel in-flight requests
$pool = $this->httpPool->create(errorStrategy: ErrorStrategy::StopOnFirst);

// Execute everything, then throw an aggregate PoolException
$pool = $this->httpPool->create(errorStrategy: ErrorStrategy::ThrowAll);

$pool->add('primary', 'GET', 'https://api.example.com/primary')
    ->catch(function (\Throwable $e, PoolInterface $pool) {
        // Fallback: schedule an alternative request
        $pool->add('fallback', 'GET', 'https://api.example.com/fallback');
        // Does not rethrow: error handled
    });

use Aubes\HttpPoolBundle\Exception\CallbackException;

$results = $pool->flush();

foreach ($results->getErrors() as $key => $error) {
    if ($error instanceof CallbackException) {
        // Error in the callback code, not in the HTTP request
        $originalResponse = $error->getResponse(); // the successful HTTP response
        $cause = $error->getPrevious();             // the original exception
    } else {
        // HTTP error (timeout, 500, etc.)
    }
}

// Via Symfony config (see Configuration)
// Or directly via create():
$pool = $this->httpPool->create(retry: [
    503 => 3,  // max 3 attempts on 503
]);

$pool = $this->httpPool->create(
    concurrency: 20,
    rateLimits: [
        'orders-api.internal' => 20,  // 20 req/s
        'users-api.internal' => 50,   // 50 req/s
    ],
);

use Symfony\Component\DependencyInjection\Attribute\Target;

public function __construct(
    #[Target('ordersPoolFactory')]
    private readonly PoolFactoryInterface $ordersPoolFactory,
) {}