class UvDriver
Hierarchy
- class \Revolt\EventLoop\Internal\AbstractDriver implements \Revolt\EventLoop\Driver
- class \Revolt\EventLoop\Driver\UvDriver extends \Revolt\EventLoop\Internal\AbstractDriver
Expanded class hierarchy of UvDriver
1 file declares its use of UvDriver
- DriverFactory.php in vendor/
revolt/ event-loop/ src/ EventLoop/ DriverFactory.php
File
-
vendor/
revolt/ event-loop/ src/ EventLoop/ Driver/ UvDriver.php, line 15
Namespace
Revolt\EventLoop\DriverView source
final class UvDriver extends AbstractDriver {
public static function isSupported() : bool {
return \extension_loaded("uv");
}
/** @var resource|\UVLoop A uv_loop resource created with uv_loop_new() */
private $handle;
/** @var array<string, resource> */
private array $events = [];
/** @var array<int, array<array-key, DriverCallback>> */
private array $uvCallbacks = [];
/** @var array<int, resource> */
private array $streams = [];
private readonly \Closure $ioCallback;
private readonly \Closure $timerCallback;
private readonly \Closure $signalCallback;
public function __construct() {
parent::__construct();
$this->handle = \uv_loop_new();
$this->ioCallback = function ($event, $status, $events, $resource) : void {
$callbacks = $this->uvCallbacks[(int) $event];
// Invoke the callback on errors, as this matches behavior with other loop back-ends.
// Re-enable callback as libuv disables the callback on non-zero status.
if ($status !== 0) {
$flags = 0;
foreach ($callbacks as $callback) {
\assert($callback instanceof StreamCallback);
$flags |= $callback->invokable ? $this->getStreamCallbackFlags($callback) : 0;
}
\uv_poll_start($event, $flags, $this->ioCallback);
}
foreach ($callbacks as $callback) {
\assert($callback instanceof StreamCallback);
// $events is ORed with 4 to trigger callback if no events are indicated (0) or on UV_DISCONNECT (4).
// http://docs.libuv.org/en/v1.x/poll.html
if (!($this->getStreamCallbackFlags($callback) & $events || ($events | 4) === 4)) {
continue;
}
$this->enqueueCallback($callback);
}
};
$this->timerCallback = function ($event) : void {
$callback = $this->uvCallbacks[(int) $event][0];
\assert($callback instanceof TimerCallback);
$this->enqueueCallback($callback);
};
$this->signalCallback = function ($event) : void {
$callback = $this->uvCallbacks[(int) $event][0];
$this->enqueueCallback($callback);
};
}
/**
* {@inheritdoc}
*/
public function cancel(string $callbackId) : void {
parent::cancel($callbackId);
if (!isset($this->events[$callbackId])) {
return;
}
$event = $this->events[$callbackId];
$eventId = (int) $event;
if (isset($this->uvCallbacks[$eventId][0])) {
// All except IO callbacks.
unset($this->uvCallbacks[$eventId]);
}
elseif (isset($this->uvCallbacks[$eventId][$callbackId])) {
$callback = $this->uvCallbacks[$eventId][$callbackId];
unset($this->uvCallbacks[$eventId][$callbackId]);
\assert($callback instanceof StreamCallback);
if (empty($this->uvCallbacks[$eventId])) {
unset($this->uvCallbacks[$eventId], $this->streams[(int) $callback->stream]);
}
}
unset($this->events[$callbackId]);
}
/**
* @return \UVLoop|resource
*/
public function getHandle() : mixed {
return $this->handle;
}
protected function now() : float {
\uv_update_time($this->handle);
/** @psalm-suppress TooManyArguments */
return \uv_now($this->handle) / 1000;
}
/**
* {@inheritdoc}
*/
protected function dispatch(bool $blocking) : void {
/** @psalm-suppress TooManyArguments */
\uv_run($this->handle, $blocking ? \UV::RUN_ONCE : \UV::RUN_NOWAIT);
}
/**
* {@inheritdoc}
*/
protected function activate(array $callbacks) : void {
$now = $this->now();
foreach ($callbacks as $callback) {
$id = $callback->id;
if ($callback instanceof StreamCallback) {
\assert(\is_resource($callback->stream));
$streamId = (int) $callback->stream;
if (isset($this->streams[$streamId])) {
$event = $this->streams[$streamId];
}
elseif (isset($this->events[$id])) {
$event = $this->streams[$streamId] = $this->events[$id];
}
else {
/** @psalm-suppress TooManyArguments */
$event = $this->streams[$streamId] = \uv_poll_init_socket($this->handle, $callback->stream);
}
$eventId = (int) $event;
$this->events[$id] = $event;
$this->uvCallbacks[$eventId][$id] = $callback;
$flags = 0;
foreach ($this->uvCallbacks[$eventId] as $w) {
\assert($w instanceof StreamCallback);
$flags |= $w->enabled ? $this->getStreamCallbackFlags($w) : 0;
}
\uv_poll_start($event, $flags, $this->ioCallback);
}
elseif ($callback instanceof TimerCallback) {
if (isset($this->events[$id])) {
$event = $this->events[$id];
}
else {
$event = $this->events[$id] = \uv_timer_init($this->handle);
}
$this->uvCallbacks[(int) $event] = [
$callback,
];
\uv_timer_start($event, (int) \min(\max(0, \ceil(($callback->expiration - $now) * 1000)), \PHP_INT_MAX), $callback->repeat ? (int) \min(\max(0, \ceil($callback->interval * 1000)), \PHP_INT_MAX) : 0, $this->timerCallback);
}
elseif ($callback instanceof SignalCallback) {
if (isset($this->events[$id])) {
$event = $this->events[$id];
}
else {
/** @psalm-suppress TooManyArguments */
$event = $this->events[$id] = \uv_signal_init($this->handle);
}
$this->uvCallbacks[(int) $event] = [
$callback,
];
/** @psalm-suppress TooManyArguments */
\uv_signal_start($event, $this->signalCallback, $callback->signal);
}
else {
// @codeCoverageIgnoreStart
throw new \Error("Unknown callback type");
// @codeCoverageIgnoreEnd
}
}
}
/**
* {@inheritdoc}
*/
protected function deactivate(DriverCallback $callback) : void {
$id = $callback->id;
if (!isset($this->events[$id])) {
return;
}
$event = $this->events[$id];
if (!\uv_is_active($event)) {
return;
}
if ($callback instanceof StreamCallback) {
$flags = 0;
foreach ($this->uvCallbacks[(int) $event] as $w) {
\assert($w instanceof StreamCallback);
$flags |= $w->invokable ? $this->getStreamCallbackFlags($w) : 0;
}
if ($flags) {
\uv_poll_start($event, $flags, $this->ioCallback);
}
else {
\uv_poll_stop($event);
}
}
elseif ($callback instanceof TimerCallback) {
\uv_timer_stop($event);
}
elseif ($callback instanceof SignalCallback) {
\uv_signal_stop($event);
}
else {
// @codeCoverageIgnoreStart
throw new \Error("Unknown callback type");
// @codeCoverageIgnoreEnd
}
}
private function getStreamCallbackFlags(StreamCallback $callback) : int {
if ($callback instanceof StreamWritableCallback) {
return \UV::WRITABLE;
}
if ($callback instanceof StreamReadableCallback) {
return \UV::READABLE;
}
throw new \Error('Invalid callback type');
}
}
Members
Title Sort descending | Modifiers | Object type | Summary | Overriden Title | Overrides |
---|---|---|---|---|---|
AbstractDriver::$callbackFiber | private | property | |||
AbstractDriver::$callbackQueue | private | property | @var \SplQueue<DriverCallback> | ||
AbstractDriver::$callbacks | private | property | @var array<string, DriverCallback> | ||
AbstractDriver::$enableDeferQueue | private | property | @var array<string, DriverCallback> | ||
AbstractDriver::$enableQueue | private | property | @var array<string, DriverCallback> | ||
AbstractDriver::$errorCallback | private | property | |||
AbstractDriver::$errorHandler | private | property | @var null|\Closure(\Throwable):void | ||
AbstractDriver::$fiber | private | property | |||
AbstractDriver::$idle | private | property | |||
AbstractDriver::$internalSuspensionMarker | private | property | |||
AbstractDriver::$interrupt | private | property | @var null|\Closure():mixed | ||
AbstractDriver::$interruptCallback | private | property | |||
AbstractDriver::$microtaskQueue | private | property | @var \SplQueue<array{\Closure, array}> | ||
AbstractDriver::$nextId | private | property | @var string Next callback identifier. | ||
AbstractDriver::$queueCallback | private | property | |||
AbstractDriver::$runCallback | private | property | |||
AbstractDriver::$stopped | private | property | |||
AbstractDriver::$suspensions | private | property | @var \WeakMap<object, \WeakReference<DriverSuspension>> | ||
AbstractDriver::createCallbackFiber | private | function | |||
AbstractDriver::createErrorCallback | private | function | |||
AbstractDriver::createLoopFiber | private | function | |||
AbstractDriver::defer | public | function | Defer the execution of a callback. | Overrides Driver::defer | |
AbstractDriver::delay | public | function | Delay the execution of a callback. | Overrides Driver::delay | |
AbstractDriver::disable | public | function | Disable a callback immediately. | Overrides Driver::disable | |
AbstractDriver::enable | public | function | Enable a callback to be active starting in the next tick. | Overrides Driver::enable | |
AbstractDriver::enqueueCallback | final protected | function | |||
AbstractDriver::error | final protected | function | Invokes the error handler with the given exception. | ||
AbstractDriver::getErrorHandler | public | function | Gets the error handler closure or {@code null} if none is set. | Overrides Driver::getErrorHandler | |
AbstractDriver::getIdentifiers | public | function | Returns all registered non-cancelled callback identifiers. | Overrides Driver::getIdentifiers | |
AbstractDriver::getSuspension | public | function | Returns an object used to suspend and resume execution of the current fiber or {main}. | Overrides Driver::getSuspension | |
AbstractDriver::getType | public | function | Returns the type of the callback identified by the given callback identifier. | Overrides Driver::getType | |
AbstractDriver::invokeCallbacks | private | function | |||
AbstractDriver::invokeInterrupt | private | function | |||
AbstractDriver::invokeMicrotasks | private | function | |||
AbstractDriver::isEmpty | private | function | |||
AbstractDriver::isEnabled | public | function | Returns whether the callback identified by the given callback identifier is currently enabled. | Overrides Driver::isEnabled | |
AbstractDriver::isReferenced | public | function | Returns whether the callback identified by the given callback identifier is currently referenced. | Overrides Driver::isReferenced | |
AbstractDriver::isRunning | public | function | Overrides Driver::isRunning | ||
AbstractDriver::onReadable | public | function | Execute a callback when a stream resource becomes readable or is closed for reading. | Overrides Driver::onReadable | |
AbstractDriver::onSignal | public | function | Execute a callback when a signal is received. | Overrides Driver::onSignal | 1 |
AbstractDriver::onWritable | public | function | Execute a callback when a stream resource becomes writable or is closed for writing. | Overrides Driver::onWritable | |
AbstractDriver::queue | public | function | Queue a microtask. | Overrides Driver::queue | |
AbstractDriver::reference | public | function | Reference a callback. | Overrides Driver::reference | |
AbstractDriver::repeat | public | function | Repeatedly execute a callback. | Overrides Driver::repeat | |
AbstractDriver::run | public | function | Run the event loop. | Overrides Driver::run | 2 |
AbstractDriver::setErrorHandler | public | function | Set a callback to be executed when an error occurs. | Overrides Driver::setErrorHandler | |
AbstractDriver::setInterrupt | private | function | |||
AbstractDriver::stop | public | function | Stop the event loop. | Overrides Driver::stop | 2 |
AbstractDriver::tick | private | function | Executes a single tick of the event loop. | ||
AbstractDriver::unreference | public | function | Unreference a callback. | Overrides Driver::unreference | |
AbstractDriver::__debugInfo | public | function | Returns some useful information about the event loop. | Overrides Driver::__debugInfo | |
UvDriver::$events | private | property | @var array<string, resource> | ||
UvDriver::$handle | private | property | @var resource|\UVLoop A uv_loop resource created with uv_loop_new() | ||
UvDriver::$ioCallback | private | property | |||
UvDriver::$signalCallback | private | property | |||
UvDriver::$streams | private | property | @var array<int, resource> | ||
UvDriver::$timerCallback | private | property | |||
UvDriver::$uvCallbacks | private | property | @var array<int, array<array-key, DriverCallback>> | ||
UvDriver::activate | protected | function | Activates (enables) all the given callbacks. | Overrides AbstractDriver::activate | |
UvDriver::cancel | public | function | Cancel a callback. | Overrides AbstractDriver::cancel | |
UvDriver::deactivate | protected | function | Deactivates (disables) the given callback. | Overrides AbstractDriver::deactivate | |
UvDriver::dispatch | protected | function | Dispatches any pending read/write, timer, and signal events. | Overrides AbstractDriver::dispatch | |
UvDriver::getHandle | public | function | Overrides Driver::getHandle | ||
UvDriver::getStreamCallbackFlags | private | function | |||
UvDriver::isSupported | public static | function | |||
UvDriver::now | protected | function | Returns the current event loop time in second increments. | Overrides AbstractDriver::now | |
UvDriver::__construct | public | function | Overrides AbstractDriver::__construct |