Merge pull request #540 from patchlevel/replace-pipeline

replace pipeline with subscription engine

deptrac.yaml

@@ -64,10 +64,6 @@ deptrac:
               value: MetadataMessage
             - type: layer
               value: MetadataSubscriber
-    - name: Pipeline
-      collectors:
-        - type: directory
-          value: src/Pipeline/.*
     - name: Repository
       collectors:
         - type: directory
@@ -117,8 +113,10 @@ deptrac:
       - Attribute
       - Message
     Message:
+      - Aggregate
       - MetadataMessage
       - Serializer
+      - Store
     Metadata:
     MetadataAggregate:
       - Aggregate
@@ -137,12 +135,6 @@ deptrac:
       - Attribute
       - Metadata
       - Subscription
-    Pipeline:
-      - Aggregate
-      - EventBus
-      - Message
-      - Store
-      - Subscription
     Subscription:
       - Attribute
       - Clock

docs/pages/pipeline.md

@@ -10,9 +10,9 @@ whether the migration worked.
 In this example the event `PrivacyAdded` is removed and the event `OldVisited` is replaced by `NewVisited`:
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
-use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware;
-use Patchlevel\EventSourcing\Pipeline\Middleware\ReplaceEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\ExcludeEventTranslator;
+use Patchlevel\EventSourcing\Message\Translator\RecalculatePlayheadTranslator;
+use Patchlevel\EventSourcing\Message\Translator\ReplaceEventTranslator;
 use Patchlevel\EventSourcing\Pipeline\Pipeline;
 use Patchlevel\EventSourcing\Pipeline\Source\StoreSource;
 use Patchlevel\EventSourcing\Pipeline\Target\StoreTarget;
@@ -21,11 +21,11 @@ $pipeline = new Pipeline(
     new StoreSource($oldStore),
     new StoreTarget($newStore),
     [
-        new ExcludeEventMiddleware([PrivacyAdded::class]),
-        new ReplaceEventMiddleware(OldVisited::class, static function (OldVisited $oldVisited) {
+        new ExcludeEventTranslator([PrivacyAdded::class]),
+        new ReplaceEventTranslator(OldVisited::class, static function (OldVisited $oldVisited) {
             return new NewVisited($oldVisited->profileId());
         }),
-        new RecalculatePlayheadMiddleware(),
+        new RecalculatePlayheadTranslator(),
     ],
 );
 ```
@@ -197,9 +197,9 @@ Middelwares can be used to manipulate, delete or expand messages or events durin
 With this middleware you can exclude certain events.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\ExcludeEventTranslator;
 
-$middleware = new ExcludeEventMiddleware([EmailChanged::class]);
+$middleware = new ExcludeEventTranslator([EmailChanged::class]);
 ```
 !!! warning
 
@@ -210,9 +210,9 @@ $middleware = new ExcludeEventMiddleware([EmailChanged::class]);
 With this middleware you can only allow certain events.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\IncludeEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\IncludeEventTranslator;
 
-$middleware = new IncludeEventMiddleware([ProfileCreated::class]);
+$middleware = new IncludeEventTranslator([ProfileCreated::class]);
 ```
 !!! warning
 
@@ -226,9 +226,9 @@ This middleware expects a callback that returns either true to allow events or f
 
 ```php
 use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
-use Patchlevel\EventSourcing\Pipeline\Middleware\FilterEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\FilterEventTranslator;
 
-$middleware = new FilterEventMiddleware(static function (AggregateChanged $event) {
+$middleware = new FilterEventTranslator(static function (AggregateChanged $event) {
     if (!$event instanceof ProfileCreated) {
         return true;
     }
@@ -245,9 +245,9 @@ $middleware = new FilterEventMiddleware(static function (AggregateChanged $event
 With this middleware you can exclude archived events.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeArchivedEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\ExcludeArchivedEventTranslator;
 
-$middleware = new ExcludeArchivedEventMiddleware();
+$middleware = new ExcludeArchivedEventTranslator();
 ```
 !!! warning
 
@@ -258,9 +258,9 @@ $middleware = new ExcludeArchivedEventMiddleware();
 With this middleware you can only allow archived events.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\OnlyArchivedEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\OnlyArchivedEventTranslator;
 
-$middleware = new OnlyArchivedEventMiddleware();
+$middleware = new OnlyArchivedEventTranslator();
 ```
 !!! warning
 
@@ -273,9 +273,9 @@ The first parameter you have to define is the event class that you want to repla
 And as a second parameter a callback, that the old event awaits and a new event returns.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\ReplaceEventMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\ReplaceEventTranslator;
 
-$middleware = new ReplaceEventMiddleware(OldVisited::class, static function (OldVisited $oldVisited) {
+$middleware = new ReplaceEventTranslator(OldVisited::class, static function (OldVisited $oldVisited) {
     return new NewVisited($oldVisited->profileId());
 });
 ```
@@ -302,9 +302,9 @@ The playhead must always be in ascending order so that the data is valid.
 Some middleware can break this order and the middleware `RecalculatePlayheadMiddleware` can fix this problem.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\RecalculatePlayheadTranslator;
 
-$middleware = new RecalculatePlayheadMiddleware();
+$middleware = new RecalculatePlayheadTranslator();
 ```
 !!! note
 
@@ -315,13 +315,13 @@ $middleware = new RecalculatePlayheadMiddleware();
 If you want to group your middleware, you can use one or more `ChainMiddleware`.
 
 ```php
-use Patchlevel\EventSourcing\Pipeline\Middleware\ChainMiddleware;
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
-use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware;
+use Patchlevel\EventSourcing\Message\Translator\ChainTranslator;
+use Patchlevel\EventSourcing\Message\Translator\ExcludeEventTranslator;
+use Patchlevel\EventSourcing\Message\Translator\RecalculatePlayheadTranslator;
 
-$middleware = new ChainMiddleware([
-    new ExcludeEventMiddleware([EmailChanged::class]),
-    new RecalculatePlayheadMiddleware(),
+$middleware = new ChainTranslator([
+    new ExcludeEventTranslator([EmailChanged::class]),
+    new RecalculatePlayheadTranslator(),
 ]);
 ```
 ### Custom middleware
@@ -341,9 +341,9 @@ which should replace the `ProfileCreated` event.
 
 ```php
 use Patchlevel\EventSourcing\Message\Message;
-use Patchlevel\EventSourcing\Pipeline\Middleware\Middleware;
+use Patchlevel\EventSourcing\Message\Translator\Translator;
 
-final class SplitProfileCreatedMiddleware implements Middleware
+final class SplitProfileCreatedMiddleware implements Translator
 {
     public function __invoke(Message $message): array
     {

src/Pipeline/Middleware/ChainMiddleware.php → src/Message/Translator/ChainTranslator.php

@@ -2,17 +2,17 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\Message;
 
 use function array_values;
 
-final class ChainMiddleware implements Middleware
+final class ChainTranslator implements Translator
 {
-    /** @param iterable<Middleware> $middlewares */
+    /** @param iterable<Translator> $translators */
     public function __construct(
-        private readonly iterable $middlewares,
+        private readonly iterable $translators,
     ) {
     }
 
@@ -21,8 +21,8 @@ public function __invoke(Message $message): array
     {
         $messages = [$message];
 
-        foreach ($this->middlewares as $middleware) {
-            $messages = $this->processMiddleware($middleware, $messages);
+        foreach ($this->translators as $middleware) {
+            $messages = $this->process($middleware, $messages);
         }
 
         return $messages;
@@ -33,12 +33,12 @@ public function __invoke(Message $message): array
      *
      * @return list<Message>
      */
-    private function processMiddleware(Middleware $middleware, array $messages): array
+    private function process(Translator $translator, array $messages): array
     {
         $result = [];
 
         foreach ($messages as $message) {
-            $result += $middleware($message);
+            $result += $translator($message);
         }
 
         return array_values($result);

src/Pipeline/Middleware/ExcludeArchivedEventMiddleware.php → src/Message/Translator/ExcludeArchivedEventTranslator.php

@@ -2,13 +2,13 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\ArchivedHeader;
 
-final class ExcludeArchivedEventMiddleware implements Middleware
+final class ExcludeArchivedEventTranslator implements Translator
 {
     /** @return list<Message> */
     public function __invoke(Message $message): array

src/Pipeline/Middleware/ExcludeEventMiddleware.php → src/Message/Translator/ExcludeEventTranslator.php

@@ -2,11 +2,11 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\Message;
 
-final class ExcludeEventMiddleware implements Middleware
+final class ExcludeEventTranslator implements Translator
 {
     /** @param list<class-string> $classes */
     public function __construct(

src/Pipeline/Middleware/FilterEventMiddleware.php → src/Message/Translator/FilterEventTranslator.php

@@ -2,11 +2,11 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\Message;
 
-final class FilterEventMiddleware implements Middleware
+final class FilterEventTranslator implements Translator
 {
     /** @var callable(object $event):bool */
     private $callable;

src/Pipeline/Middleware/IncludeEventMiddleware.php → src/Message/Translator/IncludeEventTranslator.php

@@ -2,11 +2,11 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\Message;
 
-final class IncludeEventMiddleware implements Middleware
+final class IncludeEventTranslator implements Translator
 {
     /** @param list<class-string> $classes */
     public function __construct(

src/Pipeline/Middleware/OnlyArchivedEventMiddleware.php → src/Message/Translator/OnlyArchivedEventTranslator.php

@@ -2,13 +2,13 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\ArchivedHeader;
 
-final class OnlyArchivedEventMiddleware implements Middleware
+final class OnlyArchivedEventTranslator implements Translator
 {
     /** @return list<Message> */
     public function __invoke(Message $message): array

src/Pipeline/Middleware/RecalculatePlayheadMiddleware.php → src/Message/Translator/RecalculatePlayheadTranslator.php

@@ -2,14 +2,14 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
 use Patchlevel\EventSourcing\Message\Message;
 
 use function array_key_exists;
 
-final class RecalculatePlayheadMiddleware implements Middleware
+final class RecalculatePlayheadTranslator implements Translator
 {
     /** @var array<string, array<string, positive-int>> */
     private array $index = [];

src/Pipeline/Middleware/ReplaceEventMiddleware.php → src/Message/Translator/ReplaceEventTranslator.php

@@ -2,12 +2,12 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\Message;
 
 /** @template T of object */
-final class ReplaceEventMiddleware implements Middleware
+final class ReplaceEventTranslator implements Translator
 {
     /** @var callable(T $event):object */
     private $callable;

src/Pipeline/Middleware/Middleware.php → src/Message/Translator/Translator.php

@@ -2,11 +2,11 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Message\Message;
 
-interface Middleware
+interface Translator
 {
     /** @return list<Message> */
     public function __invoke(Message $message): array;

src/Pipeline/Middleware/UntilEventMiddleware.php → src/Message/Translator/UntilEventTranslator.php

@@ -2,13 +2,13 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
+namespace Patchlevel\EventSourcing\Message\Translator;
 
 use DateTimeImmutable;
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
 use Patchlevel\EventSourcing\Message\Message;
 
-final class UntilEventMiddleware implements Middleware
+final class UntilEventTranslator implements Translator
 {
     public function __construct(
         private readonly DateTimeImmutable $until,