rewrite pipeline api

src/Message/Translator/ChainTranslator.php

@@ -8,9 +8,7 @@
 
 use function array_values;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ChainMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ChainMiddleware instead */
 final class ChainTranslator implements Translator
 {
     /** @param iterable<Translator> $translators */

src/Message/Translator/ExcludeEventTranslator.php

@@ -6,9 +6,7 @@
 
 use Patchlevel\EventSourcing\Message\Message;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware instead */
 final class ExcludeEventTranslator implements Translator
 {
     /** @param list<class-string> $classes */

src/Message/Translator/ExcludeEventWithHeaderTranslator.php

@@ -6,9 +6,7 @@
 
 use Patchlevel\EventSourcing\Message\Message;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventWithHeaderMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventWithHeaderMiddleware instead */
 final class ExcludeEventWithHeaderTranslator implements Translator
 {
     /** @param class-string $header */

src/Message/Translator/FilterEventTranslator.php

@@ -6,9 +6,7 @@
 
 use Patchlevel\EventSourcing\Message\Message;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\FilterEventMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\FilterEventMiddleware instead */
 final class FilterEventTranslator implements Translator
 {
     /** @var callable(object $event):bool */

src/Message/Translator/IncludeEventTranslator.php

@@ -6,9 +6,7 @@
 
 use Patchlevel\EventSourcing\Message\Message;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\IncludeEventWithHeaderMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\IncludeEventWithHeaderMiddleware instead */
 final class IncludeEventTranslator implements Translator
 {
     /** @param list<class-string> $classes */

src/Message/Translator/IncludeEventWithHeaderTranslator.php

@@ -6,9 +6,7 @@
 
 use Patchlevel\EventSourcing\Message\Message;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\IncludeEventWithHeaderMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\IncludeEventWithHeaderMiddleware instead */
 final class IncludeEventWithHeaderTranslator implements Translator
 {
     /** @param class-string $header */

src/Message/Translator/RecalculatePlayheadTranslator.php

@@ -11,9 +11,7 @@
 
 use function array_key_exists;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware instead */
 final class RecalculatePlayheadTranslator implements Translator
 {
     /** @var array<string, positive-int> */

src/Message/Translator/ReplaceEventTranslator.php

@@ -7,9 +7,9 @@
 use Patchlevel\EventSourcing\Message\Message;
 
 /**
- * @template T of object
- *
  * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\ReplaceEventMiddleware instead
+ *
+ * @template T of object
  */
 final class ReplaceEventTranslator implements Translator
 {

src/Message/Translator/Translator.php

@@ -6,9 +6,7 @@
 
 use Patchlevel\EventSourcing\Message\Message;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\Middleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\Middleware instead */
 interface Translator
 {
     /** @return list<Message> */

src/Message/Translator/UntilEventTranslator.php

@@ -10,9 +10,7 @@
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\StreamHeader;
 
-/**
- * @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\UntilEventMiddleware instead
- */
+/** @deprecated use Patchlevel\EventSourcing\Pipeline\Middleware\UntilEventMiddleware instead */
 final class UntilEventTranslator implements Translator
 {
     public function __construct(

src/Pipeline/Middleware/AggregateToStreamHeaderMiddleware.php

@@ -8,9 +8,7 @@
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\StreamHeader;
 
-/**
- * @experimental
- */
+/** @experimental */
 final class AggregateToStreamHeaderMiddleware implements Middleware
 {
     /** @return list<Message> */
@@ -29,7 +27,7 @@ public function __invoke(Message $message): array
                     $aggregateHeader->streamName(),
                     $aggregateHeader->playhead,
                     $aggregateHeader->recordedOn,
-                ))
+                )),
         ];
     }
 }

src/Pipeline/Pipeline.php

@@ -4,21 +4,24 @@
 
 namespace Patchlevel\EventSourcing\Pipeline;
 
+use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Pipeline\Middleware\ChainMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Middleware\Middleware;
-use Patchlevel\EventSourcing\Pipeline\Source\Source;
 use Patchlevel\EventSourcing\Pipeline\Target\Target;
 
+use function array_push;
+use function count;
+use function is_array;
+
 final class Pipeline
 {
     private readonly Middleware $middleware;
 
     /** @param list<Middleware>|Middleware $middlewares */
     public function __construct(
-        private readonly Source $source,
         private readonly Target $target,
         array|Middleware $middlewares = [],
-        private readonly int $bufferSize = 1_000,
+        private readonly float|int $bufferSize = INF,
     ) {
         if (is_array($middlewares)) {
             $this->middleware = new ChainMiddleware($middlewares);
@@ -27,34 +30,28 @@ public function __construct(
         }
     }
 
-    public function run(): void
+    /** @param iterable<Message> $messages */
+    public function run(iterable $messages): void
     {
         $buffer = [];
 
-        foreach ($this->source->load() as $message) {
+        foreach ($messages as $message) {
             $result = ($this->middleware)($message);
 
             array_push($buffer, ...$result);
 
-            if (count($buffer) >= $this->bufferSize) {
-                $this->target->save(...$result);
-                $buffer = [];
+            if (count($buffer) < $this->bufferSize) {
+                continue;
             }
-        }
 
-        if (count($buffer) > 0) {
             $this->target->save(...$buffer);
+            $buffer = [];
         }
-    }
 
-    public static function execute(
-        Source $source,
-        Target $target,
-        array|Middleware $middlewares = [],
-        $bufferSize = 1_000,
-    ): void
-    {
-        $pipeline = new self($source, $target, $middlewares, $bufferSize);
-        $pipeline->run();
+        if ($buffer === []) {
+            return;
+        }
+
+        $this->target->save(...$buffer);
     }
-}
+}

src/Pipeline/Target/EventBusTarget.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Pipeline\Target;
+
+use Patchlevel\EventSourcing\EventBus\EventBus;
+use Patchlevel\EventSourcing\Message\Message;
+
+final class EventBusTarget implements Target
+{
+    public function __construct(
+        private readonly EventBus $eventBus,
+    ) {
+    }
+
+    public function save(Message ...$message): void
+    {
+        $this->eventBus->dispatch(...$message);
+    }
+}

src/Pipeline/Target/InMemoryTarget.php

@@ -14,7 +14,7 @@ final class InMemoryTarget implements Target
     public function save(Message ...$message): void
     {
         foreach ($message as $m) {
-        	$this->messages[] = $m;
+            $this->messages[] = $m;
         }
     }
 
@@ -28,4 +28,4 @@ public function clear(): void
     {
         $this->messages = [];
     }
-}
+}

src/Pipeline/Target/StoreTarget.php

@@ -10,12 +10,12 @@
 final class StoreTarget implements Target
 {
     public function __construct(
-        private readonly Store $store
+        private readonly Store $store,
     ) {
     }
 
     public function save(Message ...$message): void
     {
         $this->store->save(...$message);
     }
-}
+}

src/Pipeline/Target/Target.php

@@ -9,4 +9,4 @@
 interface Target
 {
     public function save(Message ...$message): void;
-}
+}

tests/Integration/Subscription/Subscriber/MigrateAggregateToStreamStoreSubscriber.php

@@ -11,7 +11,6 @@
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Pipeline\Middleware\AggregateToStreamHeaderMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Pipeline;
-use Patchlevel\EventSourcing\Pipeline\Source\InMemorySource;
 use Patchlevel\EventSourcing\Pipeline\Target\StoreTarget;
 use Patchlevel\EventSourcing\Schema\ChainDoctrineSchemaConfigurator;
 use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
@@ -20,26 +19,32 @@
 use Patchlevel\EventSourcing\Subscription\RunMode;
 use Patchlevel\EventSourcing\Subscription\Subscriber\BatchableSubscriber;
 
+use function count;
+
 #[Subscriber('migrate', RunMode::Once)]
 final class MigrateAggregateToStreamStoreSubscriber implements BatchableSubscriber
 {
     private const BATCH_SIZE = 10_000;
 
     private readonly SchemaDirector $schemaDirector;
 
-    /**
-     * @var list<Message>
-     */
+    /** @var list<Message> */
     private array $messages = [];
 
+    private readonly Pipeline $pipeline;
+
     public function __construct(
         private readonly StreamDoctrineDbalStore $targetStore,
     ) {
         $this->schemaDirector = new DoctrineSchemaDirector(
             $targetStore->connection(),
-            new ChainDoctrineSchemaConfigurator([
-                $targetStore,
-            ]),
+            new ChainDoctrineSchemaConfigurator([$targetStore]),
+        );
+
+        $this->pipeline = new Pipeline(
+            new StoreTarget($this->targetStore),
+            new AggregateToStreamHeaderMiddleware(),
+            self::BATCH_SIZE * 10, // make sure we have only one batch
         );
     }
 
@@ -56,15 +61,9 @@ public function beginBatch(): void
 
     public function commitBatch(): void
     {
-        $messages = $this->messages;
-        $this->messages = [];
+        $this->pipeline->run($this->messages);
 
-        Pipeline::execute(
-            new InMemorySource($messages),
-            new StoreTarget($this->targetStore),
-            new AggregateToStreamHeaderMiddleware(),
-            self::BATCH_SIZE * 10, // make sure we have only one batch
-        );
+        $this->messages = [];
     }
 
     public function rollbackBatch(): void