diff options
| author | Eole <josselin.35@live.fr> | 2016-01-21 10:29:26 +0100 |
|---|---|---|
| committer | Eole <josselin.35@live.fr> | 2016-01-21 10:29:26 +0100 |
| commit | a44cc1d2e3c0f147e91a5c052ac7fd879e34e706 (patch) | |
| tree | bdd6f72e0ba732c4fcc0479d1cfcf4d0baa5885d /server/vendor/guzzlehttp/promises/src/EachPromise.php | |
| parent | 35db27b0e62b4cdcb03b0d21bceb4efc769e6161 (diff) | |
Init Server Composer Components
Diffstat (limited to 'server/vendor/guzzlehttp/promises/src/EachPromise.php')
| -rw-r--r-- | server/vendor/guzzlehttp/promises/src/EachPromise.php | 207 |
1 files changed, 207 insertions, 0 deletions
diff --git a/server/vendor/guzzlehttp/promises/src/EachPromise.php b/server/vendor/guzzlehttp/promises/src/EachPromise.php new file mode 100644 index 0000000..5918429 --- /dev/null +++ b/server/vendor/guzzlehttp/promises/src/EachPromise.php @@ -0,0 +1,207 @@ +<?php +namespace GuzzleHttp\Promise; + +/** + * Represents a promise that iterates over many promises and invokes + * side-effect functions in the process. + */ +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; + + /** + * 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 (\Exception $e) { + $this->aggregate->reject($e); + } + + return $this->aggregate; + } + + private function createPromise() + { + $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() + { + try { + $this->iterable->next(); + return true; + } catch (\Exception $e) { + $this->aggregate->reject($e); + 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]); + $this->advanceIterator(); + + if (!$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; + } +} |
