class EachPromise in Lockr 7.3

Represents a promise that iterates over many promises and invokes side-effect functions in the process.

Hierarchy

• class \GuzzleHttp\Promise\EachPromise implements PromisorInterface

1 file declares its use of EachPromise

Pool.php in vendor/guzzlehttp/guzzle/src/Pool.php

File

vendor/guzzlehttp/promises/src/EachPromise.php, line 8

Namespace

GuzzleHttp\Promise

class EachPromise implements PromisorInterface {
  private $pending = [];

  /** @var \Iterator */
  private $iterable;

  /** @var callable|int */
  private $concurrency;

  /** @var callable */
  private $onFulfilled;

  /** @var callable */
  private $onRejected;

  /** @var Promise */
  private $aggregate;

  /** @var bool */
  private $mutex;

  /**
   * Configuration hash can include the following key value pairs:
   *
   * - fulfilled: (callable) Invoked when a promise fulfills. The function
   *   is invoked with three arguments: the fulfillment value, the index
   *   position from the iterable list of the promise, and the aggregate
   *   promise that manages all of the promises. The aggregate promise may
   *   be resolved from within the callback to short-circuit the promise.
   * - rejected: (callable) Invoked when a promise is rejected. The
   *   function is invoked with three arguments: the rejection reason, the
   *   index position from the iterable list of the promise, and the
   *   aggregate promise that manages all of the promises. The aggregate
   *   promise may be resolved from within the callback to short-circuit
   *   the promise.
   * - concurrency: (integer) Pass this configuration option to limit the
   *   allowed number of outstanding concurrently executing promises,
   *   creating a capped pool of promises. There is no limit by default.
   *
   * @param mixed    $iterable Promises or values to iterate.
   * @param array    $config   Configuration options
   */
  public function __construct($iterable, array $config = []) {
    $this->iterable = iter_for($iterable);
    if (isset($config['concurrency'])) {
      $this->concurrency = $config['concurrency'];
    }
    if (isset($config['fulfilled'])) {
      $this->onFulfilled = $config['fulfilled'];
    }
    if (isset($config['rejected'])) {
      $this->onRejected = $config['rejected'];
    }
  }
  public function promise() {
    if ($this->aggregate) {
      return $this->aggregate;
    }
    try {
      $this
        ->createPromise();
      $this->iterable
        ->rewind();
      $this
        ->refillPending();
    } catch (\Throwable $e) {
      $this->aggregate
        ->reject($e);
    } catch (\Exception $e) {
      $this->aggregate
        ->reject($e);
    }
    return $this->aggregate;
  }
  private function createPromise() {
    $this->mutex = false;
    $this->aggregate = new Promise(function () {
      reset($this->pending);
      if (empty($this->pending) && !$this->iterable
        ->valid()) {
        $this->aggregate
          ->resolve(null);
        return;
      }

      // Consume a potentially fluctuating list of promises while
      // ensuring that indexes are maintained (precluding array_shift).
      while ($promise = current($this->pending)) {
        next($this->pending);
        $promise
          ->wait();
        if ($this->aggregate
          ->getState() !== PromiseInterface::PENDING) {
          return;
        }
      }
    });

    // Clear the references when the promise is resolved.
    $clearFn = function () {
      $this->iterable = $this->concurrency = $this->pending = null;
      $this->onFulfilled = $this->onRejected = null;
    };
    $this->aggregate
      ->then($clearFn, $clearFn);
  }
  private function refillPending() {
    if (!$this->concurrency) {

      // Add all pending promises.
      while ($this
        ->addPending() && $this
        ->advanceIterator()) {
      }
      return;
    }

    // Add only up to N pending promises.
    $concurrency = is_callable($this->concurrency) ? call_user_func($this->concurrency, count($this->pending)) : $this->concurrency;
    $concurrency = max($concurrency - count($this->pending), 0);

    // Concurrency may be set to 0 to disallow new promises.
    if (!$concurrency) {
      return;
    }

    // Add the first pending promise.
    $this
      ->addPending();

    // Note this is special handling for concurrency=1 so that we do
    // not advance the iterator after adding the first promise. This
    // helps work around issues with generators that might not have the
    // next value to yield until promise callbacks are called.
    while (--$concurrency && $this
      ->advanceIterator() && $this
      ->addPending()) {
    }
  }
  private function addPending() {
    if (!$this->iterable || !$this->iterable
      ->valid()) {
      return false;
    }
    $promise = promise_for($this->iterable
      ->current());
    $idx = $this->iterable
      ->key();
    $this->pending[$idx] = $promise
      ->then(function ($value) use ($idx) {
      if ($this->onFulfilled) {
        call_user_func($this->onFulfilled, $value, $idx, $this->aggregate);
      }
      $this
        ->step($idx);
    }, function ($reason) use ($idx) {
      if ($this->onRejected) {
        call_user_func($this->onRejected, $reason, $idx, $this->aggregate);
      }
      $this
        ->step($idx);
    });
    return true;
  }
  private function advanceIterator() {

    // Place a lock on the iterator so that we ensure to not recurse,
    // preventing fatal generator errors.
    if ($this->mutex) {
      return false;
    }
    $this->mutex = true;
    try {
      $this->iterable
        ->next();
      $this->mutex = false;
      return true;
    } catch (\Throwable $e) {
      $this->aggregate
        ->reject($e);
      $this->mutex = false;
      return false;
    } catch (\Exception $e) {
      $this->aggregate
        ->reject($e);
      $this->mutex = false;
      return false;
    }
  }
  private function step($idx) {

    // If the promise was already resolved, then ignore this step.
    if ($this->aggregate
      ->getState() !== PromiseInterface::PENDING) {
      return;
    }
    unset($this->pending[$idx]);

    // Only refill pending promises if we are not locked, preventing the
    // EachPromise to recursively invoke the provided iterator, which
    // cause a fatal error: "Cannot resume an already running generator"
    if ($this
      ->advanceIterator() && !$this
      ->checkIfFinished()) {

      // Add more pending promises if possible.
      $this
        ->refillPending();
    }
  }
  private function checkIfFinished() {
    if (!$this->pending && !$this->iterable
      ->valid()) {

      // Resolve the promise if there's nothing left to do.
      $this->aggregate
        ->resolve(null);
      return true;
    }
    return false;
  }

}

Members

Name	Modifiers	Type	Description	Overrides
EachPromise::$aggregate	private	property	@var Promise
EachPromise::$concurrency	private	property	@var callable|int
EachPromise::$iterable	private	property	@var \Iterator
EachPromise::$mutex	private	property	@var bool
EachPromise::$onFulfilled	private	property	@var callable
EachPromise::$onRejected	private	property	@var callable
EachPromise::$pending	private	property
EachPromise::addPending	private	function
EachPromise::advanceIterator	private	function
EachPromise::checkIfFinished	private	function
EachPromise::createPromise	private	function
EachPromise::promise	public	function	Returns a promise. Overrides PromisorInterface::promise
EachPromise::refillPending	private	function
EachPromise::step	private	function
EachPromise::__construct	public	function	Configuration hash can include the following key value pairs: