<?php

[0] Fix | Delete

[1] Fix | Delete

namespace GuzzleHttp\Promise;

[2] Fix | Delete

[3] Fix | Delete

/**

[4] Fix | Delete

* Represents a promise that iterates over many promises and invokes

[5] Fix | Delete

* side-effect functions in the process.

[6] Fix | Delete

*/

[7] Fix | Delete

class EachPromise implements PromisorInterface

[8] Fix | Delete

{

[9] Fix | Delete

private $pending = [];

[10] Fix | Delete

[11] Fix | Delete

private $nextPendingIndex = 0;

[12] Fix | Delete

[13] Fix | Delete

/** @var \Iterator|null */

[14] Fix | Delete

private $iterable;

[15] Fix | Delete

[16] Fix | Delete

/** @var callable|int|null */

[17] Fix | Delete

private $concurrency;

[18] Fix | Delete

[19] Fix | Delete

/** @var callable|null */

[20] Fix | Delete

private $onFulfilled;

[21] Fix | Delete

[22] Fix | Delete

/** @var callable|null */

[23] Fix | Delete

private $onRejected;

[24] Fix | Delete

[25] Fix | Delete

/** @var Promise|null */

[26] Fix | Delete

private $aggregate;

[27] Fix | Delete

[28] Fix | Delete

/** @var bool|null */

[29] Fix | Delete

private $mutex;

[30] Fix | Delete

[31] Fix | Delete

/**

[32] Fix | Delete

* Configuration hash can include the following key value pairs:

[33] Fix | Delete

*

[34] Fix | Delete

* - fulfilled: (callable) Invoked when a promise fulfills. The function

[35] Fix | Delete

* is invoked with three arguments: the fulfillment value, the index

[36] Fix | Delete

* position from the iterable list of the promise, and the aggregate

[37] Fix | Delete

* promise that manages all of the promises. The aggregate promise may

[38] Fix | Delete

* be resolved from within the callback to short-circuit the promise.

[39] Fix | Delete

* - rejected: (callable) Invoked when a promise is rejected. The

[40] Fix | Delete

* function is invoked with three arguments: the rejection reason, the

[41] Fix | Delete

* index position from the iterable list of the promise, and the

[42] Fix | Delete

* aggregate promise that manages all of the promises. The aggregate

[43] Fix | Delete

* promise may be resolved from within the callback to short-circuit

[44] Fix | Delete

* the promise.

[45] Fix | Delete

* - concurrency: (integer) Pass this configuration option to limit the

[46] Fix | Delete

* allowed number of outstanding concurrently executing promises,

[47] Fix | Delete

* creating a capped pool of promises. There is no limit by default.

[48] Fix | Delete

*

[49] Fix | Delete

* @param mixed $iterable Promises or values to iterate.

[50] Fix | Delete

* @param array $config Configuration options

[51] Fix | Delete

*/

[52] Fix | Delete

public function __construct($iterable, array $config = [])

[53] Fix | Delete

{

[54] Fix | Delete

$this->iterable = Create::iterFor($iterable);

[55] Fix | Delete

[56] Fix | Delete

if (isset($config['concurrency'])) {

[57] Fix | Delete

$this->concurrency = $config['concurrency'];

[58] Fix | Delete

}

[59] Fix | Delete

[60] Fix | Delete

if (isset($config['fulfilled'])) {

[61] Fix | Delete

$this->onFulfilled = $config['fulfilled'];

[62] Fix | Delete

}

[63] Fix | Delete

[64] Fix | Delete

if (isset($config['rejected'])) {

[65] Fix | Delete

$this->onRejected = $config['rejected'];

[66] Fix | Delete

}

[67] Fix | Delete

}

[68] Fix | Delete

[69] Fix | Delete

/** @psalm-suppress InvalidNullableReturnType */

[70] Fix | Delete

public function promise()

[71] Fix | Delete

{

[72] Fix | Delete

if ($this->aggregate) {

[73] Fix | Delete

return $this->aggregate;

[74] Fix | Delete

}

[75] Fix | Delete

[76] Fix | Delete

try {

[77] Fix | Delete

$this->createPromise();

[78] Fix | Delete

/** @psalm-assert Promise $this->aggregate */

[79] Fix | Delete

$this->iterable->rewind();

[80] Fix | Delete

$this->refillPending();

[81] Fix | Delete

} catch (\Throwable $e) {

[82] Fix | Delete

/**

[83] Fix | Delete

* @psalm-suppress NullReference

[84] Fix | Delete

* @phpstan-ignore-next-line

[85] Fix | Delete

*/

[86] Fix | Delete

$this->aggregate->reject($e);

[87] Fix | Delete

} catch (\Exception $e) {

[88] Fix | Delete

/**

[89] Fix | Delete

* @psalm-suppress NullReference

[90] Fix | Delete

* @phpstan-ignore-next-line

[91] Fix | Delete

*/

[92] Fix | Delete

$this->aggregate->reject($e);

[93] Fix | Delete

}

[94] Fix | Delete

[95] Fix | Delete

/**

[96] Fix | Delete

* @psalm-suppress NullableReturnStatement

[97] Fix | Delete

* @phpstan-ignore-next-line

[98] Fix | Delete

*/

[99] Fix | Delete

return $this->aggregate;

[100] Fix | Delete

}

[101] Fix | Delete

[102] Fix | Delete

private function createPromise()

[103] Fix | Delete

{

[104] Fix | Delete

$this->mutex = false;

[105] Fix | Delete

$this->aggregate = new Promise(function () {

[106] Fix | Delete

if ($this->checkIfFinished()) {

[107] Fix | Delete

return;

[108] Fix | Delete

}

[109] Fix | Delete

reset($this->pending);

[110] Fix | Delete

// Consume a potentially fluctuating list of promises while

[111] Fix | Delete

// ensuring that indexes are maintained (precluding array_shift).

[112] Fix | Delete

while ($promise = current($this->pending)) {

[113] Fix | Delete

next($this->pending);

[114] Fix | Delete

$promise->wait();

[115] Fix | Delete

if (Is::settled($this->aggregate)) {

[116] Fix | Delete

return;

[117] Fix | Delete

}

[118] Fix | Delete

}

[119] Fix | Delete

});

[120] Fix | Delete

[121] Fix | Delete

// Clear the references when the promise is resolved.

[122] Fix | Delete

$clearFn = function () {

[123] Fix | Delete

$this->iterable = $this->concurrency = $this->pending = null;

[124] Fix | Delete

$this->onFulfilled = $this->onRejected = null;

[125] Fix | Delete

$this->nextPendingIndex = 0;

[126] Fix | Delete

};

[127] Fix | Delete

[128] Fix | Delete

$this->aggregate->then($clearFn, $clearFn);

[129] Fix | Delete

}

[130] Fix | Delete

[131] Fix | Delete

private function refillPending()

[132] Fix | Delete

{

[133] Fix | Delete

if (!$this->concurrency) {

[134] Fix | Delete

// Add all pending promises.

[135] Fix | Delete

while ($this->addPending() && $this->advanceIterator());

[136] Fix | Delete

return;

[137] Fix | Delete

}

[138] Fix | Delete

[139] Fix | Delete

// Add only up to N pending promises.

[140] Fix | Delete

$concurrency = is_callable($this->concurrency)

[141] Fix | Delete

? call_user_func($this->concurrency, count($this->pending))

[142] Fix | Delete

: $this->concurrency;

[143] Fix | Delete

$concurrency = max($concurrency - count($this->pending), 0);

[144] Fix | Delete

// Concurrency may be set to 0 to disallow new promises.

[145] Fix | Delete

if (!$concurrency) {

[146] Fix | Delete

return;

[147] Fix | Delete

}

[148] Fix | Delete

// Add the first pending promise.

[149] Fix | Delete

$this->addPending();

[150] Fix | Delete

// Note this is special handling for concurrency=1 so that we do

[151] Fix | Delete

// not advance the iterator after adding the first promise. This

[152] Fix | Delete

// helps work around issues with generators that might not have the

[153] Fix | Delete

// next value to yield until promise callbacks are called.

[154] Fix | Delete

while (--$concurrency

[155] Fix | Delete

&& $this->advanceIterator()

[156] Fix | Delete

&& $this->addPending());

[157] Fix | Delete

}

[158] Fix | Delete

[159] Fix | Delete

private function addPending()

[160] Fix | Delete

{

[161] Fix | Delete

if (!$this->iterable || !$this->iterable->valid()) {

[162] Fix | Delete

return false;

[163] Fix | Delete

}

[164] Fix | Delete

[165] Fix | Delete

$promise = Create::promiseFor($this->iterable->current());

[166] Fix | Delete

$key = $this->iterable->key();

[167] Fix | Delete

[168] Fix | Delete

// Iterable keys may not be unique, so we use a counter to

[169] Fix | Delete

// guarantee uniqueness

[170] Fix | Delete

$idx = $this->nextPendingIndex++;

[171] Fix | Delete

[172] Fix | Delete

$this->pending[$idx] = $promise->then(

[173] Fix | Delete

function ($value) use ($idx, $key) {

[174] Fix | Delete

if ($this->onFulfilled) {

[175] Fix | Delete

call_user_func(

[176] Fix | Delete

$this->onFulfilled,

[177] Fix | Delete

$value,

[178] Fix | Delete

$key,

[179] Fix | Delete

$this->aggregate

[180] Fix | Delete

);

[181] Fix | Delete

}

[182] Fix | Delete

$this->step($idx);

[183] Fix | Delete

},

[184] Fix | Delete

function ($reason) use ($idx, $key) {

[185] Fix | Delete

if ($this->onRejected) {

[186] Fix | Delete

call_user_func(

[187] Fix | Delete

$this->onRejected,

[188] Fix | Delete

$reason,

[189] Fix | Delete

$key,

[190] Fix | Delete

$this->aggregate

[191] Fix | Delete

);

[192] Fix | Delete

}

[193] Fix | Delete

$this->step($idx);

[194] Fix | Delete

}

[195] Fix | Delete

);

[196] Fix | Delete

[197] Fix | Delete

return true;

[198] Fix | Delete

}

[199] Fix | Delete

[200] Fix | Delete

private function advanceIterator()

[201] Fix | Delete

{

[202] Fix | Delete

// Place a lock on the iterator so that we ensure to not recurse,

[203] Fix | Delete

// preventing fatal generator errors.

[204] Fix | Delete

if ($this->mutex) {

[205] Fix | Delete

return false;

[206] Fix | Delete

}

[207] Fix | Delete

[208] Fix | Delete

$this->mutex = true;

[209] Fix | Delete

[210] Fix | Delete

try {

[211] Fix | Delete

$this->iterable->next();

[212] Fix | Delete

$this->mutex = false;

[213] Fix | Delete

return true;

[214] Fix | Delete

} catch (\Throwable $e) {

[215] Fix | Delete

$this->aggregate->reject($e);

[216] Fix | Delete

$this->mutex = false;

[217] Fix | Delete

return false;

[218] Fix | Delete

} catch (\Exception $e) {

[219] Fix | Delete

$this->aggregate->reject($e);

[220] Fix | Delete

$this->mutex = false;

[221] Fix | Delete

return false;

[222] Fix | Delete

}

[223] Fix | Delete

}

[224] Fix | Delete

[225] Fix | Delete

private function step($idx)

[226] Fix | Delete

{

[227] Fix | Delete

// If the promise was already resolved, then ignore this step.

[228] Fix | Delete

if (Is::settled($this->aggregate)) {

[229] Fix | Delete

return;

[230] Fix | Delete

}

[231] Fix | Delete

[232] Fix | Delete

unset($this->pending[$idx]);

[233] Fix | Delete

[234] Fix | Delete

// Only refill pending promises if we are not locked, preventing the

[235] Fix | Delete

// EachPromise to recursively invoke the provided iterator, which

[236] Fix | Delete

// cause a fatal error: "Cannot resume an already running generator"

[237] Fix | Delete

if ($this->advanceIterator() && !$this->checkIfFinished()) {

[238] Fix | Delete

// Add more pending promises if possible.

[239] Fix | Delete

$this->refillPending();

[240] Fix | Delete

}

[241] Fix | Delete

}

[242] Fix | Delete

[243] Fix | Delete

private function checkIfFinished()

[244] Fix | Delete

{

[245] Fix | Delete

if (!$this->pending && !$this->iterable->valid()) {

[246] Fix | Delete

// Resolve the promise if there's nothing left to do.

[247] Fix | Delete

$this->aggregate->resolve(null);

[248] Fix | Delete

return true;

[249] Fix | Delete

}

[250] Fix | Delete

[251] Fix | Delete

return false;

[252] Fix | Delete

}

[253] Fix | Delete

}

[254] Fix | Delete

[255] Fix | Delete