[Summary view] [Print] [Text view]

   1  <?php
   2  namespace GuzzleHttp;
   3  
   4  use GuzzleHttp\Promise\EachPromise;
   5  use GuzzleHttp\Promise\PromiseInterface;
   6  use GuzzleHttp\Promise\PromisorInterface;
   7  use 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   */
  20  class 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  }