krz/vox-populi

  1<?php
  2namespace GuzzleHttp;
  3
  4use GuzzleHttp\Promise\EachPromise;
  5use GuzzleHttp\Promise\PromiseInterface;
  6use GuzzleHttp\Promise\PromisorInterface;
  7use Psr\Http\Message\RequestInterface;
  8
  9/**
 10 * Sends an iterator of requests concurrently using a capped pool size.
 11 *
 12 * The pool will read from an iterator until it is cancelled or until the
 13 * iterator is consumed. When a request is yielded, the request is sent after
 14 * applying the "request_options" request options (if provided in the ctor).
 15 *
 16 * When a function is yielded by the iterator, the function is provided the
 17 * "request_options" array that should be merged on top of any existing
 18 * options, and the function MUST then return a wait-able promise.
 19 */
 20class Pool implements PromisorInterface
 21{
 22    /** @var EachPromise */
 23    private $each;
 24
 25    /**
 26     * @param ClientInterface $client   Client used to send the requests.
 27     * @param array|\Iterator $requests Requests or functions that return
 28     *                                  requests to send concurrently.
 29     * @param array           $config   Associative array of options
 30     *     - concurrency: (int) Maximum number of requests to send concurrently
 31     *     - options: Array of request options to apply to each request.
 32     *     - fulfilled: (callable) Function to invoke when a request completes.
 33     *     - rejected: (callable) Function to invoke when a request is rejected.
 34     */
 35    public function __construct(
 36        ClientInterface $client,
 37        $requests,
 38        array $config = []
 39    ) {
 40        // Backwards compatibility.
 41        if (isset($config['pool_size'])) {
 42            $config['concurrency'] = $config['pool_size'];
 43        } elseif (!isset($config['concurrency'])) {
 44            $config['concurrency'] = 25;
 45        }
 46
 47        if (isset($config['options'])) {
 48            $opts = $config['options'];
 49            unset($config['options']);
 50        } else {
 51            $opts = [];
 52        }
 53
 54        $iterable = \GuzzleHttp\Promise\iter_for($requests);
 55        $requests = function () use ($iterable, $client, $opts) {
 56            foreach ($iterable as $key => $rfn) {
 57                if ($rfn instanceof RequestInterface) {
 58                    yield $key => $client->sendAsync($rfn, $opts);
 59                } elseif (is_callable($rfn)) {
 60                    yield $key => $rfn($opts);
 61                } else {
 62                    throw new \InvalidArgumentException('Each value yielded by '
 63                        . 'the iterator must be a Psr7\Http\Message\RequestInterface '
 64                        . 'or a callable that returns a promise that fulfills '
 65                        . 'with a Psr7\Message\Http\ResponseInterface object.');
 66                }
 67            }
 68        };
 69
 70        $this->each = new EachPromise($requests(), $config);
 71    }
 72
 73    /**
 74     * Get promise
 75     *
 76     * @return PromiseInterface
 77     */
 78    public function promise()
 79    {
 80        return $this->each->promise();
 81    }
 82
 83    /**
 84     * Sends multiple requests concurrently and returns an array of responses
 85     * and exceptions that uses the same ordering as the provided requests.
 86     *
 87     * IMPORTANT: This method keeps every request and response in memory, and
 88     * as such, is NOT recommended when sending a large number or an
 89     * indeterminate number of requests concurrently.
 90     *
 91     * @param ClientInterface $client   Client used to send the requests
 92     * @param array|\Iterator $requests Requests to send concurrently.
 93     * @param array           $options  Passes through the options available in
 94     *                                  {@see GuzzleHttp\Pool::__construct}
 95     *
 96     * @return array Returns an array containing the response or an exception
 97     *               in the same order that the requests were sent.
 98     * @throws \InvalidArgumentException if the event format is incorrect.
 99     */
100    public static function batch(
101        ClientInterface $client,
102        $requests,
103        array $options = []
104    ) {
105        $res = [];
106        self::cmpCallback($options, 'fulfilled', $res);
107        self::cmpCallback($options, 'rejected', $res);
108        $pool = new static($client, $requests, $options);
109        $pool->promise()->wait();
110        ksort($res);
111
112        return $res;
113    }
114
115    /**
116     * Execute callback(s)
117     *
118     * @return void
119     */
120    private static function cmpCallback(array &$options, $name, array &$results)
121    {
122        if (!isset($options[$name])) {
123            $options[$name] = function ($v, $k) use (&$results) {
124                $results[$k] = $v;
125            };
126        } else {
127            $currentFn = $options[$name];
128            $options[$name] = function ($v, $k) use (&$results, $currentFn) {
129                $currentFn($v, $k);
130                $results[$k] = $v;
131            };
132        }
133    }
134}