Chris@0: <?php
Chris@0: namespace GuzzleHttp;
Chris@0: 
Chris@0: use GuzzleHttp\Promise\PromisorInterface;
Chris@0: use Psr\Http\Message\RequestInterface;
Chris@0: use GuzzleHttp\Promise\EachPromise;
Chris@0: 
Chris@0: /**
Chris@0:  * Sends and iterator of requests concurrently using a capped pool size.
Chris@0:  *
Chris@0:  * The pool will read from an iterator until it is cancelled or until the
Chris@0:  * iterator is consumed. When a request is yielded, the request is sent after
Chris@0:  * applying the "request_options" request options (if provided in the ctor).
Chris@0:  *
Chris@0:  * When a function is yielded by the iterator, the function is provided the
Chris@0:  * "request_options" array that should be merged on top of any existing
Chris@0:  * options, and the function MUST then return a wait-able promise.
Chris@0:  */
Chris@0: class Pool implements PromisorInterface
Chris@0: {
Chris@0:     /** @var EachPromise */
Chris@0:     private $each;
Chris@0: 
Chris@0:     /**
Chris@0:      * @param ClientInterface $client   Client used to send the requests.
Chris@0:      * @param array|\Iterator $requests Requests or functions that return
Chris@0:      *                                  requests to send concurrently.
Chris@0:      * @param array           $config   Associative array of options
Chris@0:      *     - concurrency: (int) Maximum number of requests to send concurrently
Chris@0:      *     - options: Array of request options to apply to each request.
Chris@0:      *     - fulfilled: (callable) Function to invoke when a request completes.
Chris@0:      *     - rejected: (callable) Function to invoke when a request is rejected.
Chris@0:      */
Chris@0:     public function __construct(
Chris@0:         ClientInterface $client,
Chris@0:         $requests,
Chris@0:         array $config = []
Chris@0:     ) {
Chris@0:         // Backwards compatibility.
Chris@0:         if (isset($config['pool_size'])) {
Chris@0:             $config['concurrency'] = $config['pool_size'];
Chris@0:         } elseif (!isset($config['concurrency'])) {
Chris@0:             $config['concurrency'] = 25;
Chris@0:         }
Chris@0: 
Chris@0:         if (isset($config['options'])) {
Chris@0:             $opts = $config['options'];
Chris@0:             unset($config['options']);
Chris@0:         } else {
Chris@0:             $opts = [];
Chris@0:         }
Chris@0: 
Chris@0:         $iterable = \GuzzleHttp\Promise\iter_for($requests);
Chris@0:         $requests = function () use ($iterable, $client, $opts) {
Chris@0:             foreach ($iterable as $key => $rfn) {
Chris@0:                 if ($rfn instanceof RequestInterface) {
Chris@0:                     yield $key => $client->sendAsync($rfn, $opts);
Chris@0:                 } elseif (is_callable($rfn)) {
Chris@0:                     yield $key => $rfn($opts);
Chris@0:                 } else {
Chris@0:                     throw new \InvalidArgumentException('Each value yielded by '
Chris@0:                         . 'the iterator must be a Psr7\Http\Message\RequestInterface '
Chris@0:                         . 'or a callable that returns a promise that fulfills '
Chris@0:                         . 'with a Psr7\Message\Http\ResponseInterface object.');
Chris@0:                 }
Chris@0:             }
Chris@0:         };
Chris@0: 
Chris@0:         $this->each = new EachPromise($requests(), $config);
Chris@0:     }
Chris@0: 
Chris@0:     public function promise()
Chris@0:     {
Chris@0:         return $this->each->promise();
Chris@0:     }
Chris@0: 
Chris@0:     /**
Chris@0:      * Sends multiple requests concurrently and returns an array of responses
Chris@0:      * and exceptions that uses the same ordering as the provided requests.
Chris@0:      *
Chris@0:      * IMPORTANT: This method keeps every request and response in memory, and
Chris@0:      * as such, is NOT recommended when sending a large number or an
Chris@0:      * indeterminate number of requests concurrently.
Chris@0:      *
Chris@0:      * @param ClientInterface $client   Client used to send the requests
Chris@0:      * @param array|\Iterator $requests Requests to send concurrently.
Chris@0:      * @param array           $options  Passes through the options available in
Chris@0:      *                                  {@see GuzzleHttp\Pool::__construct}
Chris@0:      *
Chris@0:      * @return array Returns an array containing the response or an exception
Chris@0:      *               in the same order that the requests were sent.
Chris@0:      * @throws \InvalidArgumentException if the event format is incorrect.
Chris@0:      */
Chris@0:     public static function batch(
Chris@0:         ClientInterface $client,
Chris@0:         $requests,
Chris@0:         array $options = []
Chris@0:     ) {
Chris@0:         $res = [];
Chris@0:         self::cmpCallback($options, 'fulfilled', $res);
Chris@0:         self::cmpCallback($options, 'rejected', $res);
Chris@0:         $pool = new static($client, $requests, $options);
Chris@0:         $pool->promise()->wait();
Chris@0:         ksort($res);
Chris@0: 
Chris@0:         return $res;
Chris@0:     }
Chris@0: 
Chris@0:     private static function cmpCallback(array &$options, $name, array &$results)
Chris@0:     {
Chris@0:         if (!isset($options[$name])) {
Chris@0:             $options[$name] = function ($v, $k) use (&$results) {
Chris@0:                 $results[$k] = $v;
Chris@0:             };
Chris@0:         } else {
Chris@0:             $currentFn = $options[$name];
Chris@0:             $options[$name] = function ($v, $k) use (&$results, $currentFn) {
Chris@0:                 $currentFn($v, $k);
Chris@0:                 $results[$k] = $v;
Chris@0:             };
Chris@0:         }
Chris@0:     }
Chris@0: }