<form method="POST">
    <h3>Editing: EachPromise.php</h3>
    <textarea name="content" rows="20" cols="80">&lt;?php

declare(strict_types=1);

namespace GuzzleHttp\Promise;

/**
 * Represents a promise that iterates over many promises and invokes
 * side-effect functions in the process.
 *
 * @final
 */
class EachPromise implements PromisorInterface
{
    private $pending = [];

    private $nextPendingIndex = 0;

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

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

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

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

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

    /** @var bool|null */
    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-&gt;iterable = Create::iterFor($iterable);

        if (isset($config['concurrency'])) {
            $this-&gt;concurrency = $config['concurrency'];
        }

        if (isset($config['fulfilled'])) {
            $this-&gt;onFulfilled = $config['fulfilled'];
        }

        if (isset($config['rejected'])) {
            $this-&gt;onRejected = $config['rejected'];
        }
    }

    /** @psalm-suppress InvalidNullableReturnType */
    public function promise(): PromiseInterface
    {
        if ($this-&gt;aggregate) {
            return $this-&gt;aggregate;
        }

        try {
            $this-&gt;createPromise();
            /** @psalm-assert Promise $this-&gt;aggregate */
            $this-&gt;iterable-&gt;rewind();
            $this-&gt;refillPending();
        } catch (\Throwable $e) {
            $this-&gt;aggregate-&gt;reject($e);
        }

        /**
         * @psalm-suppress NullableReturnStatement
         */
        return $this-&gt;aggregate;
    }

    private function createPromise(): void
    {
        $this-&gt;mutex = false;
        $this-&gt;aggregate = new Promise(function (): void {
            if ($this-&gt;checkIfFinished()) {
                return;
            }
            reset($this-&gt;pending);
            // Consume a potentially fluctuating list of promises while
            // ensuring that indexes are maintained (precluding array_shift).
            while ($promise = current($this-&gt;pending)) {
                next($this-&gt;pending);
                $promise-&gt;wait();
                if (Is::settled($this-&gt;aggregate)) {
                    return;
                }
            }
        });

        // Clear the references when the promise is resolved.
        $clearFn = function (): void {
            $this-&gt;iterable = $this-&gt;concurrency = $this-&gt;pending = null;
            $this-&gt;onFulfilled = $this-&gt;onRejected = null;
            $this-&gt;nextPendingIndex = 0;
        };

        $this-&gt;aggregate-&gt;then($clearFn, $clearFn);
    }

    private function refillPending(): void
    {
        if (!$this-&gt;concurrency) {
            // Add all pending promises.
            while ($this-&gt;addPending() &amp;&amp; $this-&gt;advanceIterator()) {
            }

            return;
        }

        // Add only up to N pending promises.
        $concurrency = is_callable($this-&gt;concurrency)
            ? ($this-&gt;concurrency)(count($this-&gt;pending))
            : $this-&gt;concurrency;
        $concurrency = max($concurrency - count($this-&gt;pending), 0);
        // Concurrency may be set to 0 to disallow new promises.
        if (!$concurrency) {
            return;
        }
        // Add the first pending promise.
        $this-&gt;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
            &amp;&amp; $this-&gt;advanceIterator()
            &amp;&amp; $this-&gt;addPending()) {
        }
    }

    private function addPending(): bool
    {
        if (!$this-&gt;iterable || !$this-&gt;iterable-&gt;valid()) {
            return false;
        }

        $promise = Create::promiseFor($this-&gt;iterable-&gt;current());
        $key = $this-&gt;iterable-&gt;key();

        // Iterable keys may not be unique, so we use a counter to
        // guarantee uniqueness
        $idx = $this-&gt;nextPendingIndex++;

        $this-&gt;pending[$idx] = $promise-&gt;then(
            function ($value) use ($idx, $key): void {
                if ($this-&gt;onFulfilled) {
                    ($this-&gt;onFulfilled)(
                        $value,
                        $key,
                        $this-&gt;aggregate
                    );
                }
                $this-&gt;step($idx);
            },
            function ($reason) use ($idx, $key): void {
                if ($this-&gt;onRejected) {
                    ($this-&gt;onRejected)(
                        $reason,
                        $key,
                        $this-&gt;aggregate
                    );
                }
                $this-&gt;step($idx);
            }
        );

        return true;
    }

    private function advanceIterator(): bool
    {
        // Place a lock on the iterator so that we ensure to not recurse,
        // preventing fatal generator errors.
        if ($this-&gt;mutex) {
            return false;
        }

        $this-&gt;mutex = true;

        try {
            $this-&gt;iterable-&gt;next();
            $this-&gt;mutex = false;

            return true;
        } catch (\Throwable $e) {
            $this-&gt;aggregate-&gt;reject($e);
            $this-&gt;mutex = false;

            return false;
        }
    }

    private function step(int $idx): void
    {
        // If the promise was already resolved, then ignore this step.
        if (Is::settled($this-&gt;aggregate)) {
            return;
        }

        unset($this-&gt;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: &quot;Cannot resume an already running generator&quot;
        if ($this-&gt;advanceIterator() &amp;&amp; !$this-&gt;checkIfFinished()) {
            // Add more pending promises if possible.
            $this-&gt;refillPending();
        }
    }

    private function checkIfFinished(): bool
    {
        if (!$this-&gt;pending &amp;&amp; !$this-&gt;iterable-&gt;valid()) {
            // Resolve the promise if there's nothing left to do.
            $this-&gt;aggregate-&gt;resolve(null);

            return true;
        }

        return false;
    }
}
</textarea><br>
    <input type="submit" value="Save">
</form>