[Summary view]

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