check event appeared callback returns a promise

resolves #3
prooph · Aug 1, 2018 · 5ce9963 · 5ce9963
1 parent 3960a7c
commit 5ce9963
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 3 deletions.
diff --git a/src/Internal/AbstractEventStorePersistentSubscription.php b/src/Internal/AbstractEventStorePersistentSubscription.php
@@ -367,7 +367,13 @@ private function processQueue(): Promise
                         }
 
                         try {
-                            yield ($this->eventAppeared)($this, $e->event(), $e->retryCount());
+                            $promise = ($this->eventAppeared)($this, $e->event(), $e->retryCount());
+
+                            if (! $promise instanceof Promise) {
+                                throw new RuntimeException('Event appeared callback needs to return an ' . Promise::class);
+                            }
+
+                            yield $promise;
 
                             if ($this->autoAck) {
                                 $this->subscription->notifyEventsProcessed([$e->originalEvent()->eventId()]);

diff --git a/src/Internal/EventStoreAllCatchUpSubscription.php b/src/Internal/EventStoreAllCatchUpSubscription.php
@@ -17,6 +17,7 @@
 use Prooph\EventStoreClient\AllEventsSlice;
 use Prooph\EventStoreClient\CatchUpSubscriptionSettings;
 use Prooph\EventStoreClient\EventStoreAsyncConnection;
+use Prooph\EventStoreClient\Exception\RuntimeException;
 use Prooph\EventStoreClient\Position;
 use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\SubscriptionDropReason;
@@ -162,7 +163,13 @@ protected function tryProcessAsync(ResolvedEvent $e): Promise
 
             if ($e->originalPosition()->greater($this->lastProcessedPosition)) {
                 try {
-                    yield ($this->eventAppeared)($this, $e);
+                    $promise = ($this->eventAppeared)($this, $e);
+
+                    if (! $promise instanceof Promise) {
+                        throw new RuntimeException('Event appeared callback needs to return an ' . Promise::class);
+                    }
+
+                    yield $promise;
                 } catch (\Throwable $ex) {
                     $this->dropSubscription(SubscriptionDropReason::eventHandlerException(), $ex);
 

diff --git a/src/Internal/EventStoreStreamCatchUpSubscription.php b/src/Internal/EventStoreStreamCatchUpSubscription.php
@@ -17,6 +17,7 @@
 use Generator;
 use Prooph\EventStoreClient\CatchUpSubscriptionSettings;
 use Prooph\EventStoreClient\EventStoreAsyncConnection;
+use Prooph\EventStoreClient\Exception\RuntimeException;
 use Prooph\EventStoreClient\Exception\StreamDeletedException;
 use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\SliceReadStatus;
@@ -173,7 +174,13 @@ protected function tryProcessAsync(ResolvedEvent $e): Promise
 
             if ($e->originalEventNumber() > $this->lastProcessedEventNumber) {
                 try {
-                    yield ($this->eventAppeared)($this, $e);
+                    $promise = ($this->eventAppeared)($this, $e);
+
+                    if (! $promise instanceof Promise) {
+                        throw new RuntimeException('Event appeared callback needs to return an ' . Promise::class);
+                    }
+
+                    yield $promise;
                 } catch (\Throwable $ex) {
                     $this->dropSubscription(SubscriptionDropReason::eventHandlerException(), $ex);