Merge c20162a into 9ab667c

prooph · Oct 7, 2016 · 935c74b · 935c74b
2 parents 9ab667c + c20162a
commit 935c74b
Show file tree

Hide file tree

Showing 2 changed files with 54 additions and 1 deletion.
diff --git a/src/DoctrineStreamIterator.php b/src/DoctrineStreamIterator.php
@@ -24,6 +24,8 @@
  */
 final class DoctrineStreamIterator implements Iterator
 {
+    const BATCH_SIZE = 100;
+
     /**
      * @var QueryBuilder
      */
@@ -64,6 +66,11 @@ final class DoctrineStreamIterator implements Iterator
      */
     private $currentKey;
 
+    /**
+     * @var int
+     */
+    private $batchPosition = 0;
+
     /**
      * @param QueryBuilder $queryBuilder
      * @param MessageFactory $messageFactory
@@ -129,7 +136,18 @@ public function next()
         if (false !== $this->currentItem) {
             $this->currentKey++;
         } else {
-            $this->currentKey = -1;
+            $this->batchPosition++;
+            $this->queryBuilder->setFirstResult(self::BATCH_SIZE * $this->batchPosition);
+            $this->queryBuilder->setMaxResults(self::BATCH_SIZE);
+            /* @var $stmt \Doctrine\DBAL\Statement */
+            $this->statement = $this->queryBuilder->execute();
+            $this->statement->setFetchMode(\PDO::FETCH_ASSOC);
+
+            $this->currentItem = $this->statement->fetch();
+
+            if (false === $this->currentItem) {
+                $this->currentKey = -1;
+            }
         }
     }
 
@@ -160,6 +178,10 @@ public function rewind()
     {
         //Only perform rewind if current item is not the first element
         if ($this->currentKey !== 0) {
+            $this->batchPosition = 0;
+            $this->queryBuilder->setFirstResult(0);
+            $this->queryBuilder->setMaxResults(self::BATCH_SIZE);
+
             /* @var $stmt \Doctrine\DBAL\Statement */
             $stmt = $this->queryBuilder->execute();
             $stmt->setFetchMode(\PDO::FETCH_ASSOC);

diff --git a/tests/DoctrineEventStoreAdapterTest.php b/tests/DoctrineEventStoreAdapterTest.php
@@ -433,6 +433,37 @@ public function it_fails_to_write_with_duplicate_aggregate_id_and_version()
         $this->adapter->appendTo(new StreamName('Prooph\Model\User'), new \ArrayIterator([$streamEvent]));
     }
 
+    /**
+     * @test
+     */
+    public function it_replays_larger_streams_in_chunks()
+    {
+        $streamName = new StreamName('Prooph\Model\User');
+
+        $streamEvents = [];
+
+        for ($i = 1; $i <= 150; $i++) {
+            $streamEvents[] = UserCreated::with(
+                ['name' => 'Max Mustermann ' . $i, 'email' => 'contact_' . $i . '@prooph.de'],
+                $i
+            );
+        }
+
+        $this->adapter->create(new Stream($streamName, new \ArrayIterator($streamEvents)));
+
+        $replay = $this->adapter->replay($streamName);
+
+        $count = 0;
+        foreach ($replay as $event) {
+            $count++;
+            $this->assertEquals('Max Mustermann ' . $count, $event->payload()['name']);
+            $this->assertEquals('contact_' . $count . '@prooph.de', $event->payload()['email']);
+            $this->assertEquals($count, $event->version());
+        }
+
+        $this->assertEquals(150, $count);
+    }
+
     /**
      * @return Stream
      */