2 namespace GuzzleHttp\Promise;
5 * Represents a promise that iterates over many promises and invokes
6 * side-effect functions in the process.
8 class EachPromise implements PromisorInterface
10 private $pending = [];
15 /** @var callable|int */
31 * Configuration hash can include the following key value pairs:
33 * - fulfilled: (callable) Invoked when a promise fulfills. The function
34 * is invoked with three arguments: the fulfillment value, the index
35 * position from the iterable list of the promise, and the aggregate
36 * promise that manages all of the promises. The aggregate promise may
37 * be resolved from within the callback to short-circuit the promise.
38 * - rejected: (callable) Invoked when a promise is rejected. The
39 * function is invoked with three arguments: the rejection reason, the
40 * index position from the iterable list of the promise, and the
41 * aggregate promise that manages all of the promises. The aggregate
42 * promise may be resolved from within the callback to short-circuit
44 * - concurrency: (integer) Pass this configuration option to limit the
45 * allowed number of outstanding concurrently executing promises,
46 * creating a capped pool of promises. There is no limit by default.
48 * @param mixed $iterable Promises or values to iterate.
49 * @param array $config Configuration options
51 public function __construct($iterable, array $config = [])
53 $this->iterable = iter_for($iterable);
55 if (isset($config['concurrency'])) {
56 $this->concurrency = $config['concurrency'];
59 if (isset($config['fulfilled'])) {
60 $this->onFulfilled = $config['fulfilled'];
63 if (isset($config['rejected'])) {
64 $this->onRejected = $config['rejected'];
68 public function promise()
70 if ($this->aggregate) {
71 return $this->aggregate;
75 $this->createPromise();
76 $this->iterable->rewind();
77 $this->refillPending();
78 } catch (\Throwable $e) {
79 $this->aggregate->reject($e);
80 } catch (\Exception $e) {
81 $this->aggregate->reject($e);
84 return $this->aggregate;
87 private function createPromise()
90 $this->aggregate = new Promise(function () {
91 reset($this->pending);
92 if (empty($this->pending) && !$this->iterable->valid()) {
93 $this->aggregate->resolve(null);
97 // Consume a potentially fluctuating list of promises while
98 // ensuring that indexes are maintained (precluding array_shift).
99 while ($promise = current($this->pending)) {
100 next($this->pending);
102 if ($this->aggregate->getState() !== PromiseInterface::PENDING) {
108 // Clear the references when the promise is resolved.
109 $clearFn = function () {
110 $this->iterable = $this->concurrency = $this->pending = null;
111 $this->onFulfilled = $this->onRejected = null;
114 $this->aggregate->then($clearFn, $clearFn);
117 private function refillPending()
119 if (!$this->concurrency) {
120 // Add all pending promises.
121 while ($this->addPending() && $this->advanceIterator());
125 // Add only up to N pending promises.
126 $concurrency = is_callable($this->concurrency)
127 ? call_user_func($this->concurrency, count($this->pending))
128 : $this->concurrency;
129 $concurrency = max($concurrency - count($this->pending), 0);
130 // Concurrency may be set to 0 to disallow new promises.
134 // Add the first pending promise.
136 // Note this is special handling for concurrency=1 so that we do
137 // not advance the iterator after adding the first promise. This
138 // helps work around issues with generators that might not have the
139 // next value to yield until promise callbacks are called.
140 while (--$concurrency
141 && $this->advanceIterator()
142 && $this->addPending());
145 private function addPending()
147 if (!$this->iterable || !$this->iterable->valid()) {
151 $promise = promise_for($this->iterable->current());
152 $idx = $this->iterable->key();
154 $this->pending[$idx] = $promise->then(
155 function ($value) use ($idx) {
156 if ($this->onFulfilled) {
158 $this->onFulfilled, $value, $idx, $this->aggregate
163 function ($reason) use ($idx) {
164 if ($this->onRejected) {
166 $this->onRejected, $reason, $idx, $this->aggregate
176 private function advanceIterator()
178 // Place a lock on the iterator so that we ensure to not recurse,
179 // preventing fatal generator errors.
187 $this->iterable->next();
188 $this->mutex = false;
190 } catch (\Throwable $e) {
191 $this->aggregate->reject($e);
192 $this->mutex = false;
194 } catch (\Exception $e) {
195 $this->aggregate->reject($e);
196 $this->mutex = false;
201 private function step($idx)
203 // If the promise was already resolved, then ignore this step.
204 if ($this->aggregate->getState() !== PromiseInterface::PENDING) {
208 unset($this->pending[$idx]);
210 // Only refill pending promises if we are not locked, preventing the
211 // EachPromise to recursively invoke the provided iterator, which
212 // cause a fatal error: "Cannot resume an already running generator"
213 if ($this->advanceIterator() && !$this->checkIfFinished()) {
214 // Add more pending promises if possible.
215 $this->refillPending();
219 private function checkIfFinished()
221 if (!$this->pending && !$this->iterable->valid()) {
222 // Resolve the promise if there's nothing left to do.
223 $this->aggregate->resolve(null);