allow passing metadata matcher with projection running configuration

Author: unixslayer

composer.json

@@ -124,8 +124,7 @@
         "laminas/laminas-code": "^4",
         "jms/serializer": "^3.32",
         "laravel/framework": "^9.5.2|^10.0|^11.0|^12.0|^13.0",
-        "prooph/event-store": "<=7.11.1",
-        "prooph/pdo-event-store": "^1.15.1 <1.16.0",
+        "prooph/pdo-event-store": "^1.16.2",
         "psr/log": "^2.0|^3.0",
         "queue-interop/queue-interop": "^0.8",
         "ramsey/uuid": "^4.0",

packages/PdoEventSourcing/composer.json

@@ -34,8 +34,7 @@
     },
     "require": {
         "ecotone/dbal": "~1.252.0",
-        "prooph/event-store": "<=7.11.1",
-        "prooph/pdo-event-store": "^1.15.1 <1.16.0"
+        "prooph/pdo-event-store": "^1.16.2"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.6|^10.5|^11.0",
@@ -80,4 +79,4 @@
             "wikimedia/composer-merge-plugin": true
         }
     }
-}
+}

packages/PdoEventSourcing/src/CachingInMemoryReadModelProjector.php

@@ -5,6 +5,7 @@
 namespace Ecotone\EventSourcing;
 
 use Closure;
+use Prooph\EventStore\Metadata\MetadataMatcher;
 use Prooph\EventStore\Projection\ReadModel;
 use Prooph\EventStore\Projection\ReadModelProjector;
 
@@ -27,6 +28,13 @@ public function init(Closure $callback): ReadModelProjector
         return $this;
     }
 
+    public function withMetadataMatcher(?MetadataMatcher $metadataMatcher = null): ReadModelProjector
+    {
+        $this->inMemoryEventStoreReadModelProjector->withMetadataMatcher($metadataMatcher);
+
+        return $this;
+    }
+
     public function fromStream(string $streamName): ReadModelProjector
     {
         if ($this->isFromStreamSetup) {

packages/PdoEventSourcing/src/InMemory/CachingInMemoryProjectionManager.php

@@ -5,6 +5,7 @@
 namespace Ecotone\EventSourcing\InMemory;
 
 use Ecotone\EventSourcing\CachingInMemoryReadModelProjector;
+use Prooph\EventStore\Projection\ProjectionManager;
 use Prooph\EventStore\Projection\ProjectionStatus;
 use Prooph\EventStore\Projection\Projector;
 use Prooph\EventStore\Projection\Query;
@@ -14,7 +15,7 @@
 /**
  * licence Apache-2.0
  */
-class CachingInMemoryProjectionManager implements \Prooph\EventStore\Projection\ProjectionManager
+class CachingInMemoryProjectionManager implements ProjectionManager
 {
     /**
      * @var ReadModelProjector[]

packages/PdoEventSourcing/src/ProjectionRunningConfiguration.php

@@ -2,9 +2,11 @@
 
 namespace Ecotone\EventSourcing;
 
+use Assert\Assertion;
 use Ecotone\Messaging\Config\Container\DefinedObject;
 use Ecotone\Messaging\Config\Container\Definition;
 use InvalidArgumentException;
+use Prooph\EventStore\Metadata\MetadataMatcher;
 
 /**
  * licence Apache-2.0
@@ -35,6 +37,9 @@ class ProjectionRunningConfiguration implements DefinedObject
     public const OPTION_LOAD_COUNT = 'load_count';
     public const DEFAULT_LOAD_COUNT = null;
 
+    public const OPTION_METADATA_MATCHER = 'metadata_matcher';
+    public const DEFAULT_METADATA_MATCHER = null;
+
     public const OPTION_IS_TESTING_SETUP = 'isTestingSetup';
     public const DEFAULT_IS_TESTING_SETUP = false;
 
@@ -54,6 +59,7 @@ public function __construct(
             self::OPTION_UPDATE_LOCK_TIMEOUT_AFTER => self::DEFAULT_UPDATE_LOCK_TIMEOUT_AFTER,
             self::OPTION_IS_TESTING_SETUP => self::DEFAULT_IS_TESTING_SETUP,
             self::OPTION_LOAD_COUNT => self::DEFAULT_LOAD_COUNT,
+            self::OPTION_METADATA_MATCHER => self::DEFAULT_METADATA_MATCHER,
         ];
     }
 
@@ -100,6 +106,10 @@ public function isTestingSetup(): bool
 
     public function withOption(string $key, mixed $value): static
     {
+        if ($key === self::OPTION_METADATA_MATCHER && $value !== null) {
+            Assertion::isInstanceOf($value, MetadataMatcher::class);
+        }
+
         $self = clone $this;
         $self->options[$key] = $value;
 

packages/PdoEventSourcing/src/ProjectionSetupConfiguration.php

@@ -2,6 +2,7 @@
 
 namespace Ecotone\EventSourcing;
 
+use Assert\Assertion;
 use Ecotone\Messaging\Config\Container\DefinedObject;
 use Ecotone\Messaging\Config\Container\Definition;
 use Ecotone\Messaging\NullableMessageChannel;

packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php

@@ -4,6 +4,7 @@
 
 use Ecotone\EventSourcing\EventSourcingConfiguration;
 use Ecotone\EventSourcing\ProjectionExecutor;
+use Ecotone\EventSourcing\ProjectionRunningConfiguration;
 use Ecotone\EventSourcing\ProjectionSetupConfiguration;
 use Ecotone\EventSourcing\ProjectionStreamSource;
 use Ecotone\Messaging\Gateway\MessagingEntrypoint;
@@ -12,9 +13,12 @@
 use Prooph\Common\Messaging\Message;
 use Prooph\EventStore\Exception\ProjectionNotFound;
 use Prooph\EventStore\Exception\RuntimeException;
+use Prooph\EventStore\Metadata\MetadataMatcher;
 use Prooph\EventStore\Pdo\Projection\MariaDbProjectionManager;
 use Prooph\EventStore\Pdo\Projection\MySqlProjectionManager;
 use Prooph\EventStore\Pdo\Projection\PostgresProjectionManager;
+use Prooph\EventStore\Projection\MetadataAwareProjector;
+use Prooph\EventStore\Projection\MetadataAwareReadModelProjector;
 use Prooph\EventStore\Projection\ProjectionManager;
 use Prooph\EventStore\Projection\ProjectionStatus;
 use Prooph\EventStore\Projection\Projector;
@@ -74,12 +78,28 @@ public function createQuery(): Query
 
     public function createProjection(string $name, array $options = []): Projector
     {
-        return $this->getProjectionManager()->createProjection($name, $options);
+        $projection = $this->getProjectionManager()->createProjection($name, $options);
+
+        $metadataMatcher = $options[ProjectionRunningConfiguration::OPTION_METADATA_MATCHER] ?? null;
+
+        if ($metadataMatcher instanceof MetadataMatcher && $projection instanceof MetadataAwareProjector) {
+            $projection = $projection->withMetadataMatcher($metadataMatcher);
+        }
+
+        return $projection;
     }
 
     public function createReadModelProjection(string $name, ReadModel $readModel, array $options = []): ReadModelProjector
     {
-        return $this->getProjectionManager()->createReadModelProjection($name, $readModel, $options);
+        $projection = $this->getProjectionManager()->createReadModelProjection($name, $readModel, $options);
+
+        $metadataMatcher = $options[ProjectionRunningConfiguration::OPTION_METADATA_MATCHER] ?? null;
+
+        if ($metadataMatcher instanceof MetadataMatcher && $projection instanceof MetadataAwareReadModelProjector) {
+            $projection = $projection->withMetadataMatcher($metadataMatcher);
+        }
+
+        return $projection;
     }
 
     public function deleteProjection(string $name, bool $deleteEmittedEvents): void

packages/PdoEventSourcing/tests/Fixture/ProjectionWithMetadataMatcher/ProjectionWithMetadataMatcherConfig.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\EventSourcing\Fixture\ProjectionWithMetadataMatcher;
+
+use Ecotone\EventSourcing\ProjectionRunningConfiguration;
+use Ecotone\Messaging\Attribute\ServiceContext;
+use Ecotone\Messaging\Endpoint\PollingMetadata;
+use Prooph\EventStore\Metadata\FieldType;
+use Prooph\EventStore\Metadata\MetadataMatcher;
+use Prooph\EventStore\Metadata\Operator;
+use Prooph\EventStore\Pdo\Projection\GapDetection;
+use Prooph\EventStore\Pdo\Projection\PdoEventStoreReadModelProjector;
+use Test\Ecotone\EventSourcing\Fixture\TicketWithPollingProjection\InProgressTicketList;
+
+final class ProjectionWithMetadataMatcherConfig
+{
+    #[ServiceContext]
+    public function enablePollingProjection(): ProjectionRunningConfiguration
+    {
+        $metadataMatcher = (new MetadataMatcher())
+            ->withMetadataMatch('test', Operator::EQUALS(), 'false', FieldType::METADATA())
+        ;
+
+        return ProjectionRunningConfiguration::createPolling(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION)
+            ->withTestingSetup()
+            ->withOption(PdoEventStoreReadModelProjector::OPTION_GAP_DETECTION, null)
+            ->withOption(ProjectionRunningConfiguration::OPTION_METADATA_MATCHER, $metadataMatcher);
+    }
+}

packages/PdoEventSourcing/tests/Integration/ProjectionWithMetadataMatcherTest.php

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\EventSourcing\Integration;
+
+use Ecotone\EventSourcing\EventSourcingConfiguration;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Enqueue\Dbal\DbalConnectionFactory;
+use Test\Ecotone\EventSourcing\EventSourcingMessagingTestCase;
+use Test\Ecotone\EventSourcing\Fixture\Ticket\Command\CloseTicket;
+use Test\Ecotone\EventSourcing\Fixture\Ticket\Command\RegisterTicket;
+use Test\Ecotone\EventSourcing\Fixture\Ticket\TicketEventConverter;
+use Test\Ecotone\EventSourcing\Fixture\TicketWithPollingProjection\InProgressTicketList;
+use Test\Ecotone\EventSourcing\Fixture\TicketWithPollingProjection\ProjectionConfiguration;
+
+final class ProjectionWithMetadataMatcherTest extends EventSourcingMessagingTestCase
+{
+    public function test_configured_metadata_matcher_is_used(): void
+    {
+        $connectionFactory = self::getConnectionFactory();
+        $connection = $connectionFactory->createContext()
+            ->getDbalConnection()
+        ;
+
+        $ecotoneLite = EcotoneLite::bootstrapFlowTestingWithEventStore(
+            classesToResolve: [ProjectionConfiguration::class, InProgressTicketList::class],
+            containerOrAvailableServices: [new InProgressTicketList($connection), new TicketEventConverter(), DbalConnectionFactory::class => $connectionFactory],
+            configuration: ServiceConfiguration::createWithDefaults()
+                ->withEnvironment('prod')
+                ->withNamespaces([
+                    'Test\Ecotone\EventSourcing\Fixture\Ticket',
+                    'Test\Ecotone\EventSourcing\Fixture\ProjectionWithMetadataMatcher',
+                ])
+                ->withExtensionObjects([
+                    EventSourcingConfiguration::createWithDefaults(),
+                ]),
+            pathToRootCatalog: __DIR__ . '/../../',
+            runForProductionEventStore: true
+        );
+
+        $ecotoneLite->initializeProjection(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION);
+
+        $ecotoneLite->sendCommand(new RegisterTicket('123', 'Johnny', 'alert'), metadata: ['test' => 'false']);
+        $ecotoneLite->sendCommand(new RegisterTicket('124', 'Johnny', 'alert'), metadata: ['test' => 'false']);
+        $ecotoneLite->sendCommand(new RegisterTicket('125', 'Johnny', 'alert'), metadata: ['test' => 'true']);
+        $ecotoneLite->sendCommand(new RegisterTicket('126', 'Johnny', 'alert'), metadata: ['test' => 'false']);
+        $ecotoneLite->sendCommand(new RegisterTicket('127', 'Johnny', 'alert'), metadata: ['test' => 'true']);
+
+        self::assertEquals([], $ecotoneLite->sendQueryWithRouting('getInProgressTickets'));
+
+        $ecotoneLite->run(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION);
+
+        self::assertEquals([
+            ['ticket_id' => '123', 'ticket_type' => 'alert'],
+            ['ticket_id' => '124', 'ticket_type' => 'alert'],
+            ['ticket_id' => '126', 'ticket_type' => 'alert'],
+        ], $ecotoneLite->sendQueryWithRouting('getInProgressTickets'));
+
+        $ecotoneLite->sendCommand(new CloseTicket('123'), metadata: ['test' => 'false']);
+
+        self::assertEquals([
+            ['ticket_id' => '123', 'ticket_type' => 'alert'],
+            ['ticket_id' => '124', 'ticket_type' => 'alert'],
+            ['ticket_id' => '126', 'ticket_type' => 'alert'],
+        ], $ecotoneLite->sendQueryWithRouting('getInProgressTickets'));
+
+        $ecotoneLite->run(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION);
+
+        self::assertEquals([
+            ['ticket_id' => '124', 'ticket_type' => 'alert'],
+            ['ticket_id' => '126', 'ticket_type' => 'alert'],
+        ], $ecotoneLite->sendQueryWithRouting('getInProgressTickets'));
+    }
+}