plausible-analytics / src / Client / lib / Lib / GuzzleHttp / Promise / EachPromise.php

EachPromise.php in Plausible Analytics trunk, at src/Client/lib/Lib/GuzzleHttp/Promise/EachPromise.php

251 lines 7.5 KB

Raw Download Zip

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Plausible\Analytics\WP\Client\Lib\GuzzleHttp\Promise;
6
7	/**
8	* Represents a promise that iterates over many promises and invokes
9	* side-effect functions in the process.
10	*
11	* @final
12	*/
13	class EachPromise implements PromisorInterface
14	{
15	private $pending = [];
16
17	private $nextPendingIndex = 0;
18
19	/** @var \Iterator\|null */
20	private $iterable;
21
22	/** @var callable\|int\|null */
23	private $concurrency;
24
25	/** @var callable\|null */
26	private $onFulfilled;
27
28	/** @var callable\|null */
29	private $onRejected;
30
31	/** @var Promise\|null */
32	private $aggregate;
33
34	/** @var bool\|null */
35	private $mutex;
36
37	/**
38	* Configuration hash can include the following key value pairs:
39	*
40	* - fulfilled: (callable) Invoked when a promise fulfills. The function
41	* is invoked with three arguments: the fulfillment value, the index
42	* position from the iterable list of the promise, and the aggregate
43	* promise that manages all of the promises. The aggregate promise may
44	* be resolved from within the callback to short-circuit the promise.
45	* - rejected: (callable) Invoked when a promise is rejected. The
46	* function is invoked with three arguments: the rejection reason, the
47	* index position from the iterable list of the promise, and the
48	* aggregate promise that manages all of the promises. The aggregate
49	* promise may be resolved from within the callback to short-circuit
50	* the promise.
51	* - concurrency: (integer) Pass this configuration option to limit the
52	* allowed number of outstanding concurrently executing promises,
53	* creating a capped pool of promises. There is no limit by default.
54	*
55	* @param mixed $iterable Promises or values to iterate.
56	* @param array $config Configuration options
57	*/
58	public function __construct($iterable, array $config = [])
59	{
60	$this->iterable = Create::iterFor($iterable);
61
62	if (isset($config['concurrency'])) {
63	$this->concurrency = $config['concurrency'];
64	}
65
66	if (isset($config['fulfilled'])) {
67	$this->onFulfilled = $config['fulfilled'];
68	}
69
70	if (isset($config['rejected'])) {
71	$this->onRejected = $config['rejected'];
72	}
73	}
74
75	/** @psalm-suppress InvalidNullableReturnType */
76	public function promise(): PromiseInterface
77	{
78	if ($this->aggregate) {
79	return $this->aggregate;
80	}
81
82	try {
83	$this->createPromise();
84	/** @psalm-assert Promise $this->aggregate */
85	$this->iterable->rewind();
86	$this->refillPending();
87	} catch (\Throwable $e) {
88	$this->aggregate->reject($e);
89	}
90
91	/**
92	* @psalm-suppress NullableReturnStatement
93	*/
94	return $this->aggregate;
95	}
96
97	private function createPromise(): void
98	{
99	$this->mutex = false;
100	$this->aggregate = new Promise(function (): void {
101	if ($this->checkIfFinished()) {
102	return;
103	}
104	reset($this->pending);
105	// Consume a potentially fluctuating list of promises while
106	// ensuring that indexes are maintained (precluding array_shift).
107	while ($promise = current($this->pending)) {
108	next($this->pending);
109	$promise->wait();
110	if (Is::settled($this->aggregate)) {
111	return;
112	}
113	}
114	});
115
116	// Clear the references when the promise is resolved.
117	$clearFn = function (): void {
118	$this->iterable = $this->concurrency = $this->pending = null;
119	$this->onFulfilled = $this->onRejected = null;
120	$this->nextPendingIndex = 0;
121	};
122
123	$this->aggregate->then($clearFn, $clearFn);
124	}
125
126	private function refillPending(): void
127	{
128	if (!$this->concurrency) {
129	// Add all pending promises.
130	while ($this->addPending() && $this->advanceIterator()) {
131	}
132
133	return;
134	}
135
136	// Add only up to N pending promises.
137	$concurrency = is_callable($this->concurrency)
138	? call_user_func($this->concurrency, count($this->pending))
139	: $this->concurrency;
140	$concurrency = max($concurrency - count($this->pending), 0);
141	// Concurrency may be set to 0 to disallow new promises.
142	if (!$concurrency) {
143	return;
144	}
145	// Add the first pending promise.
146	$this->addPending();
147	// Note this is special handling for concurrency=1 so that we do
148	// not advance the iterator after adding the first promise. This
149	// helps work around issues with generators that might not have the
150	// next value to yield until promise callbacks are called.
151	while (--$concurrency
152	&& $this->advanceIterator()
153	&& $this->addPending()) {
154	}
155	}
156
157	private function addPending(): bool
158	{
159	if (!$this->iterable \|\| !$this->iterable->valid()) {
160	return false;
161	}
162
163	$promise = Create::promiseFor($this->iterable->current());
164	$key = $this->iterable->key();
165
166	// Iterable keys may not be unique, so we use a counter to
167	// guarantee uniqueness
168	$idx = $this->nextPendingIndex++;
169
170	$this->pending[$idx] = $promise->then(
171	function ($value) use ($idx, $key): void {
172	if ($this->onFulfilled) {
173	call_user_func(
174	$this->onFulfilled,
175	$value,
176	$key,
177	$this->aggregate
178	);
179	}
180	$this->step($idx);
181	},
182	function ($reason) use ($idx, $key): void {
183	if ($this->onRejected) {
184	call_user_func(
185	$this->onRejected,
186	$reason,
187	$key,
188	$this->aggregate
189	);
190	}
191	$this->step($idx);
192	}
193	);
194
195	return true;
196	}
197
198	private function advanceIterator(): bool
199	{
200	// Place a lock on the iterator so that we ensure to not recurse,
201	// preventing fatal generator errors.
202	if ($this->mutex) {
203	return false;
204	}
205
206	$this->mutex = true;
207
208	try {
209	$this->iterable->next();
210	$this->mutex = false;
211
212	return true;
213	} catch (\Throwable $e) {
214	$this->aggregate->reject($e);
215	$this->mutex = false;
216
217	return false;
218	}
219	}
220
221	private function step(int $idx): void
222	{
223	// If the promise was already resolved, then ignore this step.
224	if (Is::settled($this->aggregate)) {
225	return;
226	}
227
228	unset($this->pending[$idx]);
229
230	// Only refill pending promises if we are not locked, preventing the
231	// EachPromise to recursively invoke the provided iterator, which
232	// cause a fatal error: "Cannot resume an already running generator"
233	if ($this->advanceIterator() && !$this->checkIfFinished()) {
234	// Add more pending promises if possible.
235	$this->refillPending();
236	}
237	}
238
239	private function checkIfFinished(): bool
240	{
241	if (!$this->pending && !$this->iterable->valid()) {
242	// Resolve the promise if there's nothing left to do.
243	$this->aggregate->resolve(null);
244
245	return true;
246	}
247
248	return false;
249	}
250	}
251