1
0
mirror of https://github.com/danog/amp.git synced 2024-12-15 19:07:22 +01:00
amp/lib/LibeventReactor.php

472 lines
14 KiB
PHP
Raw Normal View History

2013-08-05 22:05:08 +02:00
<?php
2014-09-23 04:38:32 +02:00
namespace Amp;
2013-08-05 22:05:08 +02:00
2014-04-23 05:09:42 +02:00
class LibeventReactor implements SignalReactor {
2013-08-05 22:05:08 +02:00
private $base;
private $watchers = [];
private $immediates = [];
2014-08-21 19:54:02 +02:00
private $lastWatcherId = 1;
private $resolution = 1000;
private $isRunning = false;
private $isGCScheduled = false;
private $garbage = [];
2013-08-05 22:05:08 +02:00
private $gcEvent;
private $stopException;
2014-09-22 22:47:48 +02:00
private $onGeneratorError;
2013-08-05 22:05:08 +02:00
2014-02-23 22:26:28 +01:00
public function __construct() {
$this->base = event_base_new();
2013-08-05 22:05:08 +02:00
$this->gcEvent = event_new();
event_timer_set($this->gcEvent, [$this, 'collectGarbage']);
event_base_set($this->gcEvent, $this->base);
2014-09-22 22:47:48 +02:00
$this->onGeneratorError = function($e, $r) {
if ($e) {
throw $e;
}
};
2013-08-05 22:05:08 +02:00
}
2014-08-06 22:01:24 +02:00
/**
* Start the event reactor and assume program flow control
*
* @param callable $onStart Optional callback to invoke immediately upon reactor start
* @throws \Exception Will throw if code executed during the event loop throws
* @return void
*/
public function run(callable $onStart = null) {
if ($this->isRunning) {
return;
}
$this->isRunning = true;
if ($onStart) {
$this->immediately($onStart);
}
while ($this->isRunning) {
if ($this->immediates && !$this->doImmediates()) {
break;
}
event_base_loop($this->base, EVLOOP_ONCE | EVLOOP_NONBLOCK);
}
if ($this->stopException) {
$e = $this->stopException;
$this->stopException = null;
throw $e;
}
}
private function doImmediates() {
$immediates = $this->immediates;
foreach ($immediates as $watcherId => $callback) {
$result = $callback($this, $watcherId);
if ($result instanceof \Generator) {
resolve($this, $result)->when($this->onGeneratorError);
}
unset(
$this->immediates[$watcherId],
$this->watchers[$watcherId]
);
if (!$this->isRunning) {
// If a watcher stops the reactor break out of the loop
break;
}
2013-08-05 22:05:08 +02:00
}
return $this->isRunning;
2013-08-05 22:05:08 +02:00
}
2014-08-06 22:01:24 +02:00
/**
* Execute a single event loop iteration
*
* @throws \Exception will throw any uncaught exception encountered during the loop iteration
* @return void
*/
2014-02-23 22:26:28 +01:00
public function tick() {
if ($this->isRunning) {
return;
2013-08-05 22:05:08 +02:00
}
$this->isRunning = true;
if (empty($this->immediates) || $this->doImmediates()) {
event_base_loop($this->base, EVLOOP_ONCE | EVLOOP_NONBLOCK);
}
$this->isRunning = false;
2013-08-05 22:05:08 +02:00
if ($this->stopException) {
$e = $this->stopException;
$this->stopException = null;
2013-08-05 22:05:08 +02:00
throw $e;
}
}
2014-08-06 22:01:24 +02:00
/**
* Stop the event reactor
*
* @return void
*/
2014-02-23 22:26:28 +01:00
public function stop() {
2013-08-05 22:05:08 +02:00
event_base_loopexit($this->base);
$this->isRunning = false;
2013-08-05 22:05:08 +02:00
}
2014-08-06 22:01:24 +02:00
/**
* Schedule an event to trigger once at the specified time
*
* @param callable $callback Any valid PHP callable
* @param mixed[int|string] $unixTimeOrStr A future unix timestamp or string parsable by strtotime()
* @throws \InvalidArgumentException On invalid future time
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
public function at(callable $callback, $unixTimeOrStr) {
2013-08-05 22:05:08 +02:00
$now = time();
if (is_int($unixTimeOrStr) && $unixTimeOrStr > $now) {
$secondsUntil = ($unixTimeOrStr - $now);
} elseif (($executeAt = @strtotime($unixTimeOrStr)) && $executeAt > $now) {
$secondsUntil = ($executeAt - $now);
} else {
2013-08-05 22:05:08 +02:00
throw new \InvalidArgumentException(
'Unix timestamp or future time string (parsable by strtotime()) required'
2013-08-05 22:05:08 +02:00
);
}
$msDelay = $secondsUntil * $this->resolution;
return $this->once($callback, $msDelay);
2013-08-05 22:05:08 +02:00
}
2014-08-06 22:01:24 +02:00
/**
* Schedule a callback for immediate invocation in the next event loop tick
2014-08-06 22:01:24 +02:00
*
* @param callable $callback Any valid PHP callable
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
2014-02-23 22:26:28 +01:00
public function immediately(callable $callback) {
$watcherId = (string) $this->lastWatcherId++;
$this->immediates[$watcherId] = $callback;
$watcher = new \StdClass;
$watcher->id = $watcherId;
$watcher->callback = $callback;
$watcher->isEnabled = true;
$this->watchers[$watcher->id] = $watcher;
return $watcherId;
2013-08-05 22:05:08 +02:00
}
2014-08-06 22:01:24 +02:00
/**
* Schedule a callback to execute once
*
* @param callable $callback Any valid PHP callable
* @param int $msDelay The delay in milliseconds before the callback will trigger (may be zero)
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
2014-06-11 18:21:46 +02:00
public function once(callable $callback, $msDelay) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
$eventResource = event_new();
2014-06-11 18:21:46 +02:00
$msDelay = ($msDelay > 0) ? ($msDelay * $this->resolution) : 0;
2013-08-05 22:05:08 +02:00
$watcher = new LibeventWatcher;
$watcher->id = $watcherId;
$watcher->eventResource = $eventResource;
2014-06-11 18:21:46 +02:00
$watcher->msDelay = $msDelay;
$watcher->callback = $callback;
$watcher->wrapper = $this->wrapOnceCallback($watcher);
$this->watchers[$watcherId] = $watcher;
event_timer_set($eventResource, $watcher->wrapper);
event_base_set($eventResource, $this->base);
2014-06-11 18:21:46 +02:00
event_add($eventResource, $msDelay);
return $watcherId;
}
private function wrapOnceCallback(LibeventWatcher $watcher) {
return function() use ($watcher) {
2013-08-05 22:05:08 +02:00
try {
$callback = $watcher->callback;
$watcherId = $watcher->id;
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId);
if ($result instanceof \Generator) {
resolve($this, $result)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
2013-08-05 22:05:08 +02:00
$this->cancel($watcherId);
} catch (\Exception $e) {
$this->stopException = $e;
$this->stop();
}
};
}
2013-08-05 22:05:08 +02:00
2014-08-06 22:01:24 +02:00
/**
* Schedule a recurring callback to execute every $interval seconds until cancelled
*
* @param callable $callback Any valid PHP callable
* @param int $msDelay The interval in milliseconds between callback invocations
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
2014-06-11 18:21:46 +02:00
public function repeat(callable $callback, $msDelay) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
2014-06-11 18:21:46 +02:00
$msDelay = ($msDelay > 0) ? ($msDelay * $this->resolution) : 0;
$eventResource = event_new();
$watcher = new LibeventWatcher;
$watcher->id = $watcherId;
$watcher->eventResource = $eventResource;
2014-06-11 18:21:46 +02:00
$watcher->msDelay = $msDelay;
$watcher->callback = $callback;
$watcher->wrapper = $this->wrapRepeatingCallback($watcher);
2013-08-05 22:05:08 +02:00
$this->watchers[$watcherId] = $watcher;
event_timer_set($eventResource, $watcher->wrapper);
event_base_set($eventResource, $this->base);
2014-06-11 18:21:46 +02:00
event_add($eventResource, $msDelay);
2013-08-05 22:05:08 +02:00
return $watcherId;
}
private function wrapRepeatingCallback(LibeventWatcher $watcher) {
$callback = $watcher->callback;
$watcherId = $watcher->id;
$eventResource = $watcher->eventResource;
2014-06-11 18:21:46 +02:00
$msDelay = $watcher->msDelay;
2013-08-05 22:05:08 +02:00
2014-06-11 18:21:46 +02:00
return function() use ($callback, $eventResource, $msDelay, $watcherId) {
2013-08-05 22:05:08 +02:00
try {
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId);
if ($result instanceof \Generator) {
resolve($this, $result)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
2014-06-11 18:21:46 +02:00
event_add($eventResource, $msDelay);
2013-08-05 22:05:08 +02:00
} catch (\Exception $e) {
$this->stopException = $e;
$this->stop();
}
};
}
2014-08-06 22:01:24 +02:00
/**
* Watch a stream resource for IO readable data and trigger the callback when actionable
*
* @param resource $stream A stream resource to watch for readable data
* @param callable $callback Any valid PHP callable
* @param bool $enableNow Should the watcher be enabled now or held for later use?
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
public function onReadable($stream, callable $callback, $enableNow = true) {
2013-08-05 22:05:08 +02:00
return $this->watchIoStream($stream, EV_READ | EV_PERSIST, $callback, $enableNow);
}
2014-08-06 22:01:24 +02:00
/**
* Watch a stream resource to become writable and trigger the callback when actionable
*
* @param resource $stream A stream resource to watch for writability
* @param callable $callback Any valid PHP callable
* @param bool $enableNow Should the watcher be enabled now or held for later use?
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
public function onWritable($stream, callable $callback, $enableNow = true) {
2013-08-05 22:05:08 +02:00
return $this->watchIoStream($stream, EV_WRITE | EV_PERSIST, $callback, $enableNow);
}
private function watchIoStream($stream, $flags, callable $callback, $enableNow) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
$eventResource = event_new();
$watcher = new LibeventWatcher;
$watcher->id = $watcherId;
$watcher->stream = $stream;
$watcher->callback = $callback;
$watcher->wrapper = $this->wrapStreamCallback($watcher);
$watcher->isEnabled = (bool) $enableNow;
$watcher->eventResource = $eventResource;
2013-08-05 22:05:08 +02:00
$this->watchers[$watcherId] = $watcher;
event_set($eventResource, $stream, $flags, $watcher->wrapper);
event_base_set($eventResource, $this->base);
if ($enableNow) {
event_add($eventResource);
}
return $watcherId;
}
private function wrapStreamCallback(LibeventWatcher $watcher) {
$callback = $watcher->callback;
$watcherId = $watcher->id;
$stream = $watcher->stream;
return function() use ($callback, $watcherId, $stream) {
2013-08-05 22:05:08 +02:00
try {
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId, $stream);
if ($result instanceof \Generator) {
resolve($this, $result)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
2013-08-05 22:05:08 +02:00
} catch (\Exception $e) {
$this->stopException = $e;
$this->stop();
}
};
}
2014-08-06 22:01:24 +02:00
/**
* React to process control signals
*
* @param int $signo The signal number to watch for (e.g. 2 for Uv::SIGINT)
* @param callable $onSignal
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-06 22:01:24 +02:00
*/
public function onSignal($signo, callable $onSignal) {
$signo = (int) $signo;
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
2014-04-23 02:45:20 +02:00
$eventResource = event_new();
$watcher = new LibeventWatcher;
$watcher->id = $watcherId;
$watcher->signo = $signo;
2014-04-23 02:45:20 +02:00
$watcher->eventResource = $eventResource;
2014-08-06 22:01:24 +02:00
$watcher->callback = $onSignal;
2014-04-23 02:45:20 +02:00
$watcher->wrapper = $this->wrapSignalCallback($watcher);
$this->watchers[$watcherId] = $watcher;
2014-08-06 22:01:24 +02:00
event_set($eventResource, $signo, EV_SIGNAL | EV_PERSIST, $watcher->wrapper);
2014-04-23 02:45:20 +02:00
event_base_set($eventResource, $this->base);
event_add($eventResource);
return $watcherId;
}
private function wrapSignalCallback(LibeventWatcher $watcher) {
$callback = $watcher->callback;
$watcherId = $watcher->id;
$signo = $watcher->signo;
2014-04-23 02:45:20 +02:00
return function() use ($callback, $watcherId, $signo) {
2014-04-23 02:45:20 +02:00
try {
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId, $signo);
if ($result instanceof \Generator) {
resolve($this, $result)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
2014-04-23 02:45:20 +02:00
} catch (\Exception $e) {
$this->stopException = $e;
$this->stop();
}
};
}
2014-08-06 22:01:24 +02:00
/**
* Cancel an existing watcher
*
* @param int $watcherId
* @return void
*/
2014-02-23 22:26:28 +01:00
public function cancel($watcherId) {
if (empty($this->watchers[$watcherId])) {
return;
2013-08-05 22:05:08 +02:00
}
$watcher = $this->watchers[$watcherId];
if (empty($watcher->eventResource)) {
// It's an immediately watcher
unset(
$this->watchers[$watcherId],
$this->immediates[$watcherId]
);
} else {
event_del($watcher->eventResource);
unset($this->watchers[$watcherId]);
}
$this->garbage[] = $watcher;
$this->scheduleGarbageCollection();
2013-08-05 22:05:08 +02:00
}
2014-08-06 22:01:24 +02:00
/**
* Temporarily disable (but don't cancel) an existing timer/stream watcher
*
* @param int $watcherId
* @return void
*/
2014-02-23 22:26:28 +01:00
public function disable($watcherId) {
if (empty($this->watchers[$watcherId])) {
2013-08-05 22:05:08 +02:00
return;
}
$watcher = $this->watchers[$watcherId];
if (empty($watcher->eventResource)) {
// It's an immediately watcher
unset($this->immediates[$watcherId]);
$watcher->isEnabled = false;
} elseif ($watcher->isEnabled) {
event_del($watcher->eventResource);
$watcher->isEnabled = false;
2013-08-05 22:05:08 +02:00
}
}
2014-08-06 22:01:24 +02:00
/**
* Enable a disabled timer/stream watcher
*
* @param int $watcherId
* @return void
*/
2014-02-23 22:26:28 +01:00
public function enable($watcherId) {
if (empty($this->watchers[$watcherId])) {
2013-08-05 22:05:08 +02:00
return;
}
$watcher = $this->watchers[$watcherId];
2013-08-05 22:05:08 +02:00
if ($watcher->isEnabled) {
return;
}
if (empty($watcher->eventResource)) {
// It's an immediately watcher
$this->immediates[$watcherId] = $watcher->callback;
} else {
2014-06-11 18:21:46 +02:00
event_add($watcher->eventResource, $watcher->msDelay);
2013-08-05 22:05:08 +02:00
}
$watcher->isEnabled = true;
2013-08-05 22:05:08 +02:00
}
private function scheduleGarbageCollection() {
if (!$this->isGCScheduled) {
event_add($this->gcEvent, 0);
$this->isGCScheduled = true;
2013-08-05 22:05:08 +02:00
}
}
private function collectGarbage() {
$this->garbage = [];
$this->isGCScheduled = false;
2013-08-05 22:05:08 +02:00
event_del($this->gcEvent);
}
2014-08-21 19:54:02 +02:00
/**
* Access the underlying libevent extension event base
*
* This method exists outside the base Reactor API. It provides access to the underlying
* libevent base for code that wishes to interact with lower-level libevent extension
* functionality.
*
* @return resource
*/
public function getUnderlyingLoop() {
return $this->base;
}
}