danog/amp
1
0
Fork 0
mirror of https://github.com/danog/amp.git synced 2024-11-30 04:29:08 +01:00
Code Issues Projects Releases Wiki Activity

Minor libevent reactor bugfixes

Browse Source
This commit is contained in:
Daniel Lowrey 2014-12-04 19:30:23 -05:00
parent b11e7a0fd1
commit 49b7a73a8d
2 changed files with 57 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
lib/LibeventReactor.php
View File

@ -90,7 +90,7 @@ class LibeventReactor implements SignalReactor {
}
if (!$this->isRunning) {
// If a watcher stops the reactor break out of the loop
// If one of the immediately watchers stops the reactor break out of the loop
return false;
}
}
@ -201,8 +201,8 @@ class LibeventReactor implements SignalReactor {
$watcher->eventResource = $eventResource;
$watcher->msDelay = $msDelay;
$watcher->callback = $callback;
$watcher->wrapper = $this->wrapOnceCallback($watcher);
$watcher->isEnabled = true;
$this->watchers[$watcherId] = $watcher;
@ -216,7 +216,6 @@ class LibeventReactor implements SignalReactor {
private function wrapOnceCallback(LibeventWatcher $watcher) {
return function() use ($watcher) {
try {
$this->enabledWatcherCount--;
$callback = $watcher->callback;
$watcherId = $watcher->id;
$result = $callback($this, $watcherId);
@ -265,6 +264,7 @@ class LibeventReactor implements SignalReactor {
$watcher->msDelay = $msDelay;
$watcher->callback = $callback;
$watcher->wrapper = $this->wrapRepeatingCallback($watcher);
$watcher->isEnabled = true;
$this->watchers[$watcherId] = $watcher;
@ -287,7 +287,11 @@ class LibeventReactor implements SignalReactor {
if ($result instanceof \Generator) {
$this->resolveGenerator($result)->when($this->onCallbackResolution);
}
event_add($eventResource, $msDelay);
// If the watcher cancelled itself this will no longer be set
if (isset($this->watchers[$watcherId])) {
event_add($eventResource, $msDelay);
}
} catch (\Exception $e) {
$this->handleRunError($e);
}
@ -381,6 +385,7 @@ class LibeventReactor implements SignalReactor {
$watcher->signo = $signo;
$watcher->eventResource = $eventResource;
$watcher->callback = $onSignal;
$watcher->isEnabled = true;
$watcher->wrapper = $this->wrapSignalCallback($watcher);
@ -422,7 +427,7 @@ class LibeventReactor implements SignalReactor {
}
$watcher = $this->watchers[$watcherId];
$this->enabledWatcherCount =- $watcher->isEnabled;
$this->enabledWatcherCount -= (int) $watcher->isEnabled;
if (empty($watcher->eventResource)) {
// It's an immediately watcher
@ -570,6 +575,7 @@ class LibeventReactor implements SignalReactor {
}
return [
'enabled' => $this->enabledWatcherCount,
'timers' => $timers,
'immediates' => $immediates,
'io_readers' => $readers,

46
test/ReactorTest.php
View File

@ -330,4 +330,50 @@ abstract class ReactorTest extends \PHPUnit_Framework_TestCase {
$reactor->run(function() { throw new \Exception('test'); });
$this->assertSame('test', $var);
}
public function testReactorRunsUntilNoWatchersRemain() {
$reactor = $this->getReactor();
$var1 = 0;
$reactor->repeat(function($reactor, $watcherId) use (&$var1) {
if (++$var1 === 3) {
$reactor->cancel($watcherId);
}
}, 0);
$var2 = 0;
$reactor->onWritable(STDOUT, function($reactor, $watcherId) use (&$var2) {
if (++$var2 === 4) {
$reactor->cancel($watcherId);
}
});
$reactor->run();
$this->assertSame(3, $var1);
$this->assertSame(4, $var2);
}
public function testReactorRunsUntilNoWatchersRemainWhenStartedImmediately() {
$reactor = $this->getReactor();
$var1 = 0;
$var2 = 0;
$reactor->run(function($reactor) use (&$var1, &$var2) {
$reactor->repeat(function($reactor, $watcherId) use (&$var1) {
if (++$var1 === 3) {
$reactor->cancel($watcherId);
}
}, 0);
$reactor->onWritable(STDOUT, function($reactor, $watcherId) use (&$var2) {
if (++$var2 === 4) {
$reactor->cancel($watcherId);
}
});
});
$this->assertSame(3, $var1);
$this->assertSame(4, $var2);
}
}