refactor stream

phpstan-baseline.neon

@@ -54,6 +54,12 @@ parameters:
 			count: 1
 			path: src/Message/Serializer/DefaultHeadersSerializer.php
 
+		-
+			message: '#^Property Patchlevel\\EventSourcing\\Message\\Stream\:\:\$iterator \(Iterator\<int, Patchlevel\\EventSourcing\\Message\\Message\>\|null\) does not accept IteratorIterator\<mixed, mixed, Traversable\<TKey, TValue\>\>\.$#'
+			identifier: assign.propertyType
+			count: 1
+			path: src/Message/Stream.php
+
 		-
 			message: '#^Property Patchlevel\\EventSourcing\\Serializer\\Normalizer\\IdNormalizer\:\:\$identifierClass \(class\-string\<Patchlevel\\EventSourcing\\Identifier\\Identifier\>\|null\) does not accept string\.$#'
 			identifier: assign.propertyType
@@ -66,24 +72,6 @@ parameters:
 			count: 1
 			path: src/Snapshot/DefaultSnapshotStore.php
 
-		-
-			message: '#^Method Patchlevel\\EventSourcing\\Store\\ArrayStream\:\:current\(\) never returns null so it can be removed from the return type\.$#'
-			identifier: return.unusedType
-			count: 1
-			path: src/Store/ArrayStream.php
-
-		-
-			message: '#^Property Patchlevel\\EventSourcing\\Store\\ArrayStream\:\:\$index \(int\<1, max\>\|null\) does not accept int\<0, max\>\.$#'
-			identifier: assign.propertyType
-			count: 1
-			path: src/Store/ArrayStream.php
-
-		-
-			message: '#^Ternary operator condition is always true\.$#'
-			identifier: ternary.alwaysTrue
-			count: 1
-			path: src/Store/ArrayStream.php
-
 		-
 			message: '#^Method Patchlevel\\EventSourcing\\Store\\Criteria\\Criteria\:\:get\(\) should return T of object but returns object\.$#'
 			identifier: return.type
@@ -96,59 +84,23 @@ parameters:
 			count: 1
 			path: src/Store/Criteria/StreamCriterion.php
 
-		-
-			message: '#^Method Patchlevel\\EventSourcing\\Store\\StreamDoctrineDbalStoreStream\:\:current\(\) never returns null so it can be removed from the return type\.$#'
-			identifier: return.unusedType
-			count: 1
-			path: src/Store/StreamDoctrineDbalStoreStream.php
-
 		-
 			message: '#^Parameter \#1 \$playhead of class Patchlevel\\EventSourcing\\Store\\Header\\PlayheadHeader constructor expects int\<1, max\>, int given\.$#'
 			identifier: argument.type
 			count: 1
-			path: src/Store/StreamDoctrineDbalStoreStream.php
-
-		-
-			message: '#^Ternary operator condition is always true\.$#'
-			identifier: ternary.alwaysTrue
-			count: 1
-			path: src/Store/StreamDoctrineDbalStoreStream.php
-
-		-
-			message: '#^Method Patchlevel\\EventSourcing\\Store\\TaggableDoctrineDbalStoreStream\:\:current\(\) never returns null so it can be removed from the return type\.$#'
-			identifier: return.unusedType
-			count: 1
-			path: src/Store/TaggableDoctrineDbalStoreStream.php
+			path: src/Store/StreamDoctrineDbalStore.php
 
 		-
 			message: '#^Parameter \#1 \$playhead of class Patchlevel\\EventSourcing\\Store\\Header\\PlayheadHeader constructor expects int\<1, max\>, int given\.$#'
 			identifier: argument.type
 			count: 1
-			path: src/Store/TaggableDoctrineDbalStoreStream.php
+			path: src/Store/TaggableDoctrineDbalStore.php
 
 		-
 			message: '#^Parameter \#1 \$tags of class Patchlevel\\EventSourcing\\Store\\Header\\TagsHeader constructor expects list\<string\>, mixed given\.$#'
 			identifier: argument.type
 			count: 1
-			path: src/Store/TaggableDoctrineDbalStoreStream.php
-
-		-
-			message: '#^Ternary operator condition is always true\.$#'
-			identifier: ternary.alwaysTrue
-			count: 1
-			path: src/Store/TaggableDoctrineDbalStoreStream.php
-
-		-
-			message: '#^Generator expects key type int, int\<1, max\>\|null given\.$#'
-			identifier: generator.keyType
-			count: 1
-			path: src/Subscription/Engine/GapResolverStoreMessageLoader.php
-
-		-
-			message: '#^Property Patchlevel\\EventSourcing\\Subscription\\Engine\\GeneratorStream\:\:\$index \(int\<1, max\>\|null\) does not accept int\.$#'
-			identifier: assign.propertyType
-			count: 1
-			path: src/Subscription/Engine/GeneratorStream.php
+			path: src/Store/TaggableDoctrineDbalStore.php
 
 		-
 			message: '#^Parameter \#1 \$eventClass of method Patchlevel\\EventSourcing\\Metadata\\Event\\EventRegistry\:\:eventName\(\) expects class\-string, string given\.$#'
@@ -294,12 +246,6 @@ parameters:
 			count: 1
 			path: tests/Integration/Store/Profile.php
 
-		-
-			message: '#^Using nullsafe method call on non\-nullable type Patchlevel\\EventSourcing\\Store\\Stream\. Use \-\> instead\.$#'
-			identifier: nullsafe.neverNull
-			count: 1
-			path: tests/Integration/Store/StreamDoctrineDbalStoreTest.php
-
 		-
 			message: '#^Parameter \#1 \$table of class Patchlevel\\EventSourcing\\Subscription\\Cleanup\\Dbal\\DropTableTask constructor expects non\-empty\-string, string given\.$#'
 			identifier: argument.type

src/Console/Command/StoreMigrateCommand.php

@@ -6,7 +6,6 @@
 
 use Patchlevel\EventSourcing\Console\InputHelper;
 use Patchlevel\EventSourcing\Console\OutputStyle;
-use Patchlevel\EventSourcing\Message\Pipe;
 use Patchlevel\EventSourcing\Message\Translator\Translator;
 use Patchlevel\EventSourcing\Store\Store;
 use Symfony\Component\Console\Attribute\AsCommand;
@@ -46,38 +45,23 @@
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        $buffer = InputHelper::positiveIntOrZero($input->getOption('buffer'));
+        $buffer = InputHelper::positiveInt($input->getOption('buffer'));
         $style = new OutputStyle($input, $output);
 
         $style->info('Migration initialization...');
 
         $count = $this->store->count();
-        $messages = $this->store->load();
+        $stream = $this->store->load();
 
         $style->progressStart($count);
 
-        $bufferedMessages = [];
+        $translatedStream = $stream->transform(...$this->translators);
 
-        $pipe = new Pipe(
-            $messages,
-            ...$this->translators,
-        );
+        foreach ($translatedStream->chunk($buffer) as $chunk) {
+            $messages = $chunk->toList();
 
-        foreach ($pipe as $message) {
-            $bufferedMessages[] = $message;
-
-            if (count($bufferedMessages) < $buffer) {
-                continue;
-            }
-
-            $this->newStore->save(...$bufferedMessages);
-            $bufferedMessages = [];
-            $style->progressAdvance($buffer);
-        }
-
-        if (count($bufferedMessages) !== 0) {
-            $this->newStore->save(...$bufferedMessages);
-            $style->progressAdvance(count($bufferedMessages));
+            $this->newStore->save(...$messages);
+            $style->progressAdvance(count($messages));
         }
 
         $style->progressFinish();

src/Console/InputHelper.php

@@ -66,6 +66,26 @@ public static function nullableInt(mixed $value): int|null
         return (int)$value;
     }
 
+    /** @return positive-int */
+    public static function positiveInt(mixed $value): int
+    {
+        if (!is_string($value) && !is_int($value)) {
+            throw new InvalidArgumentGiven($value, 'positive-int');
+        }
+
+        if (!is_numeric($value)) {
+            throw new InvalidArgumentGiven($value, 'positive-int');
+        }
+
+        $value = (int)$value;
+
+        if ($value <= 0) {
+            throw new InvalidArgumentGiven($value, 'positive-int');
+        }
+
+        return $value;
+    }
+
     /** @return positive-int|null */
     public static function nullablePositiveInt(mixed $value): int|null
     {