Add better error handling

composer.json

@@ -12,12 +12,14 @@
         "php": "^8.1",
         "psr/event-dispatcher": "^1.0",
         "revolt/event-loop": "^1.0",
-        "amphp/amp": "^3.0"
+        "amphp/amp": "^3.0",
+        "psr/log": "^1.1 || ^2.0 || ^3.0"
     },
     "require-dev": {
         "phpunit/phpunit": "^10.0",
         "phpstan/phpstan": "^1.10",
-        "friendsofphp/php-cs-fixer": "^3.14"
+        "friendsofphp/php-cs-fixer": "^3.14",
+        "colinodell/psr-testlogger": "^1.3"
     },
     "autoload": {
         "psr-4": {

examples/basic-usage.php

@@ -60,3 +60,17 @@ function () use ($event) {
 $event = new UserCreatedEvent('789', '[email protected]');
 $future = $dispatcher->dispatch($event, new TimeoutCancellation(30));
 EventLoop::run();
+
+// Set up logging for your dispatcher - all errors will be logged to PSR logger
+$logger = new ColinODell\PsrTestLogger\TestLogger();
+$dispatcher = new AsyncEventDispatcher(
+    $listenerProvider,
+    $logger
+);
+
+// Let errors bubble up. Useful for unit testing and debugging.
+$dispatcher = new AsyncEventDispatcher(
+    $listenerProvider,
+    $logger,
+    AsyncEventDispatcher::THROW_ON_ERROR
+);

src/AsyncEventDispatcher.php

@@ -5,16 +5,20 @@
 namespace ArchiPro\EventDispatcher;
 
 use function Amp\async;
+use function Amp\Future\await;
 
 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 Psr\Log\LoggerInterface;
+use Throwable;
 
 /**
  * Asynchronous implementation of PSR-14 EventDispatcherInterface using Revolt and AMPHP.
@@ -24,11 +28,15 @@
  */
 class AsyncEventDispatcher implements EventDispatcherInterface
 {
+    const THROW_ON_ERROR = 0b0001;
+
     /**
      * @param ListenerProviderInterface $listenerProvider The provider of event listeners
      */
     public function __construct(
-        private readonly ListenerProviderInterface $listenerProvider
+        private readonly ListenerProviderInterface $listenerProvider,
+        private readonly ?LoggerInterface $logger = null,
+        private readonly int $options = 0b0000
     ) {
     }
 
@@ -85,7 +93,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(Closure::fromCallable([$this, 'errorHandler']));
 
                 $future->await($cancellation);
 
@@ -120,14 +128,34 @@ private function dispatchNonStoppableEvent(
             foreach ($listeners as $listener) {
                 $futures[] = async(function () use ($event, $listener) {
                     $listener($event);
-                });
+                })->catch(Closure::fromCallable([$this, 'errorHandler']));
             }
 
             // Wait for all listeners to complete
-            awaitAll($futures, $cancellation);
+            if ($this->options & self::THROW_ON_ERROR) {
+                // Let the errors bubble up
+                await($futures, $cancellation);
+            } else {
+                // Carry on despite errors
+                awaitAll($futures, $cancellation);
+            }
 
             return $event;
         });
     }
 
+    /**
+     * Handler for errors thrown by listeners.
+     * Based on the settings provided to the constructor, this method can log the errors and/or rethrow them.
+     */
+    protected function errorHandler(Throwable $exception): void
+    {
+        if ($this->logger) {
+            $this->logger->error('Error dispatching event', ['exception' => $exception]);
+        }
+
+        if (($this->options & self::THROW_ON_ERROR) === self::THROW_ON_ERROR) {
+            throw $exception;
+        }
+    }
 }

tests/AsyncEventDispatcherTest.php

@@ -16,7 +16,7 @@
 use Exception;
 use PHPUnit\Framework\TestCase;
 use Revolt\EventLoop;
-
+use ColinODell\PsrTestLogger\TestLogger;
 use Throwable;
 
 /**
@@ -33,19 +33,19 @@ class AsyncEventDispatcherTest extends TestCase
 {
     private ListenerProvider $listenerProvider;
     private AsyncEventDispatcher $dispatcher;
+    private TestLogger $logger;
 
     /**
      * 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->logger = new TestLogger();
+        $this->dispatcher = new AsyncEventDispatcher(
+            $this->listenerProvider,
+            $this->logger
+        );
     }
 
     /**
@@ -80,6 +80,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->logger->records, 'No errors are logged');
     }
 
     /**
@@ -103,6 +105,8 @@ public function testSynchronousStoppableEvent(): void
 
         $this->assertCount(1, $results);
         $this->assertEquals(['listener1'], $results);
+
+        $this->assertCount(0, $this->logger->records, 'No errors are logged');
     }
 
     /**
@@ -114,6 +118,7 @@ public function testNoListenersForEvent(): void
         $dispatchedEvent = $this->dispatcher->dispatch($event);
 
         $this->assertSame($event, $dispatchedEvent->await());
+        $this->assertCount(0, $this->logger->records, 'No errors are logged');
     }
 
     /**
@@ -168,6 +173,12 @@ public function testDispatchesFailureInOneListenerDoesNotAffectOthers(): void
             $futureEvent->calledTwice,
             'The second listener should have been called despite the failure of the first listener'
         );
+
+        $this->assertCount(
+            2,
+            $this->logger->records,
+            'Errors are logged to the logger'
+        );
     }
 
     public function testCancellationOfStoppableEvent(): void
@@ -187,6 +198,8 @@ public function testCancellationOfStoppableEvent(): void
         $this->expectException(CancelledException::class);
 
         $this->dispatcher->dispatch($event, $cancellation)->await();
+
+        $this->assertCount(0, $this->logger->records, 'No errors are logged');
     }
 
     public function testCancellationOfNonStoppableEvent(): void
@@ -206,6 +219,26 @@ public function testCancellationOfNonStoppableEvent(): void
         $this->expectException(CancelledException::class);
 
         $this->dispatcher->dispatch($event, $cancellation)->await();
+
+        $this->assertCount(0, $this->logger->records, 'No errors are logged');
     }
 
+    public function testThrowsErrors(): void
+    {
+        $this->dispatcher = new AsyncEventDispatcher(
+            $this->listenerProvider,
+            $this->logger,
+            AsyncEventDispatcher::THROW_ON_ERROR
+        );
+
+        $event = new class () {};
+        $this->listenerProvider->addListener(get_class($event), function ($event) {
+            throw new Exception('This exception will bubble up because we set the THROW_ON_ERROR option');
+        });
+
+        $this->expectException(Exception::class);
+        $this->expectExceptionMessage('This exception will bubble up because we set the THROW_ON_ERROR option');
+
+        $this->dispatcher->dispatch($event)->await();
+    }
 }