*/ private array $pending = []; /** * @var list */ private array $waits = []; /** * @param (Closure(Tin): Tout) $operation */ public function __construct( private readonly Closure $operation, ) { } /** * Run the operation using the given `$input`, after all previous operations have completed. * * @param Tin $input * * @return Tout * * @see Sequence::cancel() */ public function waitFor(mixed $input): mixed { if ($this->ingoing) { $this->pending[] = $suspension = EventLoop::getSuspension(); $suspension->suspend(); } $this->ingoing = true; try { return ($this->operation)($input); } finally { $suspension = $this->pending[0] ?? null; if ($suspension !== null) { $this->pending = array_slice($this->pending, 1); $suspension->resume(); } else { foreach ($this->waits as $suspension) { $suspension->resume(); } $this->waits = []; $this->ingoing = false; } } } /** * Cancel all pending operations. * * Any pending operation will fail with the given exception. * * Future operations will continue execution as usual. */ public function cancel(Exception $exception): void { $suspensions = $this->pending; $this->pending = []; foreach ($suspensions as $suspension) { $suspension->throw($exception); } } /** * Get the number of operations pending execution. * * @return int<0, max> */ public function getPendingOperations(): int { /** @var int<0, max> */ return count($this->pending); } /** * Check if there's any operations pending execution. * * If this method returns `true`, it means future calls to `waitFor` will wait. */ public function hasPendingOperations(): bool { return $this->pending !== []; } /** * Check if the sequence has any ingoing operations. * * If this method returns `true`, it means future calls to `waitFor` will wait. * If this method returns `false`, it means future calls to `waitFor` will execute immediately. */ public function hasIngoingOperations(): bool { return $this->ingoing; } /** * Wait for all pending operations to finish execution. */ public function waitForPending(): void { if (!$this->ingoing) { return; } $suspension = EventLoop::getSuspension(); $this->waits[] = $suspension; $suspension->suspend(); } }