Merge pull request #317 from neos/catchup-speedup

Speed up catchUp() and replay()
neos · Jul 7, 2023 · 53fe558 · 53fe558
2 parents b447cbb + d972abb
commit 53fe558
Showing 1 changed file with 21 additions and 15 deletions.
diff --git a/Classes/EventListener/EventListenerInvoker.php b/Classes/EventListener/EventListenerInvoker.php
@@ -169,6 +169,7 @@ public function catchUp(): void
         $streamName = $this->eventListener instanceof StreamAwareEventListenerInterface ? $this->eventListener::listensToStream() : StreamName::all();
         $eventStream = $this->eventStore->load($streamName, $highestAppliedSequenceNumber + 1);
         $appliedEventsCounter = 0;
+        $sequenceNumber = 0;
 
         foreach ($eventStream as $eventEnvelope) {
             $sequenceNumber = $eventEnvelope->getRawEvent()->getSequenceNumber();
@@ -181,23 +182,29 @@ public function catchUp(): void
                 break;
             }
             try {
-                $this->applyEvent($eventEnvelope);
+                $eventWasApplied = $this->applyEvent($eventEnvelope);
             } catch (EventCouldNotBeAppliedException $exception) {
                 $appliedEventsStorage->releaseHighestAppliedSequenceNumber();
                 throw $exception;
             }
-            $appliedEventsCounter ++;
-            $appliedEventsStorage->saveHighestAppliedSequenceNumber($eventEnvelope->getRawEvent()->getSequenceNumber());
-            if ($this->transactionBatchSize === 1 || $appliedEventsCounter % $this->transactionBatchSize === 0) {
-                $appliedEventsStorage->releaseHighestAppliedSequenceNumber();
-                $highestAppliedSequenceNumber = $appliedEventsStorage->reserveHighestAppliedEventSequenceNumber();
-            } else {
-                $highestAppliedSequenceNumber = $sequenceNumber;
+            if ($eventWasApplied) {
+                $appliedEventsCounter ++;
+                $appliedEventsStorage->saveHighestAppliedSequenceNumber($sequenceNumber);
+                if ($this->transactionBatchSize === 1 || $appliedEventsCounter % $this->transactionBatchSize === 0) {
+                    $appliedEventsStorage->releaseHighestAppliedSequenceNumber();
+                    $highestAppliedSequenceNumber = $appliedEventsStorage->reserveHighestAppliedEventSequenceNumber();
+                } else {
+                    $highestAppliedSequenceNumber = $sequenceNumber;
+                }
             }
             foreach ($this->progressCallbacks as $callback) {
                 $callback($eventEnvelope);
             }
         }
+
+        if ($sequenceNumber > 0) {
+            $appliedEventsStorage->saveHighestAppliedSequenceNumber($sequenceNumber);
+        }
         $appliedEventsStorage->releaseHighestAppliedSequenceNumber();
 
         if ($this->eventListener instanceof AfterCatchUpInterface) {
@@ -207,19 +214,17 @@ public function catchUp(): void
 
     /**
      * @param EventEnvelope $eventEnvelope
+     * @return bool If the event was applied, false if no matching listener method existed
      * @throws EventCouldNotBeAppliedException
      */
-    private function applyEvent(EventEnvelope $eventEnvelope): void
+    private function applyEvent(EventEnvelope $eventEnvelope): bool
     {
         $event = $eventEnvelope->getDomainEvent();
         $rawEvent = $eventEnvelope->getRawEvent();
-        try {
-            $listenerMethodName = 'when' . (new \ReflectionClass($event))->getShortName();
-        } catch (\ReflectionException $exception) {
-            throw new \RuntimeException(sprintf('Could not extract listener method name for listener %s and event %s', \get_class($this->eventListener), \get_class($event)), 1541003718, $exception);
-        }
+        $eventClassName = get_class($event);
+        $listenerMethodName = 'when' . substr($eventClassName, strrpos($eventClassName, '\\') + 1);
         if (!method_exists($this->eventListener, $listenerMethodName)) {
-            return;
+            return false;
         }
         if ($this->eventListener instanceof BeforeInvokeInterface) {
             $this->eventListener->beforeInvoke($eventEnvelope);
@@ -232,6 +237,7 @@ private function applyEvent(EventEnvelope $eventEnvelope): void
         if ($this->eventListener instanceof AfterInvokeInterface) {
             $this->eventListener->afterInvoke($eventEnvelope);
         }
+        return true;
     }
 
     /**