improve batch processing in projectionist

patchlevel · Mar 3, 2024 · 4ebeacc · 4ebeacc
1 parent f898567
commit 4ebeacc
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 20 deletions.
diff --git a/src/Projection/Projectionist/DefaultProjectionist.php b/src/Projection/Projectionist/DefaultProjectionist.php
@@ -84,14 +84,13 @@ function ($projections) use ($limit): void {
                 );
 
                 $stream = null;
+                $messageCounter = 0;
 
                 try {
                     $stream = $this->messageStore->load(
                         new Criteria(fromIndex: $startIndex),
                     );
 
-                    $messageCounter = 0;
-
                     foreach ($stream as $message) {
                         $index = $stream->index();
 
@@ -141,6 +140,16 @@ function ($projections) use ($limit): void {
                         }
                     }
                 } finally {
+                    if ($messageCounter > 0) {
+                        foreach ($projections as $projection) {
+                            if (!$projection->isBooting()) {
+                                continue;
+                            }
+
+                            $this->projectionStore->update($projection);
+                        }
+                    }
+
                     $stream?->close();
                 }
 
@@ -212,13 +221,12 @@ function (array $projections) use ($limit): void {
                 );
 
                 $stream = null;
+                $messageCounter = 0;
 
                 try {
                     $criteria = new Criteria(fromIndex: $startIndex);
                     $stream = $this->messageStore->load($criteria);
 
-                    $messageCounter = 0;
-
                     foreach ($stream as $message) {
                         $index = $stream->index();
 
@@ -263,6 +271,16 @@ function (array $projections) use ($limit): void {
                         }
                     }
                 } finally {
+                    if ($messageCounter > 0) {
+                        foreach ($projections as $projection) {
+                            if (!$projection->isActive()) {
+                                continue;
+                            }
+
+                            $this->projectionStore->update($projection);
+                        }
+                    }
+
                     $stream?->close();
                 }
 
@@ -547,7 +565,6 @@ private function handleMessage(int $index, Message $message, Projection $project
 
         if ($subscribeMethods === []) {
             $projection->changePosition($index);
-            $this->projectionStore->update($projection);
 
             $this->logger?->debug(
                 sprintf(
@@ -583,7 +600,6 @@ private function handleMessage(int $index, Message $message, Projection $project
 
         $projection->changePosition($index);
         $projection->resetRetry();
-        $this->projectionStore->update($projection);
 
         $this->logger?->debug(
             sprintf(

diff --git a/tests/Unit/Projection/Projectionist/DefaultProjectionistTest.php b/tests/Unit/Projection/Projectionist/DefaultProjectionistTest.php
@@ -351,6 +351,13 @@ public function handle(Message $message): void
                 ProjectionStatus::Booting,
                 1,
             ),
+            new Projection(
+                $projectionId2,
+                Projection::DEFAULT_GROUP,
+                RunMode::FromBeginning,
+                ProjectionStatus::Booting,
+                1,
+            ),
             new Projection(
                 $projectionId1,
                 Projection::DEFAULT_GROUP,
@@ -461,13 +468,6 @@ public function handle(Message $message): void
         $projectionist->boot();
 
         self::assertEquals([
-            new Projection(
-                $projectionId,
-                Projection::DEFAULT_GROUP,
-                RunMode::FromBeginning,
-                ProjectionStatus::Booting,
-                1,
-            ),
             new Projection(
                 $projectionId,
                 Projection::DEFAULT_GROUP,
@@ -834,6 +834,13 @@ public function handle(Message $message): void
                 ProjectionStatus::Active,
                 1,
             ),
+            new Projection(
+                $projectionId2,
+                Projection::DEFAULT_GROUP,
+                RunMode::FromBeginning,
+                ProjectionStatus::Active,
+                1,
+            ),
         ], $projectionStore->updatedProjections);
 
         self::assertSame($message, $projector1->message);
@@ -999,13 +1006,6 @@ public function handle(Message $message): void
         $projectionist->run();
 
         self::assertEquals([
-            new Projection(
-                $projectionId,
-                Projection::DEFAULT_GROUP,
-                RunMode::FromBeginning,
-                ProjectionStatus::Active,
-                1,
-            ),
             new Projection(
                 $projectionId,
                 Projection::DEFAULT_GROUP,