*/ class Pool implements PromisorInterface { use NonSerializableTrait; /** * @var EachPromise */ private EachPromise $each; /** * @param ClientInterface $client Client used to send the requests. * @param iterable): (ResponseInterface|PromiseInterface)> $requests Requests or functions that return responses or response promises. * @param array{ * concurrency?: int|(callable(int): int), * options?: array{ * base_uri?: string|UriInterface, * allow_redirects?: bool|array{ * max?: int, * strict?: bool, * referer?: bool, * protocols?: non-empty-array, * on_redirect?: callable(RequestInterface, ResponseInterface, UriInterface, int|string): mixed, * track_redirects?: bool * }, * auth?: array{ * 0: string, * 1: string, * 2?: string|null * }|string|false|null, * body?: resource|string|null|StreamInterface|(callable&object)|\Iterator|\Stringable, * cert?: string|array{ * 0: string, * 1?: string|null * }, * cert_type?: string, * connect_timeout?: int|float, * cookies?: false|CookieJarInterface, * crypto_method?: int, * crypto_method_max?: int, * debug?: bool|resource, * decode_content?: bool|string, * delay?: int|float, * expect?: bool|int, * form_params?: array, * force_ip_resolve?: string, * headers?: array>|null, * http_errors?: bool, * idn_conversion?: bool|int|null, * json?: mixed, * multipart?: array, * filename?: string * }>, * multiplex?: string, * on_headers?: callable(ResponseInterface, RequestInterface, int|string): mixed, * on_stats?: callable(TransferStats, int|string): mixed, * on_trailers?: callable(array>, ResponseInterface, RequestInterface, int|string): mixed, * progress?: callable(int, int, int, int, int|string): mixed, * protocols?: non-empty-array, * proxy?: string|array{ * http?: string|null, * https?: string|null, * no?: string|array|null * }, * query?: array|string, * read_timeout?: int|float, * retries?: int, * request_factory?: RequestFactoryInterface, * response_factory?: ResponseFactoryInterface, * sink?: resource|string|StreamInterface, * ssl_key?: string|array{ * 0: string, * 1?: string|null * }, * ssl_key_type?: string, * stream?: bool, * stream_factory?: StreamFactoryInterface, * stream_context?: array, * synchronous?: bool, * timeout?: int|float, * uri_factory?: UriFactoryInterface, * verify?: bool|string, * version?: string|int|float, * curl?: array, * ... * }, * fulfilled?: callable(ResponseInterface, int|string, PromiseInterface): mixed, * rejected?: callable(mixed, int|string, PromiseInterface): mixed * } $config Pool configuration. */ public function __construct( ClientInterface $client, iterable $requests, #[\SensitiveParameter] array $config = [] ) { if (!isset($config['concurrency'])) { $config['concurrency'] = 25; } if (isset($config['options'])) { $opts = $config['options']; unset($config['options']); } else { $opts = []; } $requestGenerator = static function () use ($requests, $client, $opts): \Generator { foreach ($requests as $key => $rfn) { $keyedOpts = self::keyedRequestOptions($opts, $key); if ($rfn instanceof RequestInterface) { yield $key => $client->sendAsync($rfn, $keyedOpts); } elseif (\is_callable($rfn)) { yield $key => $rfn($keyedOpts); } else { throw new \InvalidArgumentException('Each value yielded by the iterator must be a Psr\Http\Message\RequestInterface or a callable that returns a promise that fulfills with a Psr\Http\Message\ResponseInterface object.'); } } }; $this->each = new EachPromise($requestGenerator(), $config); } /** * Get promise * * @return PromiseInterface */ public function promise(): PromiseInterface { return $this->each->promise(); } /** * Sends multiple requests concurrently and returns an array of responses * and exceptions that uses the same ordering as the provided requests. * * IMPORTANT: This method keeps every request and response in memory, and * as such, is NOT recommended when sending a large number or an * indeterminate number of requests concurrently. * * @param ClientInterface $client Client used to send the requests * @param iterable): (ResponseInterface|PromiseInterface)> $requests Requests or functions that return responses or response promises. * @param array{ * concurrency?: int|(callable(int): int), * options?: array{ * base_uri?: string|UriInterface, * allow_redirects?: bool|array{ * max?: int, * strict?: bool, * referer?: bool, * protocols?: non-empty-array, * on_redirect?: callable(RequestInterface, ResponseInterface, UriInterface, int|string): mixed, * track_redirects?: bool * }, * auth?: array{ * 0: string, * 1: string, * 2?: string|null * }|string|false|null, * body?: resource|string|null|StreamInterface|(callable&object)|\Iterator|\Stringable, * cert?: string|array{ * 0: string, * 1?: string|null * }, * cert_type?: string, * connect_timeout?: int|float, * cookies?: false|CookieJarInterface, * crypto_method?: int, * crypto_method_max?: int, * debug?: bool|resource, * decode_content?: bool|string, * delay?: int|float, * expect?: bool|int, * form_params?: array, * force_ip_resolve?: string, * headers?: array>|null, * http_errors?: bool, * idn_conversion?: bool|int|null, * json?: mixed, * multipart?: array, * filename?: string * }>, * multiplex?: string, * on_headers?: callable(ResponseInterface, RequestInterface, int|string): mixed, * on_stats?: callable(TransferStats, int|string): mixed, * on_trailers?: callable(array>, ResponseInterface, RequestInterface, int|string): mixed, * progress?: callable(int, int, int, int, int|string): mixed, * protocols?: non-empty-array, * proxy?: string|array{ * http?: string|null, * https?: string|null, * no?: string|array|null * }, * query?: array|string, * read_timeout?: int|float, * retries?: int, * request_factory?: RequestFactoryInterface, * response_factory?: ResponseFactoryInterface, * sink?: resource|string|StreamInterface, * ssl_key?: string|array{ * 0: string, * 1?: string|null * }, * ssl_key_type?: string, * stream?: bool, * stream_factory?: StreamFactoryInterface, * stream_context?: array, * synchronous?: bool, * timeout?: int|float, * uri_factory?: UriFactoryInterface, * verify?: bool|string, * version?: string|int|float, * curl?: array, * ... * }, * fulfilled?: callable(ResponseInterface, int|string): mixed, * rejected?: callable(mixed, int|string): mixed * } $options Passes through the options available in {@see Pool::__construct}. * * @return array Returns an array containing the response or rejection reason in the same order that the requests were sent. * * @throws \InvalidArgumentException if the event format is incorrect. */ public static function batch( ClientInterface $client, iterable $requests, #[\SensitiveParameter] array $options = [] ): array { $res = []; self::cmpCallback($options, 'fulfilled', $res); self::cmpCallback($options, 'rejected', $res); $pool = new static($client, $requests, $options); $pool->promise()->wait(); \ksort($res); return $res; } /** * Execute callback(s) */ private static function cmpCallback(array &$options, string $name, array &$results): void { if (!isset($options[$name])) { $options[$name] = static function ( #[\SensitiveParameter] $v, $k ) use (&$results): void { $results[$k] = $v; }; } else { $currentFn = $options[$name]; $options[$name] = static function ( #[\SensitiveParameter] $v, $k ) use (&$results, $currentFn): void { $currentFn($v, $k); $results[$k] = $v; }; } } /** * Returns the request options with any observer callbacks wrapped so that * they also receive the request's iterable key as a trailing argument. * * @param array{ * base_uri?: string|UriInterface, * allow_redirects?: bool|array{ * max?: int, * strict?: bool, * referer?: bool, * protocols?: non-empty-array, * on_redirect?: callable(RequestInterface, ResponseInterface, UriInterface, int|string): mixed, * track_redirects?: bool * }, * auth?: array{ * 0: string, * 1: string, * 2?: string|null * }|string|false|null, * body?: resource|string|null|StreamInterface|(callable&object)|\Iterator|\Stringable, * cert?: string|array{ * 0: string, * 1?: string|null * }, * cert_type?: string, * connect_timeout?: int|float, * cookies?: false|CookieJarInterface, * crypto_method?: int, * crypto_method_max?: int, * debug?: bool|resource, * decode_content?: bool|string, * delay?: int|float, * expect?: bool|int, * form_params?: array, * force_ip_resolve?: string, * headers?: array>|null, * http_errors?: bool, * idn_conversion?: bool|int|null, * json?: mixed, * multipart?: array, * filename?: string * }>, * multiplex?: string, * on_headers?: callable(ResponseInterface, RequestInterface, int|string): mixed, * on_stats?: callable(TransferStats, int|string): mixed, * on_trailers?: callable(array>, ResponseInterface, RequestInterface, int|string): mixed, * progress?: callable(int, int, int, int, int|string): mixed, * protocols?: non-empty-array, * proxy?: string|array{ * http?: string|null, * https?: string|null, * no?: string|array|null * }, * query?: array|string, * read_timeout?: int|float, * retries?: int, * request_factory?: RequestFactoryInterface, * response_factory?: ResponseFactoryInterface, * sink?: resource|string|StreamInterface, * ssl_key?: string|array{ * 0: string, * 1?: string|null * }, * ssl_key_type?: string, * stream?: bool, * stream_factory?: StreamFactoryInterface, * stream_context?: array, * synchronous?: bool, * timeout?: int|float, * uri_factory?: UriFactoryInterface, * verify?: bool|string, * version?: string|int|float, * curl?: array, * ... * } $options * @param int|string $key * * @return array{ * base_uri?: string|UriInterface, * allow_redirects?: bool|array{ * max?: int, * strict?: bool, * referer?: bool, * protocols?: non-empty-array, * on_redirect?: callable(RequestInterface, ResponseInterface, UriInterface): mixed, * track_redirects?: bool * }, * auth?: array{ * 0: string, * 1: string, * 2?: string|null * }|string|false|null, * body?: resource|string|null|StreamInterface|(callable&object)|\Iterator|\Stringable, * cert?: string|array{ * 0: string, * 1?: string|null * }, * cert_type?: string, * connect_timeout?: int|float, * cookies?: false|CookieJarInterface, * crypto_method?: int, * crypto_method_max?: int, * debug?: bool|resource, * decode_content?: bool|string, * delay?: int|float, * expect?: bool|int, * form_params?: array, * force_ip_resolve?: string, * headers?: array>|null, * http_errors?: bool, * idn_conversion?: bool|int|null, * json?: mixed, * multipart?: array, * filename?: string * }>, * multiplex?: string, * on_headers?: callable(ResponseInterface, RequestInterface): mixed, * on_stats?: callable(TransferStats): mixed, * on_trailers?: callable(array>, ResponseInterface, RequestInterface): mixed, * progress?: callable(int, int, int, int): mixed, * protocols?: non-empty-array, * proxy?: string|array{ * http?: string|null, * https?: string|null, * no?: string|array|null * }, * query?: array|string, * read_timeout?: int|float, * retries?: int, * request_factory?: RequestFactoryInterface, * response_factory?: ResponseFactoryInterface, * sink?: resource|string|StreamInterface, * ssl_key?: string|array{ * 0: string, * 1?: string|null * }, * ssl_key_type?: string, * stream?: bool, * stream_factory?: StreamFactoryInterface, * stream_context?: array, * synchronous?: bool, * timeout?: int|float, * uri_factory?: UriFactoryInterface, * verify?: bool|string, * version?: string|int|float, * curl?: array, * ... * } */ private static function keyedRequestOptions(array $options, $key): array { if (\is_array($options['allow_redirects'] ?? null) && \is_callable($options['allow_redirects']['on_redirect'] ?? null) ) { $onRedirect = $options['allow_redirects']['on_redirect']; $options['allow_redirects']['on_redirect'] = static function ( #[\SensitiveParameter] RequestInterface $request, #[\SensitiveParameter] ResponseInterface $response, #[\SensitiveParameter] UriInterface $uri ) use ($onRedirect, $key): void { $onRedirect($request, $response, $uri, $key); }; } if (\is_callable($options['on_headers'] ?? null)) { $onHeaders = $options['on_headers']; $options['on_headers'] = static function ( #[\SensitiveParameter] ResponseInterface $response, #[\SensitiveParameter] RequestInterface $request ) use ($onHeaders, $key): void { $onHeaders($response, $request, $key); }; } if (\is_callable($options['on_stats'] ?? null)) { $onStats = $options['on_stats']; $options['on_stats'] = static function (TransferStats $stats) use ($onStats, $key): void { $onStats($stats, $key); }; } if (\is_callable($options['on_trailers'] ?? null)) { $onTrailers = $options['on_trailers']; $options['on_trailers'] = static function ( array $trailers, #[\SensitiveParameter] ResponseInterface $response, #[\SensitiveParameter] RequestInterface $request ) use ($onTrailers, $key): void { $onTrailers($trailers, $response, $request, $key); }; } if (\is_callable($options['progress'] ?? null)) { $progress = $options['progress']; $options['progress'] = static function (int $downloadTotal, int $downloadedBytes, int $uploadTotal, int $uploadedBytes) use ($progress, $key) { return $progress($downloadTotal, $downloadedBytes, $uploadTotal, $uploadedBytes, $key); }; } return $options; } }