Merge pull request #466 from patchlevel/use-index-for-position

use index for position in projectionist
patchlevel · Jan 15, 2024 · b0e4be2 · b0e4be2
2 parents e1da663 + 4fb9b2d
commit b0e4be2
Show file tree

Hide file tree

Showing 8 changed files with 250 additions and 33 deletions.
diff --git a/baseline.xml b/baseline.xml
@@ -94,6 +94,11 @@
       <code><![CDATA[$data['payload']]]></code>
     </MixedArgument>
   </file>
+  <file src="src/Store/ArrayStream.php">
+    <InvalidPropertyAssignmentValue>
+      <code>$index</code>
+    </InvalidPropertyAssignmentValue>
+  </file>
   <file src="src/Store/DoctrineHelper.php">
     <MixedReturnTypeCoercion>
       <code>$normalizedCustomHeaders</code>

diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon
@@ -20,6 +20,11 @@ parameters:
 			count: 1
 			path: src/Snapshot/DefaultSnapshotStore.php
 
+		-
+			message: "#^Property Patchlevel\\\\EventSourcing\\\\Store\\\\ArrayStream\\:\\:\\$index \\(int\\<1, max\\>\\|null\\) does not accept int\\<0, max\\>\\.$#"
+			count: 1
+			path: src/Store/ArrayStream.php
+
 		-
 			message: "#^Ternary operator condition is always true\\.$#"
 			count: 1

diff --git a/src/Projection/Projection/Projection.php b/src/Projection/Projection/Projection.php
@@ -35,9 +35,9 @@ public function projectionError(): ProjectionError|null
         return $this->error;
     }
 
-    public function incrementPosition(): void
+    public function changePosition(int $position): void
     {
-        $this->position++;
+        $this->position = $position;
     }
 
     public function isNew(): bool

diff --git a/src/Projection/Projectionist/DefaultProjectionist.php b/src/Projection/Projectionist/DefaultProjectionist.php
@@ -116,35 +116,53 @@ public function boot(
             return;
         }
 
-        $currentPosition = $projections->getLowestProjectionPosition();
+        $startIndex = $projections->getLowestProjectionPosition();
 
         $this->logger?->debug(
             sprintf(
                 'Projectionist: Event stream is processed for booting from position %s.',
-                $currentPosition,
+                $startIndex,
             ),
         );
 
         $stream = null;
 
         try {
-            $criteria = new Criteria(fromIndex: $currentPosition);
+            $criteria = new Criteria(fromIndex: $startIndex);
             $stream = $this->streamableMessageStore->load($criteria);
 
             $messageCounter = 0;
 
             foreach ($stream as $message) {
+                $index = $stream->index();
+
+                if ($index === null) {
+                    throw new UnexpectedError('Stream index is null, this should not happen.');
+                }
+
                 foreach ($projections->filterByProjectionStatus(ProjectionStatus::Booting) as $projection) {
-                    $this->handleMessage($message, $projection, $throwByError);
+                    if ($projection->position() >= $index) {
+                        $this->logger?->debug(
+                            sprintf(
+                                'Projectionist: Projection "%s" is farther than the current position (%d > %d), continue booting.',
+                                $projection->id()->toString(),
+                                $projection->position(),
+                                $index,
+                            ),
+                        );
+
+                        continue;
+                    }
+
+                    $this->handleMessage($index, $message, $projection, $throwByError);
                 }
 
-                $currentPosition++;
                 $messageCounter++;
 
                 $this->logger?->debug(
                     sprintf(
                         'Projectionist: Current event stream position for booting: %s',
-                        $currentPosition,
+                        $index,
                     ),
                 );
 
@@ -198,45 +216,50 @@ public function run(
             return;
         }
 
-        $currentPosition = $projections->getLowestProjectionPosition();
+        $startIndex = $projections->getLowestProjectionPosition();
 
         $this->logger?->debug(
             sprintf(
                 'Projectionist: Event stream is processed from position %d.',
-                $currentPosition,
+                $startIndex,
             ),
         );
 
         $stream = null;
 
         try {
-            $criteria = new Criteria(fromIndex: $currentPosition);
+            $criteria = new Criteria(fromIndex: $startIndex);
             $stream = $this->streamableMessageStore->load($criteria);
 
             $messageCounter = 0;
 
             foreach ($stream as $message) {
+                $index = $stream->index();
+
+                if ($index === null) {
+                    throw new UnexpectedError('Stream index is null, this should not happen.');
+                }
+
                 foreach ($projections->filterByProjectionStatus(ProjectionStatus::Active) as $projection) {
-                    if ($projection->position() > $currentPosition) {
+                    if ($projection->position() >= $index) {
                         $this->logger?->debug(
                             sprintf(
                                 'Projectionist: Projection "%s" is farther than the current position (%d > %d), continue processing.',
                                 $projection->id()->toString(),
                                 $projection->position(),
-                                $currentPosition,
+                                $index,
                             ),
                         );
 
                         continue;
                     }
 
-                    $this->handleMessage($message, $projection, $throwByError);
+                    $this->handleMessage($index, $message, $projection, $throwByError);
                 }
 
-                $currentPosition++;
                 $messageCounter++;
 
-                $this->logger?->debug(sprintf('Projectionist: Current event stream position: %s', $currentPosition));
+                $this->logger?->debug(sprintf('Projectionist: Current event stream position: %s', $index));
 
                 if ($limit !== null && $messageCounter >= $limit) {
                     $this->logger?->info(
@@ -256,7 +279,7 @@ public function run(
         $this->logger?->info(
             sprintf(
                 'Projectionist: End of stream on position "%d" has been reached, finish processing.',
-                $currentPosition,
+                $stream->index() ?: 'unknown',
             ),
         );
     }
@@ -430,7 +453,7 @@ public function projections(): ProjectionCollection
         return $projections;
     }
 
-    private function handleMessage(Message $message, Projection $projection, bool $throwByError): void
+    private function handleMessage(int $index, Message $message, Projection $projection, bool $throwByError): void
     {
         $projector = $this->projector($projection->id());
 
@@ -441,7 +464,7 @@ private function handleMessage(Message $message, Projection $projection, bool $t
         $subscribeMethod = $this->projectorResolver->resolveSubscribeMethod($projector, $message);
 
         if (!$subscribeMethod) {
-            $projection->incrementPosition();
+            $projection->changePosition($index);
             $this->projectionStore->save($projection);
 
             $this->logger?->debug(
@@ -484,7 +507,7 @@ private function handleMessage(Message $message, Projection $projection, bool $t
             return;
         }
 
-        $projection->incrementPosition();
+        $projection->changePosition($index);
         $projection->resetRetry();
         $this->projectionStore->save($projection);
 

diff --git a/src/Projection/Projectionist/UnexpectedError.php b/src/Projection/Projectionist/UnexpectedError.php
@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Projection\Projectionist;
+
+use RuntimeException;
+
+final class UnexpectedError extends RuntimeException
+{
+}
diff --git a/src/Store/ArrayStream.php b/src/Store/ArrayStream.php
@@ -20,11 +20,15 @@ final class ArrayStream implements Stream, IteratorAggregate
     /** @var positive-int|0|null */
     private int|null $position;
 
-    /** @param list<Message> $messages The index is based on position. An offset is not supported. */
+    /** @var positive-int|null */
+    private int|null $index;
+
+    /** @param array<positive-int|0, Message> $messages The index is the key. An offset is not supported. */
     public function __construct(array $messages = [])
     {
         $this->iterator = $messages === [] ? new ArrayIterator() : $this->createGenerator($messages);
         $this->position = null;
+        $this->index = null;
     }
 
     public function close(): void
@@ -54,13 +58,11 @@ public function position(): int|null
      */
     public function index(): int|null
     {
-        $position = $this->position();
-
-        if ($position === null) {
-            return null;
+        if (!$this->index) {
+            $this->iterator->key();
         }
 
-        return $position + 1;
+        return $this->index;
     }
 
     public function current(): Message|null
@@ -69,19 +71,31 @@ public function current(): Message|null
     }
 
     /**
-     * @param list<Message> $messages
+     * @param array<positive-int|0, Message> $messages
      *
      * @return Generator<Message>
      */
     private function createGenerator(array $messages): Generator
     {
-        foreach ($messages as $message) {
+        $hasIndex = true;
+
+        foreach ($messages as $index => $message) {
             if ($this->position === null) {
                 $this->position = 0;
             } else {
                 $this->position++;
             }
 
+            if ($index === 0) {
+                $hasIndex = false;
+            }
+
+            if ($hasIndex) {
+                $this->index = $index;
+            } else {
+                $this->index = $this->position + 1;
+            }
+
             yield $message;
         }
     }

diff --git a/tests/Unit/Projection/Projection/ProjectionTest.php b/tests/Unit/Projection/Projection/ProjectionTest.php
@@ -103,14 +103,14 @@ public function testOutdated(): void
         self::assertTrue($projection->isOutdated());
     }
 
-    public function testIncrementPosition(): void
+    public function testChangePosition(): void
     {
         $projection = new Projection(
             new ProjectionId('test', 1),
         );
 
-        $projection->incrementPosition();
+        $projection->changePosition(10);
 
-        self::assertEquals(1, $projection->position());
+        self::assertEquals(10, $projection->position());
     }
 }