LICENSE000064400000002161144760112760005561 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.json000064400000003415144760112760007301 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" }, "require-dev": { "ext-json": "*", "amphp/phpunit-util": "^1", "amphp/php-cs-fixer-config": "dev-master", "react/promise": "^2", "phpunit/phpunit": "^6.0.9 | ^7", "vimeo/psalm": "^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.php000064400000005224144760112760010515 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.php000064400000002757144760112760011443 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.php000064400000000524144760112760011565 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.php000064400000012562144760112760010010 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.php000064400000003074144760112760007557 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; } }; $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); } } lib/Delayed.php000064400000002324144760112760007403 0ustar00 */ final class Delayed implements Promise { use Internal\Placeholder; /** @var string 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->resolve($value); }); } /** * References the internal watcher in the event loop, keeping the loop running while this promise is pending. * * @return void */ public function reference() { Loop::reference($this->watcher); } /** * 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 void */ public function unreference() { Loop::unreference($this->watcher); } } lib/Emitter.php000064400000003767144760112760007461 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.php000064400000002256144760112760007427 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.php000064400000011746144760112760012042 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)); } } lib/Internal/PrivateIterator.php000064400000001572144760112760012740 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.php000064400000001006144760112760012555 0ustar00promise = $promise; } public function onResolve(callable $onResolved) { $this->promise->onResolve($onResolved); } } lib/Internal/Producer.php000064400000013266144760112760011402 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.php000064400000005444144760112760012766 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.php000064400000007060144760112760011622 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.php000064400000002144144760112760011423 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.php000064400000002105144760112760007622 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.php000064400000002261144760112760010312 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.php000064400000042620144760112760006750 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.php000064400000063062144760112760010206 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->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->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; 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.php000064400000003125144760112760011530 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.php000064400000021072144760112760010474 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 { if ($this->nowUpdateNeeded) { $this->now = getCurrentTime() - $this->nowOffset; $this->nowUpdateNeeded = false; } return $this->now; } /** * {@inheritdoc} */ public function getHandle(): \EvLoop { return $this->handle; } /** * {@inheritdoc} * * @return void */ protected function dispatch(bool $blocking) { $this->nowUpdateNeeded = true; $this->handle->run($blocking ? \Ev::RUN_ONCE : \Ev::RUN_ONCE | \Ev::RUN_NOWAIT); } /** * {@inheritdoc} * * @return void */ protected function activate(array $watchers) { 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( $interval, ($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.php000064400000024036144760112760011206 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 { if ($this->nowUpdateNeeded) { $this->now = getCurrentTime() - $this->nowOffset; $this->nowUpdateNeeded = false; } return $this->now; } /** * {@inheritdoc} */ public function getHandle(): \EventBase { return $this->handle; } /** * {@inheritdoc} * * @return void */ protected function dispatch(bool $blocking) { $this->nowUpdateNeeded = true; $this->handle->loop($blocking ? \EventBase::LOOP_ONCE : \EventBase::LOOP_ONCE | \EventBase::LOOP_NONBLOCK); } /** * {@inheritdoc} * * @return void */ protected function activate(array $watchers) { $now = getCurrentTime() - $this->nowOffset; 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 = $watcher->value - ($now - $this->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.php000064400000007716144760112760012620 0ustar00 $watcher * * @return void */ public function insert(Watcher $watcher, int $expiration) { $entry = new TimerQueueEntry($watcher, $expiration); $node = \count($this->data); $this->data[$node] = $entry; $this->pointers[$watcher->id] = $node; while ($node !== 0 && $entry->expiration < $this->data[$parent = ($node - 1) >> 1]->expiration) { $temp = $this->data[$parent]; $this->data[$node] = $temp; $this->pointers[$temp->watcher->id] = $node; $this->data[$parent] = $entry; $this->pointers[$watcher->id] = $parent; $node = $parent; } } /** * 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; } $data = $this->data[0]; if ($data->expiration > $now) { return null; } $this->removeAndRebuild(0); return $data->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 from that node downward. * * @return void */ private function removeAndRebuild(int $node) { $length = \count($this->data) - 1; $id = $this->data[$node]->watcher->id; $left = $this->data[$node] = $this->data[$length]; $this->pointers[$left->watcher->id] = $node; unset($this->data[$length], $this->pointers[$id]); 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; } $left = $this->data[$node]; $right = $this->data[$swap]; $this->data[$node] = $right; $this->pointers[$right->watcher->id] = $node; $this->data[$swap] = $left; $this->pointers[$left->watcher->id] = $swap; $node = $swap; } } } lib/Loop/Internal/TimerQueueEntry.php000064400000000731144760112760013630 0ustar00watcher = $watcher; $this->expiration = $expiration; } } lib/Loop/InvalidWatcherError.php000064400000001425144760112760012664 0ustar00watcherId = $watcherId; parent::__construct($message); } /** * @return string The watcher identifier. */ public function getWatcherId() { return $this->watcherId; } } lib/Loop/NativeDriver.php000064400000030764144760112760011360 0ustar00timerQueue = new Internal\TimerQueue; $this->signalHandling = \extension_loaded("pcntl"); $this->nowOffset = getCurrentTime(); $this->now = \random_int(0, $this->nowOffset); $this->nowOffset -= $this->now; } /** * {@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 { if ($this->nowUpdateNeeded) { $this->now = getCurrentTime() - $this->nowOffset; $this->nowUpdateNeeded = false; } return $this->now; } /** * {@inheritdoc} */ public function getHandle() { return null; } /** * @param bool $blocking * * @return void * * @throws \Throwable */ protected function dispatch(bool $blocking) { $this->nowUpdateNeeded = true; $this->selectStreams( $this->readStreams, $this->writeStreams, $blocking ? $this->getTimeout() : 0 ); $scheduleQueue = []; try { $now = $this->now(); while ($watcher = $this->timerQueue->extract($now)) { if ($watcher->type & Watcher::REPEAT) { $expiration = $now + $watcher->value; $scheduleQueue[] = [$watcher, $expiration]; } 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); } } } finally { foreach ($scheduleQueue as list($watcher, $expiration)) { if ($watcher->enabled) { $this->timerQueue->insert($watcher, $expiration); } } } if ($this->signalHandling) { \pcntl_signal_dispatch(); } } /** * @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; } $except = null; // Error reporting suppressed since stream_select() emits an E_WARNING if it is interrupted by a signal. if (!($result = @\stream_select($read, $write, $except, $seconds, $microseconds))) { if ($result === 0) { return; } $error = \error_get_last(); if (\strpos($error["message"] ?? '', "unable to select") !== 0) { return; } $this->error(new \Exception($error["message"] ?? 'Unknown error during stream_select')); } 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 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) { // Otherwise sleep with usleep() if $timeout > 0. \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; } /** * {@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)); $expiration = $this->now() + $watcher->value; $this->timerQueue->insert($watcher, $expiration); 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 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.php000064400000014700144760112760011511 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.php000064400000000421144760112760014464 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]); } else { $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 { /** @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) { 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, $watcher->value, ($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.php000064400000001713144760112760010343 0ustar00reasons = $reasons; } /** * @return \Throwable[] */ public function getReasons(): array { return $this->reasons; } } lib/NullCancellationToken.php000064400000001621144760112760012263 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.php000064400000001754144760112760007625 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.php000064400000002655144760112760007461 0ustar00, mixed, * mixed>|null) | callable(\Throwable|null, mixed): void $onResolved * * @return void */ public function onResolve(callable $onResolved); } lib/Struct.php000064400000004237144760112760007325 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.php000064400000002736144760112760007453 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.php000064400000003410144760112760012775 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.php000064400000000557144760112760011347 0ustar00|\Generator>) $callback * @param mixed ...$args Arguments to pass to the function. * * @return Promise * @psalm-return (TReturn is Promise ? Promise : (TReturn 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. * * @template TReturn * @template TReturnPromise * @template TGenerator * @template TGeneratorPromise * * @formatter:off * * @param callable(...mixed):(TReturn|Promise|\Generator>) $callback * @param mixed ...$args Arguments to pass to the function. * * @return void * * @formatter:on * * @throws \TypeError */ 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. * * @param Promise|ReactPromise $promise Promise to wait for. * * @return mixed Promise success value. * * @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 { $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. * * @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. * * @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. * * @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(); } /** * 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.xml000064400000004066144760112760006420 0ustar00