.github/workflows/ci.yml000064400000004745144760120120011266 0ustar00name: Continuous Integration on: - push - pull_request jobs: tests: strategy: matrix: include: - operating-system: 'ubuntu-latest' php-version: '7.1' - operating-system: 'ubuntu-latest' php-version: '7.2' - operating-system: 'ubuntu-latest' php-version: '7.3' - operating-system: 'ubuntu-latest' php-version: '7.4' - operating-system: 'ubuntu-latest' php-version: '8.0' composer-flags: '--ignore-platform-req=php' - operating-system: 'windows-latest' php-version: '7.4' job-description: 'on Windows' - operating-system: 'macos-latest' php-version: '7.4' job-description: 'on macOS' name: PHP ${{ matrix.php-version }} ${{ matrix.job-description }} runs-on: ${{ matrix.operating-system }} steps: - name: Set git to use LF run: | git config --global core.autocrlf false git config --global core.eol lf - name: Checkout code uses: actions/checkout@v2 - name: Setup PHP uses: shivammathur/setup-php@v2 with: php-version: ${{ matrix.php-version }} - name: Get Composer cache directory id: composer-cache run: echo "::set-output name=dir::$(composer config cache-dir)" - name: Cache dependencies uses: actions/cache@v2 with: path: ${{ steps.composer-cache.outputs.dir }} key: composer-${{ runner.os }}-${{ matrix.php-version }}-${{ hashFiles('**/composer.*') }}-${{ matrix.composer-flags }} restore-keys: | composer-${{ runner.os }}-${{ matrix.php-version }}-${{ hashFiles('**/composer.*') }}- composer-${{ runner.os }}-${{ matrix.php-version }}- composer-${{ runner.os }}- composer- - name: Install dependencies uses: nick-invision/retry@v2 with: timeout_minutes: 5 max_attempts: 5 retry_wait_seconds: 30 command: | composer update --optimize-autoloader --no-interaction --no-progress ${{ matrix.composer-flags }} composer info -D - name: Run tests env: PHP_CS_FIXER_IGNORE_ENV: 1 run: vendor/bin/phpunit ${{ matrix.phpunit-flags }} - name: Run style fixer env: PHP_CS_FIXER_IGNORE_ENV: 1 run: vendor/bin/php-cs-fixer --diff --dry-run -v fix LICENSE000064400000002161144760120120005546 0ustar00 The MIT License (MIT) Copyright (c) 2015-2019 amphp Copyright (c) 2016 PHP Asynchronous Interoperability Group Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. composer.json000064400000003417144760120120007270 0ustar00{ "name": "amphp/amp", "homepage": "http://amphp.org/amp", "description": "A non-blocking concurrency framework for PHP applications.", "keywords": [ "async", "asynchronous", "concurrency", "promise", "awaitable", "future", "non-blocking", "event", "event-loop" ], "license": "MIT", "authors": [ { "name": "Daniel Lowrey", "email": "rdlowrey@php.net" }, { "name": "Aaron Piotrowski", "email": "aaron@trowski.com" }, { "name": "Bob Weinand", "email": "bobwei9@hotmail.com" }, { "name": "Niklas Keller", "email": "me@kelunik.com" } ], "require": { "php": ">=7.1" }, "require-dev": { "ext-json": "*", "amphp/phpunit-util": "^1", "amphp/php-cs-fixer-config": "dev-master", "react/promise": "^2", "phpunit/phpunit": "^7 | ^8 | ^9", "psalm/phar": "^3.11@dev", "jetbrains/phpstorm-stubs": "^2019.3" }, "autoload": { "psr-4": { "Amp\\": "lib" }, "files": [ "lib/functions.php", "lib/Internal/functions.php" ] }, "autoload-dev": { "psr-4": { "Amp\\Test\\": "test" } }, "support": { "issues": "https://github.com/amphp/amp/issues", "irc": "irc://irc.freenode.org/amphp" }, "extra": { "branch-alias": { "dev-master": "2.x-dev" } }, "scripts": { "test": "@php -dzend.assertions=1 -dassert.exception=1 ./vendor/bin/phpunit", "code-style": "@php ./vendor/bin/php-cs-fixer fix" } } lib/CallableMaker.php000064400000005224144760120120010502 0ustar00getMethod($method); } return self::$__reflectionMethods[$method]->getClosure($this); } /** * Creates a callable from a protected or private static method that may be invoked by methods requiring a * publicly invokable callback. * * @param string $method Static method name. * * @return callable * * @psalm-suppress MixedInferredReturnType */ private static function callableFromStaticMethod(string $method): callable { if (!isset(self::$__reflectionMethods[$method])) { if (self::$__reflectionClass === null) { self::$__reflectionClass = new \ReflectionClass(self::class); } self::$__reflectionMethods[$method] = self::$__reflectionClass->getMethod($method); } return self::$__reflectionMethods[$method]->getClosure(); } } } else { /** @psalm-suppress DuplicateClass */ trait CallableMaker { /** * @deprecated Use \Closure::fromCallable() instead of this method in PHP 7.1. */ private function callableFromInstanceMethod(string $method): callable { return \Closure::fromCallable([$this, $method]); } /** * @deprecated Use \Closure::fromCallable() instead of this method in PHP 7.1. */ private static function callableFromStaticMethod(string $method): callable { return \Closure::fromCallable([self::class, $method]); } } } // @codeCoverageIgnoreEnd lib/CancellationToken.php000064400000002757144760120120011430 0ustar00getToken(); * * $response = yield $httpClient->request("https://example.com/stream", $token); * $responseBody = $response->getBody(); * * while (($chunk = yield $response->read()) !== null) { * // consume $chunk * * if ($noLongerInterested) { * $cancellationTokenSource->cancel(); * break; * } * } * ``` * * @see CancellationToken * @see CancelledException */ final class CancellationTokenSource { /** @var CancellationToken */ private $token; /** @var callable|null */ private $onCancel; public function __construct() { $onCancel = null; $this->token = new class($onCancel) implements CancellationToken { /** @var string */ private $nextId = "a"; /** @var callable[] */ private $callbacks = []; /** @var \Throwable|null */ private $exception; /** * @param mixed $onCancel * @param-out callable $onCancel */ public function __construct(&$onCancel) { /** @psalm-suppress MissingClosureReturnType We still support PHP 7.0 */ $onCancel = function (\Throwable $exception) { $this->exception = $exception; $callbacks = $this->callbacks; $this->callbacks = []; foreach ($callbacks as $callback) { $this->invokeCallback($callback); } }; } /** * @param callable $callback * * @return void */ private function invokeCallback(callable $callback) { // No type declaration to prevent exception outside the try! try { /** @var mixed $result */ $result = $callback($this->exception); if ($result instanceof \Generator) { /** @psalm-var \Generator $result */ $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { Loop::defer(static function () use ($exception) { throw $exception; }); } } public function subscribe(callable $callback): string { $id = $this->nextId++; if ($this->exception) { $this->invokeCallback($callback); } else { $this->callbacks[$id] = $callback; } return $id; } public function unsubscribe(string $id) { unset($this->callbacks[$id]); } public function isRequested(): bool { return isset($this->exception); } public function throwIfRequested() { if (isset($this->exception)) { throw $this->exception; } } }; $this->onCancel = $onCancel; } public function getToken(): CancellationToken { return $this->token; } /** * @param \Throwable|null $previous Exception to be used as the previous exception to CancelledException. * * @return void */ public function cancel(\Throwable $previous = null) { if ($this->onCancel === null) { return; } $onCancel = $this->onCancel; $this->onCancel = null; $onCancel(new CancelledException($previous)); } } lib/CancelledException.php000064400000000524144760120120011552 0ustar00subscribe(function (CancelledException $exception) { $this->exception = $exception; $callbacks = $this->callbacks; $this->callbacks = []; foreach ($callbacks as $callback) { asyncCall($callback, $this->exception); } }); $this->tokens[] = [$token, $id]; } } public function __destruct() { foreach ($this->tokens as list($token, $id)) { /** @var CancellationToken $token */ $token->unsubscribe($id); } } /** @inheritdoc */ public function subscribe(callable $callback): string { $id = $this->nextId++; if ($this->exception) { asyncCall($callback, $this->exception); } else { $this->callbacks[$id] = $callback; } return $id; } /** @inheritdoc */ public function unsubscribe(string $id) { unset($this->callbacks[$id]); } /** @inheritdoc */ public function isRequested(): bool { foreach ($this->tokens as list($token)) { if ($token->isRequested()) { return true; } } return false; } /** @inheritdoc */ public function throwIfRequested() { foreach ($this->tokens as list($token)) { $token->throwIfRequested(); } } } lib/Coroutine.php000064400000012562144760120120007775 0ustar00 */ final class Coroutine implements Promise { use Internal\Placeholder; /** * Attempts to transform the non-promise yielded from the generator into a promise, otherwise returns an instance * `Amp\Failure` failed with an instance of `Amp\InvalidYieldError`. * * @param mixed $yielded Non-promise yielded from generator. * @param \Generator $generator No type for performance, we already know the type. * * @return Promise */ private static function transform($yielded, $generator): Promise { $exception = null; // initialize here, see https://github.com/vimeo/psalm/issues/2951 try { if (\is_array($yielded)) { return Promise\all($yielded); } if ($yielded instanceof ReactPromise) { return Promise\adapt($yielded); } // No match, continue to returning Failure below. } catch (\Throwable $exception) { // Conversion to promise failed, fall-through to returning Failure below. } return new Failure(new InvalidYieldError( $generator, \sprintf( "Unexpected yield; Expected an instance of %s or %s or an array of such instances", Promise::class, ReactPromise::class ), $exception )); } /** * @param \Generator $generator * @psalm-param \Generator,mixed,Promise|ReactPromise|TReturn> $generator */ public function __construct(\Generator $generator) { try { $yielded = $generator->current(); if (!$yielded instanceof Promise) { if (!$generator->valid()) { $this->resolve($generator->getReturn()); return; } $yielded = self::transform($yielded, $generator); } } catch (\Throwable $exception) { $this->fail($exception); return; } /** * @param \Throwable|null $e Exception to be thrown into the generator. * @param mixed $v Value to be sent into the generator. * * @return void * * @psalm-suppress MissingClosureParamType * @psalm-suppress MissingClosureReturnType */ $onResolve = function (\Throwable $e = null, $v) use ($generator, &$onResolve) { /** @var bool $immediate Used to control iterative coroutine continuation. */ static $immediate = true; /** @var \Throwable|null $exception Promise failure reason when executing next coroutine step, null at all other times. */ static $exception; /** @var mixed $value Promise success value when executing next coroutine step, null at all other times. */ static $value; $exception = $e; /** @psalm-suppress MixedAssignment */ $value = $v; if (!$immediate) { $immediate = true; return; } try { try { do { if ($exception) { // Throw exception at current execution point. $yielded = $generator->throw($exception); } else { // Send the new value and execute to next yield statement. $yielded = $generator->send($value); } if (!$yielded instanceof Promise) { if (!$generator->valid()) { $this->resolve($generator->getReturn()); $onResolve = null; return; } $yielded = self::transform($yielded, $generator); } $immediate = false; $yielded->onResolve($onResolve); } while ($immediate); $immediate = true; } catch (\Throwable $exception) { $this->fail($exception); $onResolve = null; } finally { $exception = null; $value = null; } } catch (\Throwable $e) { Loop::defer(static function () use ($e) { throw $e; }); } }; try { $yielded->onResolve($onResolve); unset($generator, $yielded, $onResolve); } catch (\Throwable $e) { Loop::defer(static function () use ($e) { throw $e; }); } } } lib/Deferred.php000064400000003417144760120120007545 0ustar00 Has public resolve and fail methods. */ private $resolver; /** @var Promise Hides placeholder methods */ private $promise; public function __construct() { $this->resolver = new class implements Promise { use Internal\Placeholder { resolve as public; fail as public; isResolved as public; } }; $this->promise = new Internal\PrivatePromise($this->resolver); } /** * @return Promise */ public function promise(): Promise { return $this->promise; } /** * Fulfill the promise with the given value. * * @param mixed $value * * @psalm-param TValue|Promise $value * * @return void */ public function resolve($value = null) { /** @psalm-suppress UndefinedInterfaceMethod */ $this->resolver->resolve($value); } /** * Fails the promise the the given reason. * * @param \Throwable $reason * * @return void */ public function fail(\Throwable $reason) { /** @psalm-suppress UndefinedInterfaceMethod */ $this->resolver->fail($reason); } /** * @return bool True if the promise has been resolved. */ public function isResolved(): bool { return $this->resolver->isResolved(); } } lib/Delayed.php000064400000002640144760120120007371 0ustar00 */ final class Delayed implements Promise { use Internal\Placeholder; /** @var string|null Event loop watcher identifier. */ private $watcher; /** * @param int $time Milliseconds before succeeding the promise. * @param TReturn $value Succeed the promise with this value. */ public function __construct(int $time, $value = null) { $this->watcher = Loop::delay($time, function () use ($value) { $this->watcher = null; $this->resolve($value); }); } /** * References the internal watcher in the event loop, keeping the loop running while this promise is pending. * * @return self */ public function reference(): self { if ($this->watcher !== null) { Loop::reference($this->watcher); } return $this; } /** * Unreferences the internal watcher in the event loop, allowing the loop to stop while this promise is pending if * no other events are pending in the loop. * * @return self */ public function unreference(): self { if ($this->watcher !== null) { Loop::unreference($this->watcher); } return $this; } } lib/Emitter.php000064400000003767144760120120007446 0ustar00 Has public emit, complete, and fail methods. */ private $emitter; /** @var Iterator Hides producer methods. */ private $iterator; public function __construct() { $this->emitter = new class implements Iterator { use Internal\Producer { emit as public; complete as public; fail as public; } }; $this->iterator = new Internal\PrivateIterator($this->emitter); } /** * @return Iterator * @psalm-return Iterator */ public function iterate(): Iterator { return $this->iterator; } /** * Emits a value to the iterator. * * @param mixed $value * * @psalm-param TValue $value * * @return Promise * @psalm-return Promise * @psalm-suppress MixedInferredReturnType * @psalm-suppress MixedReturnStatement */ public function emit($value): Promise { /** @psalm-suppress UndefinedInterfaceMethod */ return $this->emitter->emit($value); } /** * Completes the iterator. * * @return void */ public function complete() { /** @psalm-suppress UndefinedInterfaceMethod */ $this->emitter->complete(); } /** * Fails the iterator with the given reason. * * @param \Throwable $reason * * @return void */ public function fail(\Throwable $reason) { /** @psalm-suppress UndefinedInterfaceMethod */ $this->emitter->fail($reason); } } lib/Failure.php000064400000002256144760120120007414 0ustar00 */ final class Failure implements Promise { /** @var \Throwable $exception */ private $exception; /** * @param \Throwable $exception Rejection reason. */ public function __construct(\Throwable $exception) { $this->exception = $exception; } /** * {@inheritdoc} */ public function onResolve(callable $onResolved) { try { /** @var mixed $result */ $result = $onResolved($this->exception, null); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { Promise\rethrow($result); } } catch (\Throwable $exception) { Loop::defer(static function () use ($exception) { throw $exception; }); } } } lib/Internal/Placeholder.php000064400000012212144760120120012014 0ustar00, mixed, * mixed>|null)|callable(\Throwable|null, mixed): void */ private $onResolved; /** @var null|array */ private $resolutionTrace; /** * @inheritdoc */ public function onResolve(callable $onResolved) { if ($this->resolved) { if ($this->result instanceof Promise) { $this->result->onResolve($onResolved); return; } try { /** @var mixed $result */ $result = $onResolved(null, $this->result); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { Promise\rethrow($result); } } catch (\Throwable $exception) { Loop::defer(static function () use ($exception) { throw $exception; }); } return; } if (null === $this->onResolved) { $this->onResolved = $onResolved; return; } if (!$this->onResolved instanceof ResolutionQueue) { /** @psalm-suppress InternalClass */ $this->onResolved = new ResolutionQueue($this->onResolved); } /** @psalm-suppress InternalMethod */ $this->onResolved->push($onResolved); } public function __destruct() { try { $this->result = null; } catch (\Throwable $e) { Loop::defer(static function () use ($e) { throw $e; }); } } /** * @param mixed $value * * @return void * * @throws \Error Thrown if the promise has already been resolved. */ private function resolve($value = null) { if ($this->resolved) { $message = "Promise has already been resolved"; if (isset($this->resolutionTrace)) { $trace = formatStacktrace($this->resolutionTrace); $message .= ". Previous resolution trace:\n\n{$trace}\n\n"; } else { // @codeCoverageIgnoreStart $message .= ", define environment variable AMP_DEBUG or const AMP_DEBUG = true and enable assertions " . "for a stacktrace of the previous resolution."; // @codeCoverageIgnoreEnd } throw new \Error($message); } \assert((function () { $env = \getenv("AMP_DEBUG") ?: "0"; if (($env !== "0" && $env !== "false") || (\defined("AMP_DEBUG") && \AMP_DEBUG)) { $trace = \debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS); \array_shift($trace); // remove current closure $this->resolutionTrace = $trace; } return true; })()); if ($value instanceof ReactPromise) { $value = Promise\adapt($value); } $this->resolved = true; $this->result = $value; if ($this->onResolved === null) { return; } $onResolved = $this->onResolved; $this->onResolved = null; if ($this->result instanceof Promise) { $this->result->onResolve($onResolved); return; } try { /** @var mixed $result */ $result = $onResolved(null, $this->result); $onResolved = null; // allow garbage collection of $onResolved, to catch any exceptions from destructors if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { Promise\rethrow($result); } } catch (\Throwable $exception) { Loop::defer(static function () use ($exception) { throw $exception; }); } } /** * @param \Throwable $reason Failure reason. * * @return void */ private function fail(\Throwable $reason) { $this->resolve(new Failure($reason)); } /** * @return bool True if the placeholder has been resolved. */ private function isResolved(): bool { return $this->resolved; } } lib/Internal/PrivateIterator.php000064400000001572144760120120012725 0ustar00 */ final class PrivateIterator implements Iterator { /** @var Iterator */ private $iterator; /** * @param Iterator $iterator * * @psalm-param Iterator $iterator */ public function __construct(Iterator $iterator) { $this->iterator = $iterator; } /** * @return Promise */ public function advance(): Promise { return $this->iterator->advance(); } /** * @psalm-return TValue */ public function getCurrent() { return $this->iterator->getCurrent(); } } lib/Internal/PrivatePromise.php000064400000001006144760120120012542 0ustar00promise = $promise; } public function onResolve(callable $onResolved) { $this->promise->onResolve($onResolved); } } lib/Internal/Producer.php000064400000013266144760120120011367 0ustar00 */ public function advance(): Promise { if ($this->waiting !== null) { throw new \Error("The prior promise returned must resolve before invoking this method again"); } unset($this->values[$this->consumePosition]); $position = ++$this->consumePosition; if (\array_key_exists($position, $this->values)) { \assert(isset($this->backPressure[$position])); $deferred = $this->backPressure[$position]; unset($this->backPressure[$position]); $deferred->resolve(); return new Success(true); } if ($this->complete) { return $this->complete; } $this->waiting = new Deferred; return $this->waiting->promise(); } /** * {@inheritdoc} * * @return TValue */ public function getCurrent() { if (empty($this->values) && $this->complete) { throw new \Error("The iterator has completed"); } if (!\array_key_exists($this->consumePosition, $this->values)) { throw new \Error("Promise returned from advance() must resolve before calling this method"); } return $this->values[$this->consumePosition]; } /** * Emits a value from the iterator. The returned promise is resolved once the emitted value has been consumed. * * @param mixed $value * * @return Promise * @psalm-return Promise * * @throws \Error If the iterator has completed. */ private function emit($value): Promise { if ($this->complete) { throw new \Error("Iterators cannot emit values after calling complete"); } if ($value instanceof ReactPromise) { $value = Promise\adapt($value); } if ($value instanceof Promise) { $deferred = new Deferred; $value->onResolve(function ($e, $v) use ($deferred) { if ($this->complete) { $deferred->fail( new \Error("The iterator was completed before the promise result could be emitted") ); return; } if ($e) { $this->fail($e); $deferred->fail($e); return; } $deferred->resolve($this->emit($v)); }); return $deferred->promise(); } $position = ++$this->emitPosition; $this->values[$position] = $value; if ($this->waiting !== null) { $waiting = $this->waiting; $this->waiting = null; $waiting->resolve(true); return new Success; // Consumer was already waiting for a new value, so back-pressure is unnecessary. } $this->backPressure[$position] = $pressure = new Deferred; return $pressure->promise(); } /** * Completes the iterator. * * @return void * * @throws \Error If the iterator has already been completed. */ private function complete() { if ($this->complete) { $message = "Iterator has already been completed"; if (isset($this->resolutionTrace)) { $trace = formatStacktrace($this->resolutionTrace); $message .= ". Previous completion trace:\n\n{$trace}\n\n"; } else { // @codeCoverageIgnoreStart $message .= ", define environment variable AMP_DEBUG or const AMP_DEBUG = true and enable assertions " . "for a stacktrace of the previous resolution."; // @codeCoverageIgnoreEnd } throw new \Error($message); } \assert((function () { $env = \getenv("AMP_DEBUG") ?: "0"; if (($env !== "0" && $env !== "false") || (\defined("AMP_DEBUG") && \AMP_DEBUG)) { $trace = \debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS); \array_shift($trace); // remove current closure $this->resolutionTrace = $trace; } return true; })()); $this->complete = new Success(false); if ($this->waiting !== null) { $waiting = $this->waiting; $this->waiting = null; $waiting->resolve($this->complete); } } /** * @param \Throwable $exception * * @return void */ private function fail(\Throwable $exception) { $this->complete = new Failure($exception); if ($this->waiting !== null) { $waiting = $this->waiting; $this->waiting = null; $waiting->resolve($this->complete); } } } lib/Internal/ResolutionQueue.php000064400000005444144760120120012753 0ustar00, mixed, * mixed>|null) | callable(\Throwable|null, mixed): void> */ private $queue = []; /** * @param callable|null $callback Initial callback to add to queue. * * @psalm-param null|callable(\Throwable|null, mixed): (Promise|\React\Promise\PromiseInterface|\Generator, mixed, * mixed>|null) | callable(\Throwable|null, mixed): void $callback */ public function __construct(callable $callback = null) { if ($callback !== null) { $this->push($callback); } } /** * Unrolls instances of self to avoid blowing up the call stack on resolution. * * @param callable $callback * * @psalm-param callable(\Throwable|null, mixed): (Promise|\React\Promise\PromiseInterface|\Generator, mixed, * mixed>|null) | callable(\Throwable|null, mixed): void $callback * * @return void */ public function push(callable $callback) { if ($callback instanceof self) { $this->queue = \array_merge($this->queue, $callback->queue); return; } $this->queue[] = $callback; } /** * Calls each callback in the queue, passing the provided values to the function. * * @param \Throwable|null $exception * @param mixed $value * * @return void */ public function __invoke($exception, $value) { foreach ($this->queue as $callback) { try { $result = $callback($exception, $value); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { Promise\rethrow($result); } } catch (\Throwable $exception) { Loop::defer(static function () use ($exception) { throw $exception; }); } } } } lib/Internal/functions.php000064400000007060144760120120011607 0ustar00 $trace Output of * `debug_backtrace()`. * * @return string Formatted stacktrace. * * @codeCoverageIgnore * @internal */ function formatStacktrace(array $trace): string { return \implode("\n", \array_map(static function ($e, $i) { $line = "#{$i} "; if (isset($e["file"])) { $line .= "{$e['file']}:{$e['line']} "; } if (isset($e["type"])) { $line .= $e["class"] . $e["type"]; } return $line . $e["function"] . "()"; }, $trace, \array_keys($trace))); } /** * Creates a `TypeError` with a standardized error message. * * @param string[] $expected Expected types. * @param mixed $given Given value. * * @return \TypeError * * @internal */ function createTypeError(array $expected, $given): \TypeError { $givenType = \is_object($given) ? \sprintf("instance of %s", \get_class($given)) : \gettype($given); if (\count($expected) === 1) { $expectedType = "Expected the following type: " . \array_pop($expected); } else { $expectedType = "Expected one of the following types: " . \implode(", ", $expected); } return new \TypeError("{$expectedType}; {$givenType} given"); } /** * Returns the current time relative to an arbitrary point in time. * * @return int Time in milliseconds. */ function getCurrentTime(): int { /** @var int|null $startTime */ static $startTime; /** @var int|null $nextWarning */ static $nextWarning; if (\PHP_INT_SIZE === 4) { // @codeCoverageIgnoreStart if ($startTime === null) { $startTime = \PHP_VERSION_ID >= 70300 ? \hrtime(false)[0] : \time(); $nextWarning = \PHP_INT_MAX - 86400 * 7; } if (\PHP_VERSION_ID >= 70300) { list($seconds, $nanoseconds) = \hrtime(false); $seconds -= $startTime; if ($seconds >= $nextWarning) { $timeToOverflow = (\PHP_INT_MAX - $seconds * 1000) / 1000; \trigger_error( "getCurrentTime() will overflow in $timeToOverflow seconds, please restart the process before that. " . "You're using a 32 bit version of PHP, so time will overflow about every 24 days. Regular restarts are required.", \E_USER_WARNING ); /** @psalm-suppress PossiblyNullOperand */ $nextWarning += 600; // every 10 minutes } return (int) ($seconds * 1000 + $nanoseconds / 1000000); } $seconds = \microtime(true) - $startTime; if ($seconds >= $nextWarning) { $timeToOverflow = (\PHP_INT_MAX - $seconds * 1000) / 1000; \trigger_error( "getCurrentTime() will overflow in $timeToOverflow seconds, please restart the process before that. " . "You're using a 32 bit version of PHP, so time will overflow about every 24 days. Regular restarts are required.", \E_USER_WARNING ); /** @psalm-suppress PossiblyNullOperand */ $nextWarning += 600; // every 10 minutes } return (int) ($seconds * 1000); // @codeCoverageIgnoreEnd } if (\PHP_VERSION_ID >= 70300) { list($seconds, $nanoseconds) = \hrtime(false); return (int) ($seconds * 1000 + $nanoseconds / 1000000); } return (int) (\microtime(true) * 1000); } lib/InvalidYieldError.php000064400000002144144760120120011410 0ustar00current(); $prefix .= \sprintf( "; %s yielded at key %s", \is_object($yielded) ? \get_class($yielded) : \gettype($yielded), \var_export($generator->key(), true) ); if (!$generator->valid()) { parent::__construct($prefix, 0, $previous); return; } $reflGen = new \ReflectionGenerator($generator); $exeGen = $reflGen->getExecutingGenerator(); if ($isSubgenerator = ($exeGen !== $generator)) { $reflGen = new \ReflectionGenerator($exeGen); } parent::__construct(\sprintf( "%s on line %s in %s", $prefix, $reflGen->getExecutingLine(), $reflGen->getExecutingFile() ), 0, $previous); } } lib/Iterator.php000064400000002105144760120120007607 0ustar00 * * @throws \Error If the prior promise returned from this method has not resolved. * @throws \Throwable The exception used to fail the iterator. */ public function advance(): Promise; /** * Gets the last emitted value or throws an exception if the iterator has completed. * * @return mixed Value emitted from the iterator. * @psalm-return TValue * * @throws \Error If the iterator has resolved or advance() was not called before calling this method. * @throws \Throwable The exception used to fail the iterator. */ public function getCurrent(); } lib/LazyPromise.php000064400000002261144760120120010277 0ustar00promisor = $promisor; } /** * {@inheritdoc} */ public function onResolve(callable $onResolved) { if ($this->promise === null) { \assert($this->promisor !== null); $provider = $this->promisor; $this->promisor = null; $this->promise = call($provider); } \assert($this->promise !== null); $this->promise->onResolve($onResolved); } } lib/Loop.php000064400000042620144760120120006735 0ustar00defer($callback); } self::$driver->run(); } /** * Stop the event loop. * * When an event loop is stopped, it continues with its current tick and exits the loop afterwards. Multiple calls * to stop MUST be ignored and MUST NOT raise an exception. * * @return void */ public static function stop() { self::$driver->stop(); } /** * Defer the execution of a callback. * * The deferred callable MUST be executed before any other type of watcher in a tick. Order of enabling MUST be * preserved when executing the callbacks. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param callable(string $watcherId, mixed $data) $callback The callback to defer. The `$watcherId` will be * invalidated before the callback call. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public static function defer(callable $callback, $data = null): string { return self::$driver->defer($callback, $data); } /** * Delay the execution of a callback. * * The delay is a minimum and approximate, accuracy is not guaranteed. Order of calls MUST be determined by which * timers expire first, but timers with the same expiration time MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param int $delay The amount of time, in milliseconds, to delay the execution for. * @param callable(string $watcherId, mixed $data) $callback The callback to delay. The `$watcherId` will be * invalidated before the callback call. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public static function delay(int $delay, callable $callback, $data = null): string { return self::$driver->delay($delay, $callback, $data); } /** * Repeatedly execute a callback. * * The interval between executions is a minimum and approximate, accuracy is not guaranteed. Order of calls MUST be * determined by which timers expire first, but timers with the same expiration time MAY be executed in any order. * The first execution is scheduled after the first interval period. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param int $interval The time interval, in milliseconds, to wait between executions. * @param callable(string $watcherId, mixed $data) $callback The callback to repeat. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public static function repeat(int $interval, callable $callback, $data = null): string { return self::$driver->repeat($interval, $callback, $data); } /** * Execute a callback when a stream resource becomes readable or is closed for reading. * * Warning: Closing resources locally, e.g. with `fclose`, might not invoke the callback. Be sure to `cancel` the * watcher when closing the resource locally. Drivers MAY choose to notify the user if there are watchers on invalid * resources, but are not required to, due to the high performance impact. Watchers on closed resources are * therefore undefined behavior. * * Multiple watchers on the same stream MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param resource $stream The stream to monitor. * @param callable(string $watcherId, resource $stream, mixed $data) $callback The callback to execute. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public static function onReadable($stream, callable $callback, $data = null): string { return self::$driver->onReadable($stream, $callback, $data); } /** * Execute a callback when a stream resource becomes writable or is closed for writing. * * Warning: Closing resources locally, e.g. with `fclose`, might not invoke the callback. Be sure to `cancel` the * watcher when closing the resource locally. Drivers MAY choose to notify the user if there are watchers on invalid * resources, but are not required to, due to the high performance impact. Watchers on closed resources are * therefore undefined behavior. * * Multiple watchers on the same stream MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param resource $stream The stream to monitor. * @param callable(string $watcherId, resource $stream, mixed $data) $callback The callback to execute. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public static function onWritable($stream, callable $callback, $data = null): string { return self::$driver->onWritable($stream, $callback, $data); } /** * Execute a callback when a signal is received. * * Warning: Installing the same signal on different instances of this interface is deemed undefined behavior. * Implementations MAY try to detect this, if possible, but are not required to. This is due to technical * limitations of the signals being registered globally per process. * * Multiple watchers on the same signal MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param int $signo The signal number to monitor. * @param callable(string $watcherId, int $signo, mixed $data) $callback The callback to execute. * @param mixed $data Arbitrary data given to the callback function as the $data parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. * * @throws UnsupportedFeatureException If signal handling is not supported. */ public static function onSignal(int $signo, callable $callback, $data = null): string { return self::$driver->onSignal($signo, $callback, $data); } /** * Enable a watcher to be active starting in the next tick. * * Watchers MUST immediately be marked as enabled, but only be activated (i.e. callbacks can be called) right before * the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param string $watcherId The watcher identifier. * * @return void * * @throws InvalidWatcherError If the watcher identifier is invalid. */ public static function enable(string $watcherId) { self::$driver->enable($watcherId); } /** * Disable a watcher immediately. * * A watcher MUST be disabled immediately, e.g. if a defer watcher disables a later defer watcher, the second defer * watcher isn't executed in this tick. * * Disabling a watcher MUST NOT invalidate the watcher. Calling this function MUST NOT fail, even if passed an * invalid watcher. * * @param string $watcherId The watcher identifier. * * @return void */ public static function disable(string $watcherId) { if (\PHP_VERSION_ID < 70200 && !isset(self::$driver)) { // Prior to PHP 7.2, self::$driver may be unset during destruct. // See https://github.com/amphp/amp/issues/212. return; } self::$driver->disable($watcherId); } /** * Cancel a watcher. * * This will detatch the event loop from all resources that are associated to the watcher. After this operation the * watcher is permanently invalid. Calling this function MUST NOT fail, even if passed an invalid watcher. * * @param string $watcherId The watcher identifier. * * @return void */ public static function cancel(string $watcherId) { if (\PHP_VERSION_ID < 70200 && !isset(self::$driver)) { // Prior to PHP 7.2, self::$driver may be unset during destruct. // See https://github.com/amphp/amp/issues/212. return; } self::$driver->cancel($watcherId); } /** * Reference a watcher. * * This will keep the event loop alive whilst the watcher is still being monitored. Watchers have this state by * default. * * @param string $watcherId The watcher identifier. * * @return void * * @throws InvalidWatcherError If the watcher identifier is invalid. */ public static function reference(string $watcherId) { self::$driver->reference($watcherId); } /** * Unreference a watcher. * * The event loop should exit the run method when only unreferenced watchers are still being monitored. Watchers * are all referenced by default. * * @param string $watcherId The watcher identifier. * * @return void */ public static function unreference(string $watcherId) { if (\PHP_VERSION_ID < 70200 && !isset(self::$driver)) { // Prior to PHP 7.2, self::$driver may be unset during destruct. // See https://github.com/amphp/amp/issues/212. return; } self::$driver->unreference($watcherId); } /** * Returns the current loop time in millisecond increments. Note this value does not necessarily correlate to * wall-clock time, rather the value returned is meant to be used in relative comparisons to prior values returned * by this method (intervals, expiration calculations, etc.) and is only updated once per loop tick. * * @return int */ public static function now(): int { return self::$driver->now(); } /** * Stores information in the loop bound registry. * * Stored information is package private. Packages MUST NOT retrieve the stored state of other packages. Packages * MUST use their namespace as prefix for keys. They may do so by using `SomeClass::class` as key. * * If packages want to expose loop bound state to consumers other than the package, they SHOULD provide a dedicated * interface for that purpose instead of sharing the storage key. * * @param string $key The namespaced storage key. * @param mixed $value The value to be stored. * * @return void */ public static function setState(string $key, $value) { self::$driver->setState($key, $value); } /** * Gets information stored bound to the loop. * * Stored information is package private. Packages MUST NOT retrieve the stored state of other packages. Packages * MUST use their namespace as prefix for keys. They may do so by using `SomeClass::class` as key. * * If packages want to expose loop bound state to consumers other than the package, they SHOULD provide a dedicated * interface for that purpose instead of sharing the storage key. * * @param string $key The namespaced storage key. * * @return mixed The previously stored value or `null` if it doesn't exist. */ public static function getState(string $key) { return self::$driver->getState($key); } /** * Set a callback to be executed when an error occurs. * * The callback receives the error as the first and only parameter. The return value of the callback gets ignored. * If it can't handle the error, it MUST throw the error. Errors thrown by the callback or during its invocation * MUST be thrown into the `run` loop and stop the driver. * * Subsequent calls to this method will overwrite the previous handler. * * @param callable(\Throwable $error)|null $callback The callback to execute. `null` will clear the * current handler. * * @return callable(\Throwable $error)|null The previous handler, `null` if there was none. */ public static function setErrorHandler(callable $callback = null) { return self::$driver->setErrorHandler($callback); } /** * Retrieve an associative array of information about the event loop driver. * * The returned array MUST contain the following data describing the driver's currently registered watchers: * * [ * "defer" => ["enabled" => int, "disabled" => int], * "delay" => ["enabled" => int, "disabled" => int], * "repeat" => ["enabled" => int, "disabled" => int], * "on_readable" => ["enabled" => int, "disabled" => int], * "on_writable" => ["enabled" => int, "disabled" => int], * "on_signal" => ["enabled" => int, "disabled" => int], * "enabled_watchers" => ["referenced" => int, "unreferenced" => int], * "running" => bool * ]; * * Implementations MAY optionally add more information in the array but at minimum the above `key => value` format * MUST always be provided. * * @return array Statistics about the loop in the described format. */ public static function getInfo(): array { return self::$driver->getInfo(); } /** * Retrieve the event loop driver that is in scope. * * @return Driver */ public static function get(): Driver { return self::$driver; } } // Default factory, don't move this to a file loaded by the composer "files" autoload mechanism, otherwise custom // implementations might have issues setting a default loop, because it's overridden by us then. // @codeCoverageIgnoreStart Loop::set((new DriverFactory)->create()); // @codeCoverageIgnoreEnd lib/Loop/Driver.php000064400000063664144760120120010203 0ustar00running = true; try { while ($this->running) { if ($this->isEmpty()) { return; } $this->tick(); } } finally { $this->stop(); } } /** * @return bool True if no enabled and referenced watchers remain in the loop. */ private function isEmpty(): bool { foreach ($this->watchers as $watcher) { if ($watcher->enabled && $watcher->referenced) { return false; } } return true; } /** * Executes a single tick of the event loop. * * @return void */ private function tick() { if (empty($this->deferQueue)) { $this->deferQueue = $this->nextTickQueue; } else { $this->deferQueue = \array_merge($this->deferQueue, $this->nextTickQueue); } $this->nextTickQueue = []; $this->activate($this->enableQueue); $this->enableQueue = []; foreach ($this->deferQueue as $watcher) { if (!isset($this->deferQueue[$watcher->id])) { continue; // Watcher disabled by another defer watcher. } unset($this->watchers[$watcher->id], $this->deferQueue[$watcher->id]); try { /** @var mixed $result */ $result = ($watcher->callback)($watcher->id, $watcher->data); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } } /** @psalm-suppress RedundantCondition */ $this->dispatch(empty($this->nextTickQueue) && empty($this->enableQueue) && $this->running && !$this->isEmpty()); } /** * Activates (enables) all the given watchers. * * @param Watcher[] $watchers * * @return void */ abstract protected function activate(array $watchers); /** * Dispatches any pending read/write, timer, and signal events. * * @param bool $blocking * * @return void */ abstract protected function dispatch(bool $blocking); /** * Stop the event loop. * * When an event loop is stopped, it continues with its current tick and exits the loop afterwards. Multiple calls * to stop MUST be ignored and MUST NOT raise an exception. * * @return void */ public function stop() { $this->running = false; } /** * Defer the execution of a callback. * * The deferred callable MUST be executed before any other type of watcher in a tick. Order of enabling MUST be * preserved when executing the callbacks. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param callable (string $watcherId, mixed $data) $callback The callback to defer. The `$watcherId` will be * invalidated before the callback call. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public function defer(callable $callback, $data = null): string { /** @psalm-var Watcher $watcher */ $watcher = new Watcher; $watcher->type = Watcher::DEFER; $watcher->id = $this->nextId++; $watcher->callback = $callback; $watcher->data = $data; $this->watchers[$watcher->id] = $watcher; $this->nextTickQueue[$watcher->id] = $watcher; return $watcher->id; } /** * Delay the execution of a callback. * * The delay is a minimum and approximate, accuracy is not guaranteed. Order of calls MUST be determined by which * timers expire first, but timers with the same expiration time MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param int $delay The amount of time, in milliseconds, to delay the execution for. * @param callable (string $watcherId, mixed $data) $callback The callback to delay. The `$watcherId` will be * invalidated before the callback call. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public function delay(int $delay, callable $callback, $data = null): string { if ($delay < 0) { throw new \Error("Delay must be greater than or equal to zero"); } /** @psalm-var Watcher $watcher */ $watcher = new Watcher; $watcher->type = Watcher::DELAY; $watcher->id = $this->nextId++; $watcher->callback = $callback; $watcher->value = $delay; $watcher->expiration = $this->now() + $delay; $watcher->data = $data; $this->watchers[$watcher->id] = $watcher; $this->enableQueue[$watcher->id] = $watcher; return $watcher->id; } /** * Repeatedly execute a callback. * * The interval between executions is a minimum and approximate, accuracy is not guaranteed. Order of calls MUST be * determined by which timers expire first, but timers with the same expiration time MAY be executed in any order. * The first execution is scheduled after the first interval period. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param int $interval The time interval, in milliseconds, to wait between executions. * @param callable (string $watcherId, mixed $data) $callback The callback to repeat. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public function repeat(int $interval, callable $callback, $data = null): string { if ($interval < 0) { throw new \Error("Interval must be greater than or equal to zero"); } /** @psalm-var Watcher $watcher */ $watcher = new Watcher; $watcher->type = Watcher::REPEAT; $watcher->id = $this->nextId++; $watcher->callback = $callback; $watcher->value = $interval; $watcher->expiration = $this->now() + $interval; $watcher->data = $data; $this->watchers[$watcher->id] = $watcher; $this->enableQueue[$watcher->id] = $watcher; return $watcher->id; } /** * Execute a callback when a stream resource becomes readable or is closed for reading. * * Warning: Closing resources locally, e.g. with `fclose`, might not invoke the callback. Be sure to `cancel` the * watcher when closing the resource locally. Drivers MAY choose to notify the user if there are watchers on invalid * resources, but are not required to, due to the high performance impact. Watchers on closed resources are * therefore undefined behavior. * * Multiple watchers on the same stream MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param resource $stream The stream to monitor. * @param callable (string $watcherId, resource $stream, mixed $data) $callback The callback to execute. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public function onReadable($stream, callable $callback, $data = null): string { /** @psalm-var Watcher $watcher */ $watcher = new Watcher; $watcher->type = Watcher::READABLE; $watcher->id = $this->nextId++; $watcher->callback = $callback; $watcher->value = $stream; $watcher->data = $data; $this->watchers[$watcher->id] = $watcher; $this->enableQueue[$watcher->id] = $watcher; return $watcher->id; } /** * Execute a callback when a stream resource becomes writable or is closed for writing. * * Warning: Closing resources locally, e.g. with `fclose`, might not invoke the callback. Be sure to `cancel` the * watcher when closing the resource locally. Drivers MAY choose to notify the user if there are watchers on invalid * resources, but are not required to, due to the high performance impact. Watchers on closed resources are * therefore undefined behavior. * * Multiple watchers on the same stream MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param resource $stream The stream to monitor. * @param callable (string $watcherId, resource $stream, mixed $data) $callback The callback to execute. * @param mixed $data Arbitrary data given to the callback function as the `$data` parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. */ public function onWritable($stream, callable $callback, $data = null): string { /** @psalm-var Watcher $watcher */ $watcher = new Watcher; $watcher->type = Watcher::WRITABLE; $watcher->id = $this->nextId++; $watcher->callback = $callback; $watcher->value = $stream; $watcher->data = $data; $this->watchers[$watcher->id] = $watcher; $this->enableQueue[$watcher->id] = $watcher; return $watcher->id; } /** * Execute a callback when a signal is received. * * Warning: Installing the same signal on different instances of this interface is deemed undefined behavior. * Implementations MAY try to detect this, if possible, but are not required to. This is due to technical * limitations of the signals being registered globally per process. * * Multiple watchers on the same signal MAY be executed in any order. * * The created watcher MUST immediately be marked as enabled, but only be activated (i.e. callback can be called) * right before the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param int $signo The signal number to monitor. * @param callable (string $watcherId, int $signo, mixed $data) $callback The callback to execute. * @param mixed $data Arbitrary data given to the callback function as the $data parameter. * * @return string An unique identifier that can be used to cancel, enable or disable the watcher. * * @throws UnsupportedFeatureException If signal handling is not supported. */ public function onSignal(int $signo, callable $callback, $data = null): string { /** @psalm-var Watcher $watcher */ $watcher = new Watcher; $watcher->type = Watcher::SIGNAL; $watcher->id = $this->nextId++; $watcher->callback = $callback; $watcher->value = $signo; $watcher->data = $data; $this->watchers[$watcher->id] = $watcher; $this->enableQueue[$watcher->id] = $watcher; return $watcher->id; } /** * Enable a watcher to be active starting in the next tick. * * Watchers MUST immediately be marked as enabled, but only be activated (i.e. callbacks can be called) right before * the next tick. Callbacks of watchers MUST NOT be called in the tick they were enabled. * * @param string $watcherId The watcher identifier. * * @return void * * @throws InvalidWatcherError If the watcher identifier is invalid. */ public function enable(string $watcherId) { if (!isset($this->watchers[$watcherId])) { throw new InvalidWatcherError($watcherId, "Cannot enable an invalid watcher identifier: '{$watcherId}'"); } $watcher = $this->watchers[$watcherId]; if ($watcher->enabled) { return; // Watcher already enabled. } $watcher->enabled = true; switch ($watcher->type) { case Watcher::DEFER: $this->nextTickQueue[$watcher->id] = $watcher; break; case Watcher::REPEAT: case Watcher::DELAY: \assert(\is_int($watcher->value)); $watcher->expiration = $this->now() + $watcher->value; $this->enableQueue[$watcher->id] = $watcher; break; default: $this->enableQueue[$watcher->id] = $watcher; break; } } /** * Cancel a watcher. * * This will detach the event loop from all resources that are associated to the watcher. After this operation the * watcher is permanently invalid. Calling this function MUST NOT fail, even if passed an invalid watcher. * * @param string $watcherId The watcher identifier. * * @return void */ public function cancel(string $watcherId) { $this->disable($watcherId); unset($this->watchers[$watcherId]); } /** * Disable a watcher immediately. * * A watcher MUST be disabled immediately, e.g. if a defer watcher disables a later defer watcher, the second defer * watcher isn't executed in this tick. * * Disabling a watcher MUST NOT invalidate the watcher. Calling this function MUST NOT fail, even if passed an * invalid watcher. * * @param string $watcherId The watcher identifier. * * @return void */ public function disable(string $watcherId) { if (!isset($this->watchers[$watcherId])) { return; } $watcher = $this->watchers[$watcherId]; if (!$watcher->enabled) { return; // Watcher already disabled. } $watcher->enabled = false; $id = $watcher->id; switch ($watcher->type) { case Watcher::DEFER: if (isset($this->nextTickQueue[$id])) { // Watcher was only queued to be enabled. unset($this->nextTickQueue[$id]); } else { unset($this->deferQueue[$id]); } break; default: if (isset($this->enableQueue[$id])) { // Watcher was only queued to be enabled. unset($this->enableQueue[$id]); } else { $this->deactivate($watcher); } break; } } /** * Deactivates (disables) the given watcher. * * @param Watcher $watcher * * @return void */ abstract protected function deactivate(Watcher $watcher); /** * Reference a watcher. * * This will keep the event loop alive whilst the watcher is still being monitored. Watchers have this state by * default. * * @param string $watcherId The watcher identifier. * * @return void * * @throws InvalidWatcherError If the watcher identifier is invalid. */ public function reference(string $watcherId) { if (!isset($this->watchers[$watcherId])) { throw new InvalidWatcherError($watcherId, "Cannot reference an invalid watcher identifier: '{$watcherId}'"); } $this->watchers[$watcherId]->referenced = true; } /** * Unreference a watcher. * * The event loop should exit the run method when only unreferenced watchers are still being monitored. Watchers * are all referenced by default. * * @param string $watcherId The watcher identifier. * * @return void */ public function unreference(string $watcherId) { if (!isset($this->watchers[$watcherId])) { return; } $this->watchers[$watcherId]->referenced = false; } /** * Stores information in the loop bound registry. * * Stored information is package private. Packages MUST NOT retrieve the stored state of other packages. Packages * MUST use their namespace as prefix for keys. They may do so by using `SomeClass::class` as key. * * If packages want to expose loop bound state to consumers other than the package, they SHOULD provide a dedicated * interface for that purpose instead of sharing the storage key. * * @param string $key The namespaced storage key. * @param mixed $value The value to be stored. * * @return void */ final public function setState(string $key, $value) { if ($value === null) { unset($this->registry[$key]); } else { $this->registry[$key] = $value; } } /** * Gets information stored bound to the loop. * * Stored information is package private. Packages MUST NOT retrieve the stored state of other packages. Packages * MUST use their namespace as prefix for keys. They may do so by using `SomeClass::class` as key. * * If packages want to expose loop bound state to consumers other than the package, they SHOULD provide a dedicated * interface for that purpose instead of sharing the storage key. * * @param string $key The namespaced storage key. * * @return mixed The previously stored value or `null` if it doesn't exist. */ final public function getState(string $key) { return isset($this->registry[$key]) ? $this->registry[$key] : null; } /** * Set a callback to be executed when an error occurs. * * The callback receives the error as the first and only parameter. The return value of the callback gets ignored. * If it can't handle the error, it MUST throw the error. Errors thrown by the callback or during its invocation * MUST be thrown into the `run` loop and stop the driver. * * Subsequent calls to this method will overwrite the previous handler. * * @param callable(\Throwable $error):void|null $callback The callback to execute. `null` will clear the * current handler. * * @return callable(\Throwable $error):void|null The previous handler, `null` if there was none. */ public function setErrorHandler(callable $callback = null) { $previous = $this->errorHandler; $this->errorHandler = $callback; return $previous; } /** * Invokes the error handler with the given exception. * * @param \Throwable $exception The exception thrown from a watcher callback. * * @return void * @throws \Throwable If no error handler has been set. */ protected function error(\Throwable $exception) { if ($this->errorHandler === null) { throw $exception; } ($this->errorHandler)($exception); } /** * Returns the current loop time in millisecond increments. Note this value does not necessarily correlate to * wall-clock time, rather the value returned is meant to be used in relative comparisons to prior values returned * by this method (intervals, expiration calculations, etc.) and is only updated once per loop tick. * * Extending classes should override this function to return a value cached once per loop tick. * * @return int */ public function now(): int { return (int) (\microtime(true) * self::MILLISEC_PER_SEC); } /** * Get the underlying loop handle. * * Example: the `uv_loop` resource for `libuv` or the `EvLoop` object for `libev` or `null` for a native driver. * * Note: This function is *not* exposed in the `Loop` class. Users shall access it directly on the respective loop * instance. * * @return null|object|resource The loop handle the event loop operates on. `null` if there is none. */ abstract public function getHandle(); /** * Returns the same array of data as getInfo(). * * @return array */ public function __debugInfo() { // @codeCoverageIgnoreStart return $this->getInfo(); // @codeCoverageIgnoreEnd } /** * Retrieve an associative array of information about the event loop driver. * * The returned array MUST contain the following data describing the driver's currently registered watchers: * * [ * "defer" => ["enabled" => int, "disabled" => int], * "delay" => ["enabled" => int, "disabled" => int], * "repeat" => ["enabled" => int, "disabled" => int], * "on_readable" => ["enabled" => int, "disabled" => int], * "on_writable" => ["enabled" => int, "disabled" => int], * "on_signal" => ["enabled" => int, "disabled" => int], * "enabled_watchers" => ["referenced" => int, "unreferenced" => int], * "running" => bool * ]; * * Implementations MAY optionally add more information in the array but at minimum the above `key => value` format * MUST always be provided. * * @return array Statistics about the loop in the described format. */ public function getInfo(): array { $watchers = [ "referenced" => 0, "unreferenced" => 0, ]; $defer = $delay = $repeat = $onReadable = $onWritable = $onSignal = [ "enabled" => 0, "disabled" => 0, ]; foreach ($this->watchers as $watcher) { switch ($watcher->type) { case Watcher::READABLE: $array = &$onReadable; break; case Watcher::WRITABLE: $array = &$onWritable; break; case Watcher::SIGNAL: $array = &$onSignal; break; case Watcher::DEFER: $array = &$defer; break; case Watcher::DELAY: $array = &$delay; break; case Watcher::REPEAT: $array = &$repeat; break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } if ($watcher->enabled) { ++$array["enabled"]; if ($watcher->referenced) { ++$watchers["referenced"]; } else { ++$watchers["unreferenced"]; } } else { ++$array["disabled"]; } } return [ "enabled_watchers" => $watchers, "defer" => $defer, "delay" => $delay, "repeat" => $repeat, "on_readable" => $onReadable, "on_writable" => $onWritable, "on_signal" => $onSignal, "running" => (bool) $this->running, ]; } } lib/Loop/DriverFactory.php000064400000003125144760120120011515 0ustar00createDriverFromEnv()) { return $driver; } if (UvDriver::isSupported()) { return new UvDriver; } if (EvDriver::isSupported()) { return new EvDriver; } if (EventDriver::isSupported()) { return new EventDriver; } return new NativeDriver; })(); if (\getenv("AMP_DEBUG_TRACE_WATCHERS")) { return new TracingDriver($driver); } return $driver; } /** * @return Driver|null */ private function createDriverFromEnv() { $driver = \getenv("AMP_LOOP_DRIVER"); if (!$driver) { return null; } if (!\class_exists($driver)) { throw new \Error(\sprintf( "Driver '%s' does not exist.", $driver )); } if (!\is_subclass_of($driver, Driver::class)) { throw new \Error(\sprintf( "Driver '%s' is not a subclass of '%s'.", $driver, Driver::class )); } return new $driver; } } // @codeCoverageIgnoreEnd lib/Loop/EvDriver.php000064400000020760144760120120010464 0ustar00handle = new \EvLoop; $this->nowOffset = getCurrentTime(); $this->now = \random_int(0, $this->nowOffset); $this->nowOffset -= $this->now; if (self::$activeSignals === null) { self::$activeSignals = &$this->signals; } /** * @param \EvIO $event * * @return void */ $this->ioCallback = function (\EvIO $event) { /** @var Watcher $watcher */ $watcher = $event->data; try { $result = ($watcher->callback)($watcher->id, $watcher->value, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; /** * @param \EvTimer $event * * @return void */ $this->timerCallback = function (\EvTimer $event) { /** @var Watcher $watcher */ $watcher = $event->data; if ($watcher->type & Watcher::DELAY) { $this->cancel($watcher->id); } elseif ($watcher->value === 0) { // Disable and re-enable so it's not executed repeatedly in the same tick // See https://github.com/amphp/amp/issues/131 $this->disable($watcher->id); $this->enable($watcher->id); } try { $result = ($watcher->callback)($watcher->id, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; /** * @param \EvSignal $event * * @return void */ $this->signalCallback = function (\EvSignal $event) { /** @var Watcher $watcher */ $watcher = $event->data; try { $result = ($watcher->callback)($watcher->id, $watcher->value, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; } /** * {@inheritdoc} */ public function cancel(string $watcherId) { parent::cancel($watcherId); unset($this->events[$watcherId]); } public function __destruct() { foreach ($this->events as $event) { /** @psalm-suppress all */ if ($event !== null) { // Events may have been nulled in extension depending on destruct order. $event->stop(); } } // We need to clear all references to events manually, see // https://bitbucket.org/osmanov/pecl-ev/issues/31/segfault-in-ev_timer_stop $this->events = []; } /** * {@inheritdoc} */ public function run() { $active = self::$activeSignals; \assert($active !== null); foreach ($active as $event) { $event->stop(); } self::$activeSignals = &$this->signals; foreach ($this->signals as $event) { $event->start(); } try { parent::run(); } finally { foreach ($this->signals as $event) { $event->stop(); } self::$activeSignals = &$active; foreach ($active as $event) { $event->start(); } } } /** * {@inheritdoc} */ public function stop() { $this->handle->stop(); parent::stop(); } /** * {@inheritdoc} */ public function now(): int { $this->now = getCurrentTime() - $this->nowOffset; return $this->now; } /** * {@inheritdoc} */ public function getHandle(): \EvLoop { return $this->handle; } /** * {@inheritdoc} * * @return void */ protected function dispatch(bool $blocking) { $this->handle->run($blocking ? \Ev::RUN_ONCE : \Ev::RUN_ONCE | \Ev::RUN_NOWAIT); } /** * {@inheritdoc} * * @return void */ protected function activate(array $watchers) { $this->handle->nowUpdate(); $now = $this->now(); foreach ($watchers as $watcher) { if (!isset($this->events[$id = $watcher->id])) { switch ($watcher->type) { case Watcher::READABLE: \assert(\is_resource($watcher->value)); $this->events[$id] = $this->handle->io($watcher->value, \Ev::READ, $this->ioCallback, $watcher); break; case Watcher::WRITABLE: \assert(\is_resource($watcher->value)); $this->events[$id] = $this->handle->io( $watcher->value, \Ev::WRITE, $this->ioCallback, $watcher ); break; case Watcher::DELAY: case Watcher::REPEAT: \assert(\is_int($watcher->value)); $interval = $watcher->value / self::MILLISEC_PER_SEC; $this->events[$id] = $this->handle->timer( \max(0, ($watcher->expiration - $now) / self::MILLISEC_PER_SEC), ($watcher->type & Watcher::REPEAT) ? $interval : 0, $this->timerCallback, $watcher ); break; case Watcher::SIGNAL: \assert(\is_int($watcher->value)); $this->events[$id] = $this->handle->signal($watcher->value, $this->signalCallback, $watcher); break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } } else { $this->events[$id]->start(); } if ($watcher->type === Watcher::SIGNAL) { /** @psalm-suppress PropertyTypeCoercion */ $this->signals[$id] = $this->events[$id]; } } } /** * {@inheritdoc} * * @return void */ protected function deactivate(Watcher $watcher) { if (isset($this->events[$id = $watcher->id])) { $this->events[$id]->stop(); if ($watcher->type === Watcher::SIGNAL) { unset($this->signals[$id]); } } } } lib/Loop/EventDriver.php000064400000023501144760120120011167 0ustar00handle = new \EventBase; $this->nowOffset = getCurrentTime(); $this->now = \random_int(0, $this->nowOffset); $this->nowOffset -= $this->now; if (self::$activeSignals === null) { self::$activeSignals = &$this->signals; } /** * @param $resource * @param $what * @param Watcher $watcher * * @return void */ $this->ioCallback = function ($resource, $what, Watcher $watcher) { \assert(\is_resource($watcher->value)); try { $result = ($watcher->callback)($watcher->id, $watcher->value, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; /** * @param $resource * @param $what * @param Watcher $watcher * * @return void */ $this->timerCallback = function ($resource, $what, Watcher $watcher) { \assert(\is_int($watcher->value)); if ($watcher->type & Watcher::DELAY) { $this->cancel($watcher->id); } else { $this->events[$watcher->id]->add($watcher->value / self::MILLISEC_PER_SEC); } try { $result = ($watcher->callback)($watcher->id, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; /** * @param $signum * @param $what * @param Watcher $watcher * * @return void */ $this->signalCallback = function ($signum, $what, Watcher $watcher) { try { $result = ($watcher->callback)($watcher->id, $watcher->value, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; } /** * {@inheritdoc} */ public function cancel(string $watcherId) { parent::cancel($watcherId); if (isset($this->events[$watcherId])) { $this->events[$watcherId]->free(); unset($this->events[$watcherId]); } } public static function isSupported(): bool { return \extension_loaded("event"); } /** * @codeCoverageIgnore */ public function __destruct() { foreach ($this->events as $event) { if ($event !== null) { // Events may have been nulled in extension depending on destruct order. $event->free(); } } // Unset here, otherwise $event->del() fails with a warning, because __destruct order isn't defined. // See https://github.com/amphp/amp/issues/159. $this->events = []; // Manually free the loop handle to fully release loop resources. // See https://github.com/amphp/amp/issues/177. if ($this->handle !== null) { $this->handle->free(); $this->handle = null; } } /** * {@inheritdoc} */ public function run() { $active = self::$activeSignals; \assert($active !== null); foreach ($active as $event) { $event->del(); } self::$activeSignals = &$this->signals; foreach ($this->signals as $event) { /** @psalm-suppress TooFewArguments https://github.com/JetBrains/phpstorm-stubs/pull/763 */ $event->add(); } try { parent::run(); } finally { foreach ($this->signals as $event) { $event->del(); } self::$activeSignals = &$active; foreach ($active as $event) { /** @psalm-suppress TooFewArguments https://github.com/JetBrains/phpstorm-stubs/pull/763 */ $event->add(); } } } /** * {@inheritdoc} */ public function stop() { $this->handle->stop(); parent::stop(); } /** * {@inheritdoc} */ public function now(): int { $this->now = getCurrentTime() - $this->nowOffset; return $this->now; } /** * {@inheritdoc} */ public function getHandle(): \EventBase { return $this->handle; } /** * {@inheritdoc} * * @return void */ protected function dispatch(bool $blocking) { $this->handle->loop($blocking ? \EventBase::LOOP_ONCE : \EventBase::LOOP_ONCE | \EventBase::LOOP_NONBLOCK); } /** * {@inheritdoc} * * @return void */ protected function activate(array $watchers) { $now = $this->now(); foreach ($watchers as $watcher) { if (!isset($this->events[$id = $watcher->id])) { switch ($watcher->type) { case Watcher::READABLE: \assert(\is_resource($watcher->value)); $this->events[$id] = new \Event( $this->handle, $watcher->value, \Event::READ | \Event::PERSIST, $this->ioCallback, $watcher ); break; case Watcher::WRITABLE: \assert(\is_resource($watcher->value)); $this->events[$id] = new \Event( $this->handle, $watcher->value, \Event::WRITE | \Event::PERSIST, $this->ioCallback, $watcher ); break; case Watcher::DELAY: case Watcher::REPEAT: \assert(\is_int($watcher->value)); $this->events[$id] = new \Event( $this->handle, -1, \Event::TIMEOUT, $this->timerCallback, $watcher ); break; case Watcher::SIGNAL: \assert(\is_int($watcher->value)); $this->events[$id] = new \Event( $this->handle, $watcher->value, \Event::SIGNAL | \Event::PERSIST, $this->signalCallback, $watcher ); break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } } switch ($watcher->type) { case Watcher::DELAY: case Watcher::REPEAT: \assert(\is_int($watcher->value)); $interval = \max(0, $watcher->expiration - $now); $this->events[$id]->add($interval > 0 ? $interval / self::MILLISEC_PER_SEC : 0); break; case Watcher::SIGNAL: $this->signals[$id] = $this->events[$id]; // no break default: /** @psalm-suppress TooFewArguments https://github.com/JetBrains/phpstorm-stubs/pull/763 */ $this->events[$id]->add(); break; } } } /** * {@inheritdoc} * * @return void */ protected function deactivate(Watcher $watcher) { if (isset($this->events[$id = $watcher->id])) { $this->events[$id]->del(); if ($watcher->type === Watcher::SIGNAL) { unset($this->signals[$id]); } } } } lib/Loop/Internal/TimerQueue.php000064400000011067144760120120012577 0ustar00data[$node]; while ($node !== 0 && $entry->expiration < $this->data[$parent = ($node - 1) >> 1]->expiration) { $this->swap($node, $parent); $node = $parent; } } /** * @param int $node Rebuild the data array from the given node downward. * * @return void */ private function heapifyDown(int $node) { $length = \count($this->data); while (($child = ($node << 1) + 1) < $length) { if ($this->data[$child]->expiration < $this->data[$node]->expiration && ($child + 1 >= $length || $this->data[$child]->expiration < $this->data[$child + 1]->expiration) ) { // Left child is less than parent and right child. $swap = $child; } elseif ($child + 1 < $length && $this->data[$child + 1]->expiration < $this->data[$node]->expiration) { // Right child is less than parent and left child. $swap = $child + 1; } else { // Left and right child are greater than parent. break; } $this->swap($node, $swap); $node = $swap; } } private function swap(int $left, int $right) { $temp = $this->data[$left]; $this->data[$left] = $this->data[$right]; $this->pointers[$this->data[$right]->id] = $left; $this->data[$right] = $temp; $this->pointers[$temp->id] = $right; } /** * Inserts the watcher into the queue. Time complexity: O(log(n)). * * @param Watcher $watcher * * @psalm-param Watcher $watcher * * @return void */ public function insert(Watcher $watcher) { \assert($watcher->expiration !== null); \assert(!isset($this->pointers[$watcher->id])); $node = \count($this->data); $this->data[$node] = $watcher; $this->pointers[$watcher->id] = $node; $this->heapifyUp($node); } /** * Removes the given watcher from the queue. Time complexity: O(log(n)). * * @param Watcher $watcher * * @psalm-param Watcher $watcher * * @return void */ public function remove(Watcher $watcher) { $id = $watcher->id; if (!isset($this->pointers[$id])) { return; } $this->removeAndRebuild($this->pointers[$id]); } /** * Deletes and returns the Watcher on top of the heap if it has expired, otherwise null is returned. * Time complexity: O(log(n)). * * @param int $now Current loop time. * * @return Watcher|null Expired watcher at the top of the heap or null if the watcher has not expired. * * @psalm-return Watcher|null */ public function extract(int $now) { if (empty($this->data)) { return null; } $watcher = $this->data[0]; if ($watcher->expiration > $now) { return null; } $this->removeAndRebuild(0); return $watcher; } /** * Returns the expiration time value at the top of the heap. Time complexity: O(1). * * @return int|null Expiration time of the watcher at the top of the heap or null if the heap is empty. */ public function peek() { return isset($this->data[0]) ? $this->data[0]->expiration : null; } /** * @param int $node Remove the given node and then rebuild the data array. * * @return void */ private function removeAndRebuild(int $node) { $length = \count($this->data) - 1; $id = $this->data[$node]->id; $left = $this->data[$node] = $this->data[$length]; $this->pointers[$left->id] = $node; unset($this->data[$length], $this->pointers[$id]); if ($node < $length) { // don't need to do anything if we removed the last element $parent = ($node - 1) >> 1; if ($parent >= 0 && $this->data[$node]->expiration < $this->data[$parent]->expiration) { $this->heapifyUp($node); } else { $this->heapifyDown($node); } } } } lib/Loop/InvalidWatcherError.php000064400000001425144760120120012651 0ustar00watcherId = $watcherId; parent::__construct($message); } /** * @return string The watcher identifier. */ public function getWatcherId() { return $this->watcherId; } } lib/Loop/NativeDriver.php000064400000034177144760120120011347 0ustar00timerQueue = new Internal\TimerQueue; $this->signalHandling = \extension_loaded("pcntl"); $this->nowOffset = getCurrentTime(); $this->now = \random_int(0, $this->nowOffset); $this->nowOffset -= $this->now; $this->streamSelectErrorHandler = function ($errno, $message) { // Casing changed in PHP 8 from 'unable' to 'Unable' if (\stripos($message, "stream_select(): unable to select [4]: ") === 0) { // EINTR $this->streamSelectIgnoreResult = true; return; } if (\strpos($message, 'FD_SETSIZE') !== false) { $message = \str_replace(["\r\n", "\n", "\r"], " ", $message); $pattern = '(stream_select\(\): You MUST recompile PHP with a larger value of FD_SETSIZE. It is set to (\d+), but you have descriptors numbered at least as high as (\d+)\.)'; if (\preg_match($pattern, $message, $match)) { $helpLink = 'https://amphp.org/amp/event-loop/#implementations'; $message = 'You have reached the limits of stream_select(). It has a FD_SETSIZE of ' . $match[1] . ', but you have file descriptors numbered at least as high as ' . $match[2] . '. ' . "You can install one of the extensions listed on {$helpLink} to support a higher number of " . "concurrent file descriptors. If a large number of open file descriptors is unexpected, you " . "might be leaking file descriptors that aren't closed correctly."; } } throw new \Exception($message, $errno); }; } /** * {@inheritdoc} * * @throws \Amp\Loop\UnsupportedFeatureException If the pcntl extension is not available. */ public function onSignal(int $signo, callable $callback, $data = null): string { if (!$this->signalHandling) { throw new UnsupportedFeatureException("Signal handling requires the pcntl extension"); } return parent::onSignal($signo, $callback, $data); } /** * {@inheritdoc} */ public function now(): int { $this->now = getCurrentTime() - $this->nowOffset; return $this->now; } /** * {@inheritdoc} */ public function getHandle() { return null; } /** * @param bool $blocking * * @return void * * @throws \Throwable */ protected function dispatch(bool $blocking) { $this->selectStreams( $this->readStreams, $this->writeStreams, $blocking ? $this->getTimeout() : 0 ); $now = $this->now(); while ($watcher = $this->timerQueue->extract($now)) { if ($watcher->type & Watcher::REPEAT) { $watcher->enabled = false; // Trick base class into adding to enable queue when calling enable() $this->enable($watcher->id); } else { $this->cancel($watcher->id); } try { // Execute the timer. $result = ($watcher->callback)($watcher->id, $watcher->data); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } } if ($this->signalHandling) { \pcntl_signal_dispatch(); } } /** * {@inheritdoc} * * @return void */ protected function activate(array $watchers) { foreach ($watchers as $watcher) { switch ($watcher->type) { case Watcher::READABLE: \assert(\is_resource($watcher->value)); $streamId = (int) $watcher->value; $this->readWatchers[$streamId][$watcher->id] = $watcher; $this->readStreams[$streamId] = $watcher->value; break; case Watcher::WRITABLE: \assert(\is_resource($watcher->value)); $streamId = (int) $watcher->value; $this->writeWatchers[$streamId][$watcher->id] = $watcher; $this->writeStreams[$streamId] = $watcher->value; break; case Watcher::DELAY: case Watcher::REPEAT: \assert(\is_int($watcher->value)); $this->timerQueue->insert($watcher); break; case Watcher::SIGNAL: \assert(\is_int($watcher->value)); if (!isset($this->signalWatchers[$watcher->value])) { if (!@\pcntl_signal($watcher->value, $this->callableFromInstanceMethod('handleSignal'))) { $message = "Failed to register signal handler"; if ($error = \error_get_last()) { $message .= \sprintf("; Errno: %d; %s", $error["type"], $error["message"]); } throw new \Error($message); } } $this->signalWatchers[$watcher->value][$watcher->id] = $watcher; break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } } } /** * {@inheritdoc} * * @return void */ protected function deactivate(Watcher $watcher) { switch ($watcher->type) { case Watcher::READABLE: $streamId = (int) $watcher->value; unset($this->readWatchers[$streamId][$watcher->id]); if (empty($this->readWatchers[$streamId])) { unset($this->readWatchers[$streamId], $this->readStreams[$streamId]); } break; case Watcher::WRITABLE: $streamId = (int) $watcher->value; unset($this->writeWatchers[$streamId][$watcher->id]); if (empty($this->writeWatchers[$streamId])) { unset($this->writeWatchers[$streamId], $this->writeStreams[$streamId]); } break; case Watcher::DELAY: case Watcher::REPEAT: $this->timerQueue->remove($watcher); break; case Watcher::SIGNAL: \assert(\is_int($watcher->value)); if (isset($this->signalWatchers[$watcher->value])) { unset($this->signalWatchers[$watcher->value][$watcher->id]); if (empty($this->signalWatchers[$watcher->value])) { unset($this->signalWatchers[$watcher->value]); @\pcntl_signal($watcher->value, \SIG_DFL); } } break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } } /** * @param resource[] $read * @param resource[] $write * @param int $timeout * * @return void */ private function selectStreams(array $read, array $write, int $timeout) { $timeout /= self::MILLISEC_PER_SEC; if (!empty($read) || !empty($write)) { // Use stream_select() if there are any streams in the loop. if ($timeout >= 0) { $seconds = (int) $timeout; $microseconds = (int) (($timeout - $seconds) * self::MICROSEC_PER_SEC); } else { $seconds = null; $microseconds = null; } // Failed connection attempts are indicated via except on Windows // @link https://github.com/reactphp/event-loop/blob/8bd064ce23c26c4decf186c2a5a818c9a8209eb0/src/StreamSelectLoop.php#L279-L287 // @link https://docs.microsoft.com/de-de/windows/win32/api/winsock2/nf-winsock2-select $except = null; if (\DIRECTORY_SEPARATOR === '\\') { $except = $write; } \set_error_handler($this->streamSelectErrorHandler); try { $result = \stream_select($read, $write, $except, $seconds, $microseconds); } finally { \restore_error_handler(); } if ($this->streamSelectIgnoreResult || $result === 0) { $this->streamSelectIgnoreResult = false; return; } if (!$result) { $this->error(new \Exception('Unknown error during stream_select')); return; } foreach ($read as $stream) { $streamId = (int) $stream; if (!isset($this->readWatchers[$streamId])) { continue; // All read watchers disabled. } foreach ($this->readWatchers[$streamId] as $watcher) { if (!isset($this->readWatchers[$streamId][$watcher->id])) { continue; // Watcher disabled by another IO watcher. } try { $result = ($watcher->callback)($watcher->id, $stream, $watcher->data); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } } } \assert(\is_array($write)); // See https://github.com/vimeo/psalm/issues/3036 if ($except) { foreach ($except as $key => $socket) { $write[$key] = $socket; } } foreach ($write as $stream) { $streamId = (int) $stream; if (!isset($this->writeWatchers[$streamId])) { continue; // All write watchers disabled. } foreach ($this->writeWatchers[$streamId] as $watcher) { if (!isset($this->writeWatchers[$streamId][$watcher->id])) { continue; // Watcher disabled by another IO watcher. } try { $result = ($watcher->callback)($watcher->id, $stream, $watcher->data); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } } } return; } if ($timeout < 0) { // Only signal watchers are enabled, so sleep indefinitely. \usleep(\PHP_INT_MAX); return; } if ($timeout > 0) { // Sleep until next timer expires. \usleep((int) ($timeout * self::MICROSEC_PER_SEC)); } } /** * @return int Milliseconds until next timer expires or -1 if there are no pending times. */ private function getTimeout(): int { $expiration = $this->timerQueue->peek(); if ($expiration === null) { return -1; } $expiration -= getCurrentTime() - $this->nowOffset; return $expiration > 0 ? $expiration : 0; } /** * @param int $signo * * @return void */ private function handleSignal(int $signo) { foreach ($this->signalWatchers[$signo] as $watcher) { if (!isset($this->signalWatchers[$signo][$watcher->id])) { continue; } try { $result = ($watcher->callback)($watcher->id, $signo, $watcher->data); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } } } } lib/Loop/TracingDriver.php000064400000014700144760120120011476 0ustar00driver = $driver; } public function run() { $this->driver->run(); } public function stop() { $this->driver->stop(); } public function defer(callable $callback, $data = null): string { $id = $this->driver->defer(function (...$args) use ($callback) { $this->cancel($args[0]); return $callback(...$args); }, $data); $this->creationTraces[$id] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); $this->enabledWatchers[$id] = true; return $id; } public function delay(int $delay, callable $callback, $data = null): string { $id = $this->driver->delay($delay, function (...$args) use ($callback) { $this->cancel($args[0]); return $callback(...$args); }, $data); $this->creationTraces[$id] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); $this->enabledWatchers[$id] = true; return $id; } public function repeat(int $interval, callable $callback, $data = null): string { $id = $this->driver->repeat($interval, $callback, $data); $this->creationTraces[$id] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); $this->enabledWatchers[$id] = true; return $id; } public function onReadable($stream, callable $callback, $data = null): string { $id = $this->driver->onReadable($stream, $callback, $data); $this->creationTraces[$id] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); $this->enabledWatchers[$id] = true; return $id; } public function onWritable($stream, callable $callback, $data = null): string { $id = $this->driver->onWritable($stream, $callback, $data); $this->creationTraces[$id] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); $this->enabledWatchers[$id] = true; return $id; } public function onSignal(int $signo, callable $callback, $data = null): string { $id = $this->driver->onSignal($signo, $callback, $data); $this->creationTraces[$id] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); $this->enabledWatchers[$id] = true; return $id; } public function enable(string $watcherId) { try { $this->driver->enable($watcherId); $this->enabledWatchers[$watcherId] = true; } catch (InvalidWatcherError $e) { throw new InvalidWatcherError( $watcherId, $e->getMessage() . "\r\n\r\n" . $this->getTraces($watcherId) ); } } public function cancel(string $watcherId) { $this->driver->cancel($watcherId); if (!isset($this->cancelTraces[$watcherId])) { $this->cancelTraces[$watcherId] = formatStacktrace(\debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS)); } unset($this->enabledWatchers[$watcherId], $this->unreferencedWatchers[$watcherId]); } public function disable(string $watcherId) { $this->driver->disable($watcherId); unset($this->enabledWatchers[$watcherId]); } public function reference(string $watcherId) { try { $this->driver->reference($watcherId); unset($this->unreferencedWatchers[$watcherId]); } catch (InvalidWatcherError $e) { throw new InvalidWatcherError( $watcherId, $e->getMessage() . "\r\n\r\n" . $this->getTraces($watcherId) ); } } public function unreference(string $watcherId) { $this->driver->unreference($watcherId); $this->unreferencedWatchers[$watcherId] = true; } public function setErrorHandler(callable $callback = null) { return $this->driver->setErrorHandler($callback); } /** @inheritdoc */ public function getHandle() { $this->driver->getHandle(); } public function dump(): string { $dump = "Enabled, referenced watchers keeping the loop running: "; foreach ($this->enabledWatchers as $watcher => $_) { if (isset($this->unreferencedWatchers[$watcher])) { continue; } $dump .= "Watcher ID: " . $watcher . "\r\n"; $dump .= $this->getCreationTrace($watcher); $dump .= "\r\n\r\n"; } return \rtrim($dump); } public function getInfo(): array { return $this->driver->getInfo(); } public function __debugInfo() { return $this->driver->__debugInfo(); } public function now(): int { return $this->driver->now(); } protected function error(\Throwable $exception) { $this->driver->error($exception); } /** * @inheritdoc * * @return void */ protected function activate(array $watchers) { // nothing to do in a decorator } /** * @inheritdoc * * @return void */ protected function dispatch(bool $blocking) { // nothing to do in a decorator } /** * @inheritdoc * * @return void */ protected function deactivate(Watcher $watcher) { // nothing to do in a decorator } private function getTraces(string $watcherId): string { return "Creation Trace:\r\n" . $this->getCreationTrace($watcherId) . "\r\n\r\n" . "Cancellation Trace:\r\n" . $this->getCancelTrace($watcherId); } private function getCreationTrace(string $watcher): string { if (!isset($this->creationTraces[$watcher])) { return 'No creation trace, yet.'; } return $this->creationTraces[$watcher]; } private function getCancelTrace(string $watcher): string { if (!isset($this->cancelTraces[$watcher])) { return 'No cancellation trace, yet.'; } return $this->cancelTraces[$watcher]; } } lib/Loop/UnsupportedFeatureException.php000064400000000421144760120120014451 0ustar00handle = \uv_loop_new(); /** * @param $event * @param $status * @param $events * @param $resource * * @return void */ $this->ioCallback = function ($event, $status, $events, $resource) { $watchers = $this->watchers[(int) $event]; switch ($status) { case 0: // OK break; default: // Invoke the callback on errors, as this matches behavior with other loop back-ends. // Re-enable watcher as libuv disables the watcher on non-zero status. $flags = 0; foreach ($watchers as $watcher) { $flags |= $watcher->enabled ? $watcher->type : 0; } \uv_poll_start($event, $flags, $this->ioCallback); break; } foreach ($watchers as $watcher) { // $events is OR'ed with 4 to trigger watcher if no events are indicated (0) or on UV_DISCONNECT (4). // http://docs.libuv.org/en/v1.x/poll.html if (!($watcher->enabled && ($watcher->type & $events || ($events | 4) === 4))) { continue; } try { $result = ($watcher->callback)($watcher->id, $resource, $watcher->data); if ($result === null) { continue; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } } }; /** * @param $event * * @return void */ $this->timerCallback = function ($event) { $watcher = $this->watchers[(int) $event][0]; if ($watcher->type & Watcher::DELAY) { unset($this->events[$watcher->id], $this->watchers[(int) $event]); // Avoid call to uv_is_active(). $this->cancel($watcher->id); // Remove reference to watcher in parent. } elseif ($watcher->value === 0) { // Disable and re-enable so it's not executed repeatedly in the same tick // See https://github.com/amphp/amp/issues/131 $this->disable($watcher->id); $this->enable($watcher->id); } try { $result = ($watcher->callback)($watcher->id, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; /** * @param $event * @param $signo * * @return void */ $this->signalCallback = function ($event, $signo) { $watcher = $this->watchers[(int) $event][0]; try { $result = ($watcher->callback)($watcher->id, $signo, $watcher->data); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { rethrow($result); } } catch (\Throwable $exception) { $this->error($exception); } }; } /** * {@inheritdoc} */ public function cancel(string $watcherId) { parent::cancel($watcherId); if (!isset($this->events[$watcherId])) { return; } $event = $this->events[$watcherId]; $eventId = (int) $event; if (isset($this->watchers[$eventId][0])) { // All except IO watchers. unset($this->watchers[$eventId]); } elseif (isset($this->watchers[$eventId][$watcherId])) { $watcher = $this->watchers[$eventId][$watcherId]; unset($this->watchers[$eventId][$watcherId]); if (empty($this->watchers[$eventId])) { unset($this->watchers[$eventId], $this->streams[(int) $watcher->value]); } } unset($this->events[$watcherId]); } public static function isSupported(): bool { return \extension_loaded("uv"); } /** * {@inheritdoc} */ public function now(): int { \uv_update_time($this->handle); /** @psalm-suppress TooManyArguments */ return \uv_now($this->handle); } /** * {@inheritdoc} */ public function getHandle() { return $this->handle; } /** * {@inheritdoc} * * @return void */ protected function dispatch(bool $blocking) { /** @psalm-suppress TooManyArguments */ \uv_run($this->handle, $blocking ? \UV::RUN_ONCE : \UV::RUN_NOWAIT); } /** * {@inheritdoc} * * @return void */ protected function activate(array $watchers) { $now = $this->now(); foreach ($watchers as $watcher) { $id = $watcher->id; switch ($watcher->type) { case Watcher::READABLE: case Watcher::WRITABLE: \assert(\is_resource($watcher->value)); $streamId = (int) $watcher->value; if (isset($this->streams[$streamId])) { $event = $this->streams[$streamId]; } elseif (isset($this->events[$id])) { $event = $this->streams[$streamId] = $this->events[$id]; } else { /** @psalm-suppress UndefinedFunction */ $event = $this->streams[$streamId] = \uv_poll_init_socket($this->handle, $watcher->value); } $eventId = (int) $event; $this->events[$id] = $event; $this->watchers[$eventId][$id] = $watcher; $flags = 0; foreach ($this->watchers[$eventId] as $w) { $flags |= $w->enabled ? $w->type : 0; } \uv_poll_start($event, $flags, $this->ioCallback); break; case Watcher::DELAY: case Watcher::REPEAT: \assert(\is_int($watcher->value)); if (isset($this->events[$id])) { $event = $this->events[$id]; } else { $event = $this->events[$id] = \uv_timer_init($this->handle); } $this->watchers[(int) $event] = [$watcher]; \uv_timer_start( $event, \max(0, $watcher->expiration - $now), ($watcher->type & Watcher::REPEAT) ? $watcher->value : 0, $this->timerCallback ); break; case Watcher::SIGNAL: \assert(\is_int($watcher->value)); if (isset($this->events[$id])) { $event = $this->events[$id]; } else { /** @psalm-suppress UndefinedFunction */ $event = $this->events[$id] = \uv_signal_init($this->handle); } $this->watchers[(int) $event] = [$watcher]; /** @psalm-suppress UndefinedFunction */ \uv_signal_start($event, $this->signalCallback, $watcher->value); break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } } } /** * {@inheritdoc} * * @return void */ protected function deactivate(Watcher $watcher) { $id = $watcher->id; if (!isset($this->events[$id])) { return; } $event = $this->events[$id]; if (!\uv_is_active($event)) { return; } switch ($watcher->type) { case Watcher::READABLE: case Watcher::WRITABLE: $flags = 0; foreach ($this->watchers[(int) $event] as $w) { $flags |= $w->enabled ? $w->type : 0; } if ($flags) { \uv_poll_start($event, $flags, $this->ioCallback); } else { \uv_poll_stop($event); } break; case Watcher::DELAY: case Watcher::REPEAT: \uv_timer_stop($event); break; case Watcher::SIGNAL: \uv_signal_stop($event); break; default: // @codeCoverageIgnoreStart throw new \Error("Unknown watcher type"); // @codeCoverageIgnoreEnd } } } lib/Loop/Watcher.php000064400000001775144760120120010340 0ustar00reasons = $reasons; } /** * @return \Throwable[] */ public function getReasons(): array { return $this->reasons; } } lib/NullCancellationToken.php000064400000001621144760120120012250 0ustar00throwIfRequested(); * } * ``` * * potentially multiple times, it allows writing * * ```php * $token = $token ?? new NullCancellationToken; * * // ... * * $token->throwIfRequested(); * ``` * * instead. */ final class NullCancellationToken implements CancellationToken { /** @inheritdoc */ public function subscribe(callable $callback): string { return "null-token"; } /** @inheritdoc */ public function unsubscribe(string $id) { // nothing to do } /** @inheritdoc */ public function isRequested(): bool { return false; } /** @inheritdoc */ public function throwIfRequested() { // nothing to do } } lib/Producer.php000064400000001754144760120120007612 0ustar00 */ final class Producer implements Iterator { /** * @use Internal\Producer */ use CallableMaker, Internal\Producer; /** * @param callable(callable(TValue):Promise):\Generator $producer * * @throws \Error Thrown if the callable does not return a Generator. */ public function __construct(callable $producer) { $result = $producer($this->callableFromInstanceMethod("emit")); if (!$result instanceof \Generator) { throw new \Error("The callable did not return a Generator"); } $coroutine = new Coroutine($result); $coroutine->onResolve(function ($exception) { if ($this->complete) { return; } if ($exception) { $this->fail($exception); return; } $this->complete(); }); } } lib/Promise.php000064400000002655144760120120007446 0ustar00, mixed, * mixed>|null) | callable(\Throwable|null, mixed): void $onResolved * * @return void */ public function onResolve(callable $onResolved); } lib/Struct.php000064400000004237144760120120007312 0ustar00generateStructPropertyError($property) ); } /** * @param string $property * @param mixed $value * * @psalm-return no-return */ public function __set(string $property, $value) { throw new \Error( $this->generateStructPropertyError($property) ); } private function generateStructPropertyError(string $property): string { $suggestion = $this->suggestPropertyName($property); $suggestStr = ($suggestion == "") ? "" : " ... did you mean \"{$suggestion}?\""; return \sprintf( "%s property \"%s\" does not exist%s", \str_replace("\0", "@", \get_class($this)), // Handle anonymous class names. $property, $suggestStr ); } private function suggestPropertyName(string $badProperty): string { $badProperty = \strtolower($badProperty); $bestMatch = ""; $bestMatchPercentage = 0; /** @psalm-suppress RawObjectIteration */ foreach ($this as $property => $value) { // Never suggest properties that begin with an underscore if ($property[0] === "_") { continue; } \similar_text($badProperty, \strtolower($property), $byRefPercentage); if ($byRefPercentage > $bestMatchPercentage) { $bestMatchPercentage = $byRefPercentage; $bestMatch = $property; } } return ($bestMatchPercentage >= $this->__propertySuggestThreshold) ? $bestMatch : ""; } } lib/Success.php000064400000002736144760120120007440 0ustar00 */ final class Success implements Promise { /** @var mixed */ private $value; /** * @param mixed $value Anything other than a Promise object. * * @psalm-param TValue $value * * @throws \Error If a promise is given as the value. */ public function __construct($value = null) { if ($value instanceof Promise || $value instanceof ReactPromise) { throw new \Error("Cannot use a promise as success value"); } $this->value = $value; } /** * {@inheritdoc} */ public function onResolve(callable $onResolved) { try { $result = $onResolved(null, $this->value); if ($result === null) { return; } if ($result instanceof \Generator) { $result = new Coroutine($result); } if ($result instanceof Promise || $result instanceof ReactPromise) { Promise\rethrow($result); } } catch (\Throwable $exception) { Loop::defer(static function () use ($exception) { throw $exception; }); } } } lib/TimeoutCancellationToken.php000064400000003410144760120120012762 0ustar00token = $source->getToken(); $trace = \debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS); $this->watcher = Loop::delay($timeout, static function () use ($source, $message, $trace) { $trace = formatStacktrace($trace); $source->cancel(new TimeoutException("$message\r\nTimeoutCancellationToken was created here:\r\n$trace")); }); Loop::unreference($this->watcher); } /** * Cancels the delay watcher. */ public function __destruct() { Loop::cancel($this->watcher); } /** * {@inheritdoc} */ public function subscribe(callable $callback): string { return $this->token->subscribe($callback); } /** * {@inheritdoc} */ public function unsubscribe(string $id) { $this->token->unsubscribe($id); } /** * {@inheritdoc} */ public function isRequested(): bool { return $this->token->isRequested(); } /** * {@inheritdoc} */ public function throwIfRequested() { $this->token->throwIfRequested(); } } lib/TimeoutException.php000064400000000557144760120120011334 0ustar00 * @template T as TReturn|Promise|\Generator * * @formatter:off * * @param callable(...mixed): T $callback * * @return callable * @psalm-return (T is Promise ? (callable(mixed...): Promise) : (T is \Generator ? (TGenerator is Promise ? (callable(mixed...): Promise) : (callable(mixed...): Promise)) : (callable(mixed...): Promise))) * * @formatter:on * * @see asyncCoroutine() * * @psalm-suppress InvalidReturnType */ function coroutine(callable $callback): callable { /** @psalm-suppress InvalidReturnStatement */ return static function (...$args) use ($callback): Promise { return call($callback, ...$args); }; } /** * Returns a new function that wraps $callback in a promise/coroutine-aware function that automatically runs * Generators as coroutines. The returned function always returns void when invoked. Errors are forwarded to the * loop's error handler using `Amp\Promise\rethrow()`. * * Use this function to create a coroutine-aware callable for a non-promise-aware callback caller. * * @param callable(...mixed): mixed $callback * * @return callable * @psalm-return callable(mixed...): void * * @see coroutine() */ function asyncCoroutine(callable $callback): callable { return static function (...$args) use ($callback) { Promise\rethrow(call($callback, ...$args)); }; } /** * Calls the given function, always returning a promise. If the function returns a Generator, it will be run as a * coroutine. If the function throws, a failed promise will be returned. * * @template TReturn * @template TPromise * @template TGeneratorReturn * @template TGeneratorPromise * * @template TGenerator as TGeneratorReturn|Promise * @template T as TReturn|Promise|\Generator * * @formatter:off * * @param callable(...mixed): T $callback * @param mixed ...$args Arguments to pass to the function. * * @return Promise * @psalm-return (T is Promise ? Promise : (T is \Generator ? (TGenerator is Promise ? Promise : Promise) : Promise)) * * @formatter:on */ function call(callable $callback, ...$args): Promise { try { $result = $callback(...$args); } catch (\Throwable $exception) { return new Failure($exception); } if ($result instanceof \Generator) { return new Coroutine($result); } if ($result instanceof Promise) { return $result; } if ($result instanceof ReactPromise) { return Promise\adapt($result); } return new Success($result); } /** * Calls the given function. If the function returns a Generator, it will be run as a coroutine. If the function * throws or returns a failing promise, the failure is forwarded to the loop error handler. * * @param callable(...mixed): mixed $callback * @param mixed ...$args Arguments to pass to the function. * * @return void */ function asyncCall(callable $callback, ...$args) { Promise\rethrow(call($callback, ...$args)); } /** * Sleeps for the specified number of milliseconds. * * @param int $milliseconds * * @return Delayed */ function delay(int $milliseconds): Delayed { return new Delayed($milliseconds); } /** * Returns the current time relative to an arbitrary point in time. * * @return int Time in milliseconds. */ function getCurrentTime(): int { return Internal\getCurrentTime(); } } namespace Amp\Promise { use Amp\Deferred; use Amp\Loop; use Amp\MultiReasonException; use Amp\Promise; use Amp\Success; use Amp\TimeoutException; use React\Promise\PromiseInterface as ReactPromise; use function Amp\call; use function Amp\Internal\createTypeError; /** * Registers a callback that will forward the failure reason to the event loop's error handler if the promise fails. * * Use this function if you neither return the promise nor handle a possible error yourself to prevent errors from * going entirely unnoticed. * * @param Promise|ReactPromise $promise Promise to register the handler on. * * @return void * @throws \TypeError If $promise is not an instance of \Amp\Promise or \React\Promise\PromiseInterface. * */ function rethrow($promise) { if (!$promise instanceof Promise) { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } else { throw createTypeError([Promise::class, ReactPromise::class], $promise); } } $promise->onResolve(static function ($exception) { if ($exception) { throw $exception; } }); } /** * Runs the event loop until the promise is resolved. Should not be called within a running event loop. * * Use this function only in synchronous contexts to wait for an asynchronous operation. Use coroutines and yield to * await promise resolution in a fully asynchronous application instead. * * @template TPromise * @template T as Promise|ReactPromise * * @param Promise|ReactPromise $promise Promise to wait for. * * @return mixed Promise success value. * * @psalm-param T $promise * @psalm-return (T is Promise ? TPromise : mixed) * * @throws \TypeError If $promise is not an instance of \Amp\Promise or \React\Promise\PromiseInterface. * @throws \Error If the event loop stopped without the $promise being resolved. * @throws \Throwable Promise failure reason. */ function wait($promise) { if (!$promise instanceof Promise) { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } else { throw createTypeError([Promise::class, ReactPromise::class], $promise); } } $resolved = false; try { Loop::run(function () use (&$resolved, &$value, &$exception, $promise) { $promise->onResolve(function ($e, $v) use (&$resolved, &$value, &$exception) { Loop::stop(); $resolved = true; $exception = $e; $value = $v; }); }); } catch (\Throwable $throwable) { throw new \Error("Loop exceptionally stopped without resolving the promise", 0, $throwable); } if (!$resolved) { throw new \Error("Loop stopped without resolving the promise"); } if ($exception) { throw $exception; } return $value; } /** * Creates an artificial timeout for any `Promise`. * * If the timeout expires before the promise is resolved, the returned promise fails with an instance of * `Amp\TimeoutException`. * * @template TReturn * * @param Promise|ReactPromise $promise Promise to which the timeout is applied. * @param int $timeout Timeout in milliseconds. * * @return Promise * * @throws \TypeError If $promise is not an instance of \Amp\Promise or \React\Promise\PromiseInterface. */ function timeout($promise, int $timeout): Promise { if (!$promise instanceof Promise) { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } else { throw createTypeError([Promise::class, ReactPromise::class], $promise); } } $deferred = new Deferred; $watcher = Loop::delay($timeout, static function () use (&$deferred) { $temp = $deferred; // prevent double resolve $deferred = null; $temp->fail(new TimeoutException); }); Loop::unreference($watcher); $promise->onResolve(function () use (&$deferred, $promise, $watcher) { if ($deferred !== null) { Loop::cancel($watcher); $deferred->resolve($promise); } }); return $deferred->promise(); } /** * Creates an artificial timeout for any `Promise`. * * If the promise is resolved before the timeout expires, the result is returned * * If the timeout expires before the promise is resolved, a default value is returned * * @template TReturn * * @param Promise|ReactPromise $promise Promise to which the timeout is applied. * @param int $timeout Timeout in milliseconds. * @param TReturn $default * * @return Promise * * @throws \TypeError If $promise is not an instance of \Amp\Promise or \React\Promise\PromiseInterface. */ function timeoutWithDefault($promise, int $timeout, $default = null): Promise { $promise = timeout($promise, $timeout); return call(static function () use ($promise, $default) { try { return yield $promise; } catch (TimeoutException $exception) { return $default; } }); } /** * Adapts any object with a done(callable $onFulfilled, callable $onRejected) or then(callable $onFulfilled, * callable $onRejected) method to a promise usable by components depending on placeholders implementing * \AsyncInterop\Promise. * * @param object $promise Object with a done() or then() method. * * @return Promise Promise resolved by the $thenable object. * * @throws \Error If the provided object does not have a then() method. */ function adapt($promise): Promise { if (!\is_object($promise)) { throw new \Error("Object must be provided"); } $deferred = new Deferred; if (\method_exists($promise, 'done')) { $promise->done([$deferred, 'resolve'], [$deferred, 'fail']); } elseif (\method_exists($promise, 'then')) { $promise->then([$deferred, 'resolve'], [$deferred, 'fail']); } else { throw new \Error("Object must have a 'then' or 'done' method"); } return $deferred->promise(); } /** * Returns a promise that is resolved when all promises are resolved. The returned promise will not fail. * Returned promise succeeds with a two-item array delineating successful and failed promise results, * with keys identical and corresponding to the original given array. * * This function is the same as some() with the notable exception that it will never fail even * if all promises in the array resolve unsuccessfully. * * @template TValue * * @param Promise[]|ReactPromise[] $promises * * @return Promise * * @throws \Error If a non-Promise is in the array. */ function any(array $promises): Promise { return some($promises, 0); } /** * Returns a promise that succeeds when all promises succeed, and fails if any promise fails. Returned * promise succeeds with an array of values used to succeed each contained promise, with keys corresponding to * the array of promises. * * @param Promise[]|ReactPromise[] $promises Array of only promises. * * @return Promise * * @throws \Error If a non-Promise is in the array. * * @template TValue * * @psalm-param array|ReactPromise> $promises * @psalm-assert array|ReactPromise> $promises $promises * @psalm-return Promise> */ function all(array $promises): Promise { if (empty($promises)) { return new Success([]); } $deferred = new Deferred; $result = $deferred->promise(); $pending = \count($promises); $values = []; foreach ($promises as $key => $promise) { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } elseif (!$promise instanceof Promise) { throw createTypeError([Promise::class, ReactPromise::class], $promise); } $values[$key] = null; // add entry to array to preserve order $promise->onResolve(function ($exception, $value) use (&$deferred, &$values, &$pending, $key) { if ($pending === 0) { return; } if ($exception) { $pending = 0; $deferred->fail($exception); $deferred = null; return; } $values[$key] = $value; if (0 === --$pending) { $deferred->resolve($values); } }); } return $result; } /** * Returns a promise that succeeds when the first promise succeeds, and fails only if all promises fail. * * @template TValue * * @param Promise[]|ReactPromise[] $promises Array of only promises. * * @return Promise * * @throws \Error If the array is empty or a non-Promise is in the array. */ function first(array $promises): Promise { if (empty($promises)) { throw new \Error("No promises provided"); } $deferred = new Deferred; $result = $deferred->promise(); $pending = \count($promises); $exceptions = []; foreach ($promises as $key => $promise) { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } elseif (!$promise instanceof Promise) { throw createTypeError([Promise::class, ReactPromise::class], $promise); } $exceptions[$key] = null; // add entry to array to preserve order $promise->onResolve(function ($error, $value) use (&$deferred, &$exceptions, &$pending, $key) { if ($pending === 0) { return; } if (!$error) { $pending = 0; $deferred->resolve($value); $deferred = null; return; } $exceptions[$key] = $error; if (0 === --$pending) { $deferred->fail(new MultiReasonException($exceptions)); } }); } return $result; } /** * Resolves with a two-item array delineating successful and failed Promise results. * * The returned promise will only fail if the given number of required promises fail. * * @template TValue * * @param Promise[]|ReactPromise[] $promises Array of only promises. * @param int $required Number of promises that must succeed for the * returned promise to succeed. * * @return Promise * * @throws \Error If a non-Promise is in the array. */ function some(array $promises, int $required = 1): Promise { if ($required < 0) { throw new \Error("Number of promises required must be non-negative"); } $pending = \count($promises); if ($required > $pending) { throw new \Error("Too few promises provided"); } if (empty($promises)) { return new Success([[], []]); } $deferred = new Deferred; $result = $deferred->promise(); $values = []; $exceptions = []; foreach ($promises as $key => $promise) { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } elseif (!$promise instanceof Promise) { throw createTypeError([Promise::class, ReactPromise::class], $promise); } $values[$key] = $exceptions[$key] = null; // add entry to arrays to preserve order $promise->onResolve(static function ($exception, $value) use ( &$values, &$exceptions, &$pending, $key, $required, $deferred ) { if ($exception) { $exceptions[$key] = $exception; unset($values[$key]); } else { $values[$key] = $value; unset($exceptions[$key]); } if (0 === --$pending) { if (\count($values) < $required) { $deferred->fail(new MultiReasonException($exceptions)); } else { $deferred->resolve([$exceptions, $values]); } } }); } return $result; } /** * Wraps a promise into another promise, altering the exception or result. * * @param Promise|ReactPromise $promise * @param callable $callback * * @return Promise */ function wrap($promise, callable $callback): Promise { if ($promise instanceof ReactPromise) { $promise = adapt($promise); } elseif (!$promise instanceof Promise) { throw createTypeError([Promise::class, ReactPromise::class], $promise); } $deferred = new Deferred(); $promise->onResolve(static function (\Throwable $exception = null, $result) use ($deferred, $callback) { try { $result = $callback($exception, $result); } catch (\Throwable $exception) { $deferred->fail($exception); return; } $deferred->resolve($result); }); return $deferred->promise(); } } namespace Amp\Iterator { use Amp\Delayed; use Amp\Emitter; use Amp\Iterator; use Amp\Producer; use Amp\Promise; use function Amp\call; use function Amp\coroutine; use function Amp\Internal\createTypeError; /** * Creates an iterator from the given iterable, emitting the each value. The iterable may contain promises. If any * promise fails, the iterator will fail with the same reason. * * @param array|\Traversable $iterable Elements to emit. * @param int $delay Delay between element emissions in milliseconds. * * @return Iterator * * @throws \TypeError If the argument is not an array or instance of \Traversable. */ function fromIterable(/* iterable */ $iterable, int $delay = 0 ): Iterator { if (!$iterable instanceof \Traversable && !\is_array($iterable)) { throw createTypeError(["array", "Traversable"], $iterable); } if ($delay) { return new Producer(static function (callable $emit) use ($iterable, $delay) { foreach ($iterable as $value) { yield new Delayed($delay); yield $emit($value); } }); } return new Producer(static function (callable $emit) use ($iterable) { foreach ($iterable as $value) { yield $emit($value); } }); } /** * @template TValue * @template TReturn * * @param Iterator $iterator * @param callable (TValue $value): TReturn $onEmit * * @return Iterator */ function map(Iterator $iterator, callable $onEmit): Iterator { return new Producer(static function (callable $emit) use ($iterator, $onEmit) { while (yield $iterator->advance()) { yield $emit($onEmit($iterator->getCurrent())); } }); } /** * @template TValue * * @param Iterator $iterator * @param callable(TValue $value):bool $filter * * @return Iterator */ function filter(Iterator $iterator, callable $filter): Iterator { return new Producer(static function (callable $emit) use ($iterator, $filter) { while (yield $iterator->advance()) { if ($filter($iterator->getCurrent())) { yield $emit($iterator->getCurrent()); } } }); } /** * Creates an iterator that emits values emitted from any iterator in the array of iterators. * * @param Iterator[] $iterators * * @return Iterator */ function merge(array $iterators): Iterator { $emitter = new Emitter; $result = $emitter->iterate(); $coroutine = coroutine(static function (Iterator $iterator) use (&$emitter) { while ((yield $iterator->advance()) && $emitter !== null) { yield $emitter->emit($iterator->getCurrent()); } }); $coroutines = []; foreach ($iterators as $iterator) { if (!$iterator instanceof Iterator) { throw createTypeError([Iterator::class], $iterator); } $coroutines[] = $coroutine($iterator); } Promise\all($coroutines)->onResolve(static function ($exception) use (&$emitter) { if ($exception) { $emitter->fail($exception); $emitter = null; } else { $emitter->complete(); } }); return $result; } /** * Concatenates the given iterators into a single iterator, emitting values from a single iterator at a time. The * prior iterator must complete before values are emitted from any subsequent iterators. Iterators are concatenated * in the order given (iteration order of the array). * * @param Iterator[] $iterators * * @return Iterator */ function concat(array $iterators): Iterator { foreach ($iterators as $iterator) { if (!$iterator instanceof Iterator) { throw createTypeError([Iterator::class], $iterator); } } $emitter = new Emitter; $previous = []; $promise = Promise\all($previous); $coroutine = coroutine(static function (Iterator $iterator, callable $emit) { while (yield $iterator->advance()) { yield $emit($iterator->getCurrent()); } }); foreach ($iterators as $iterator) { $emit = coroutine(static function ($value) use ($emitter, $promise) { static $pending = true, $failed = false; if ($failed) { return; } if ($pending) { try { yield $promise; $pending = false; } catch (\Throwable $exception) { $failed = true; return; // Prior iterator failed. } } yield $emitter->emit($value); }); $previous[] = $coroutine($iterator, $emit); $promise = Promise\all($previous); } $promise->onResolve(static function ($exception) use ($emitter) { if ($exception) { $emitter->fail($exception); return; } $emitter->complete(); }); return $emitter->iterate(); } /** * Discards all remaining items and returns the number of discarded items. * * @template TValue * * @param Iterator $iterator * * @return Promise * * @psalm-param Iterator $iterator * @psalm-return Promise */ function discard(Iterator $iterator): Promise { return call(static function () use ($iterator): \Generator { $count = 0; while (yield $iterator->advance()) { $count++; } return $count; }); } /** * Collects all items from an iterator into an array. * * @template TValue * * @param Iterator $iterator * * @psalm-param Iterator $iterator * * @return Promise * @psalm-return Promise> */ function toArray(Iterator $iterator): Promise { return call(static function () use ($iterator) { /** @psalm-var list $array */ $array = []; while (yield $iterator->advance()) { $array[] = $iterator->getCurrent(); } return $array; }); } } psalm.examples.xml000064400000001075144760120120010217 0ustar00 psalm.xml000064400000004222144760120120006377 0ustar00