diff --git a/examples/basic-usage.php b/examples/basic-usage.php index 31de479..a03453f 100644 --- a/examples/basic-usage.php +++ b/examples/basic-usage.php @@ -60,3 +60,11 @@ function () use ($event) { $event = new UserCreatedEvent('789', 'user@example.com'); $future = $dispatcher->dispatch($event, new TimeoutCancellation(30)); EventLoop::run(); + +// Set up logging for your dispatcher - all errors will be logged to PSR logger +$dispatcher = new AsyncEventDispatcher( + $listenerProvider, + function (Throwable $exception) { + error_log($exception->getMessage()); + } +); diff --git a/src/AsyncEventDispatcher.php b/src/AsyncEventDispatcher.php index a4d1d54..47104a2 100644 --- a/src/AsyncEventDispatcher.php +++ b/src/AsyncEventDispatcher.php @@ -7,14 +7,17 @@ use function Amp\async; use Amp\Cancellation; + use Amp\Future; use function Amp\Future\awaitAll; use Amp\NullCancellation; +use Closure; use Psr\EventDispatcher\EventDispatcherInterface; use Psr\EventDispatcher\ListenerProviderInterface; use Psr\EventDispatcher\StoppableEventInterface; +use Throwable; /** * Asynchronous implementation of PSR-14 EventDispatcherInterface using Revolt and AMPHP. @@ -24,12 +27,22 @@ */ class AsyncEventDispatcher implements EventDispatcherInterface { + /** @var Closure(Throwable): (void) */ + private Closure $errorHandler; + /** * @param ListenerProviderInterface $listenerProvider The provider of event listeners + * @param callable(Throwable): (void) $errorHandler The handler for errors thrown by listeners */ public function __construct( - private readonly ListenerProviderInterface $listenerProvider + private readonly ListenerProviderInterface $listenerProvider, + ?callable $errorHandler = null ) { + if ($errorHandler === null) { + $this->errorHandler = function (Throwable $exception): void {}; + } else { + $this->errorHandler = Closure::fromCallable($errorHandler); + } } /** @@ -85,7 +98,7 @@ private function dispatchStoppableEvent( // that doesn't mean we want to block other listeners outside this loop. $future = async(function () use ($event, $listener) { $listener($event); - }); + })->catch($this->errorHandler); $future->await($cancellation); @@ -120,14 +133,13 @@ private function dispatchNonStoppableEvent( foreach ($listeners as $listener) { $futures[] = async(function () use ($event, $listener) { $listener($event); - }); + })->catch($this->errorHandler); } - // Wait for all listeners to complete + // Wait for all listeners to complete. This will carry on despite errors. awaitAll($futures, $cancellation); return $event; }); } - } diff --git a/tests/AsyncEventDispatcherTest.php b/tests/AsyncEventDispatcherTest.php index 6e4b7a8..19ad611 100644 --- a/tests/AsyncEventDispatcherTest.php +++ b/tests/AsyncEventDispatcherTest.php @@ -15,8 +15,6 @@ use ArchiPro\EventDispatcher\Tests\Fixture\TestEvent; use Exception; use PHPUnit\Framework\TestCase; -use Revolt\EventLoop; - use Throwable; /** @@ -34,18 +32,21 @@ class AsyncEventDispatcherTest extends TestCase private ListenerProvider $listenerProvider; private AsyncEventDispatcher $dispatcher; + /** @var array */ + private array $errors = []; + /** * Sets up the test environment before each test. */ protected function setUp(): void { $this->listenerProvider = new ListenerProvider(); - $this->dispatcher = new AsyncEventDispatcher($this->listenerProvider); - - EventLoop::setErrorHandler(function (Throwable $err) { - throw $err; - }); - + $this->dispatcher = new AsyncEventDispatcher( + $this->listenerProvider, + function (Throwable $exception) { + $this->errors[] = $exception; + } + ); } /** @@ -80,6 +81,8 @@ public function testDispatchEventToMultipleListeners(): void $this->assertCount(2, $results); $this->assertContains('listener1: test data', $results); $this->assertContains('listener2: test data', $results); + + $this->assertCount(0, $this->errors, 'No errors are logged'); } /** @@ -103,6 +106,8 @@ public function testSynchronousStoppableEvent(): void $this->assertCount(1, $results); $this->assertEquals(['listener1'], $results); + + $this->assertCount(0, $this->errors, 'No errors are logged'); } /** @@ -114,6 +119,7 @@ public function testNoListenersForEvent(): void $dispatchedEvent = $this->dispatcher->dispatch($event); $this->assertSame($event, $dispatchedEvent->await()); + $this->assertCount(0, $this->errors, 'No errors are logged'); } /** @@ -158,16 +164,18 @@ public function testDispatchesFailureInOneListenerDoesNotAffectOthers(): void $futureEvent = $this->dispatcher->dispatch($event); - $futureEvent = $futureEvent->await(); + $futureEvent->await(); $this->assertTrue( - $futureEvent->calledOnce, + $event->calledOnce, 'The first listener should have been called' ); $this->assertTrue( - $futureEvent->calledTwice, + $event->calledTwice, 'The second listener should have been called despite the failure of the first listener' ); + + $this->assertCount(2, $this->errors, 'Errors are caught for both listeners'); } public function testCancellationOfStoppableEvent(): void @@ -187,6 +195,8 @@ public function testCancellationOfStoppableEvent(): void $this->expectException(CancelledException::class); $this->dispatcher->dispatch($event, $cancellation)->await(); + + $this->assertCount(0, $this->errors, 'No errors are caught'); } public function testCancellationOfNonStoppableEvent(): void @@ -206,6 +216,7 @@ public function testCancellationOfNonStoppableEvent(): void $this->expectException(CancelledException::class); $this->dispatcher->dispatch($event, $cancellation)->await(); - } + $this->assertCount(0, $this->errors, 'No errors are caught'); + } }