Merge pull request #516 from patchlevel/projectionist-bench

add projectionist benchmark
patchlevel · Mar 3, 2024 · f898567 · f898567
2 parents d0bf33a + 6fa2d86
commit f898567
Show file tree

Hide file tree

Showing 8 changed files with 160 additions and 20 deletions.
diff --git a/baseline.xml b/baseline.xml
@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<files psalm-version="5.22.1@e9dad66e11274315dac27e08349c628c7d6a1a43">
+<files psalm-version="5.22.2@d768d914152dbbf3486c36398802f74e80cfde48">
   <file src="src/Aggregate/AggregateRootBehaviour.php">
     <UnsafeInstantiation>
       <code><![CDATA[new static()]]></code>
@@ -100,6 +100,15 @@
       <code><![CDATA[$name]]></code>
     </PropertyNotSetInConstructor>
   </file>
+  <file src="tests/Benchmark/ProjectionistBench.php">
+    <MissingConstructor>
+      <code><![CDATA[$bus]]></code>
+      <code><![CDATA[$id]]></code>
+      <code><![CDATA[$projectionist]]></code>
+      <code><![CDATA[$repository]]></code>
+      <code><![CDATA[$store]]></code>
+    </MissingConstructor>
+  </file>
   <file src="tests/Benchmark/SimpleSetupBench.php">
     <MissingConstructor>
       <code><![CDATA[$bus]]></code>

diff --git a/docs/pages/getting_started.md b/docs/pages/getting_started.md
@@ -181,7 +181,7 @@ final class HotelProjector
      */
     public function getHotels(): array 
     {
-        return $this->db->fetchAllAssociative("SELECT id, name, guests FROM ${this->table()};");
+        return $this->db->fetchAllAssociative("SELECT id, name, guests FROM {$this->table()};");
     }
 
     #[Subscribe(HotelCreated::class)]
@@ -203,7 +203,7 @@ final class HotelProjector
     public function handleGuestIsCheckedIn(Message $message): void
     {
         $this->db->executeStatement(
-            "UPDATE ${this->table()} SET guests = guests + 1 WHERE id = ?;",
+            "UPDATE {$this->table()} SET guests = guests + 1 WHERE id = ?;",
             [$message->aggregateId()]
         );
     }
@@ -212,21 +212,21 @@ final class HotelProjector
     public function handleGuestIsCheckedOut(Message $message): void
     {
         $this->db->executeStatement(
-            "UPDATE ${this->table()} SET guests = guests - 1 WHERE id = ?;",
+            "UPDATE {$this->table()} SET guests = guests - 1 WHERE id = ?;",
             [$message->aggregateId()]
         );
     }
 
     #[Setup]
     public function create(): void
     {
-        $this->db->executeStatement("CREATE TABLE IF NOT EXISTS ${this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR, guests INTEGER);");
+        $this->db->executeStatement("CREATE TABLE IF NOT EXISTS {$this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR, guests INTEGER);");
     }
 
     #[Teardown]
     public function drop(): void
     {
-        $this->db->executeStatement("DROP TABLE IF EXISTS ${this->table()};");
+        $this->db->executeStatement("DROP TABLE IF EXISTS {$this->table()};");
     }
 
     private function table(): string

diff --git a/docs/pages/projection.md b/docs/pages/projection.md
@@ -69,7 +69,7 @@ final class ProfileProjector
         $profileCreated = $message->event();
 
         $this->connection->executeStatement(
-            "INSERT INTO ${this->table()} (id, name) VALUES(?, ?);",
+            "INSERT INTO {$this->table()} (id, name) VALUES(?, ?);",
             [
                 'id' => $profileCreated->profileId->toString(),
                 'name' => $profileCreated->name
@@ -123,14 +123,14 @@ final class ProfileProjector
     public function create(): void
     {
         $this->connection->executeStatement(
-            "CREATE TABLE IF NOT EXISTS ${this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL);"
+            "CREATE TABLE IF NOT EXISTS {$this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL);"
         );
     }
 
     #[Teardown]
     public function drop(): void
     {
-        $this->connection->executeStatement("DROP TABLE IF EXISTS ${this->table()};");
+        $this->connection->executeStatement("DROP TABLE IF EXISTS {$this->table()};");
     }
 
     private function table(): string 
@@ -175,7 +175,7 @@ final class ProfileProjector
      */
     public function getProfiles(): array 
     {
-        return $this->connection->fetchAllAssociative("SELECT id, name FROM ${this->table()};");
+        return $this->connection->fetchAllAssociative("SELECT id, name FROM {$this->table()};");
     }
 
     private function table(): string 

diff --git a/phpcs.xml.dist b/phpcs.xml.dist
@@ -14,5 +14,6 @@
     <rule ref="PatchlevelCodingStandard">
         <exclude name="Generic.Files.LineLength.TooLong"/>
         <exclude name="SlevomatCodingStandard.PHP.RequireExplicitAssertion.RequiredExplicitAssertion"/>
+        <exclude name="Squiz.Strings.DoubleQuoteUsage.ContainsVar"/>
     </rule>
 </ruleset>
diff --git a/src/Projection/Projectionist/DefaultProjectionist.php b/src/Projection/Projectionist/DefaultProjectionist.php
@@ -36,7 +36,7 @@ final class DefaultProjectionist implements Projectionist
 
     /** @param iterable<object> $projectors */
     public function __construct(
-        private readonly Store $streamableMessageStore,
+        private readonly Store $messageStore,
         private readonly ProjectionStore $projectionStore,
         private readonly iterable $projectors,
         private readonly RetryStrategy $retryStrategy = new ClockBasedRetryStrategy(),
@@ -86,7 +86,7 @@ function ($projections) use ($limit): void {
                 $stream = null;
 
                 try {
-                    $stream = $this->streamableMessageStore->load(
+                    $stream = $this->messageStore->load(
                         new Criteria(fromIndex: $startIndex),
                     );
 
@@ -215,7 +215,7 @@ function (array $projections) use ($limit): void {
 
                 try {
                     $criteria = new Criteria(fromIndex: $startIndex);
-                    $stream = $this->streamableMessageStore->load($criteria);
+                    $stream = $this->messageStore->load($criteria);
 
                     $messageCounter = 0;
 
@@ -868,7 +868,7 @@ private function projectorMetadata(object $projector): ProjectorMetadata
 
     private function latestIndex(): int
     {
-        $stream = $this->streamableMessageStore->load(null, 1, null, true);
+        $stream = $this->messageStore->load(null, 1, null, true);
 
         return $stream->index() ?: 0;
     }

diff --git a/tests/Benchmark/BasicImplementation/Processor/SendEmailProcessor.php b/tests/Benchmark/BasicImplementation/Processor/SendEmailProcessor.php
@@ -4,11 +4,13 @@
 
 namespace Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Processor;
 
+use Patchlevel\EventSourcing\Attribute\Projector;
 use Patchlevel\EventSourcing\Attribute\Subscribe;
 use Patchlevel\EventSourcing\EventBus\Message;
 use Patchlevel\EventSourcing\Tests\Integration\BasicImplementation\Events\ProfileCreated;
 use Patchlevel\EventSourcing\Tests\Integration\BasicImplementation\SendEmailMock;
 
+#[Projector('send_email')]
 final class SendEmailProcessor
 {
     #[Subscribe(ProfileCreated::class)]

diff --git a/tests/Benchmark/BasicImplementation/Projection/ProfileProjector.php b/tests/Benchmark/BasicImplementation/Projection/ProfileProjector.php
@@ -10,13 +10,17 @@
 use Patchlevel\EventSourcing\Attribute\Subscribe;
 use Patchlevel\EventSourcing\Attribute\Teardown;
 use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\Projection\Projector\ProjectorUtil;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Events\NameChanged;
 use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Events\ProfileCreated;
 
 use function assert;
 
-#[Projector('dummy_1')]
+#[Projector('profile')]
 final class ProfileProjector
 {
+    use ProjectorUtil;
+
     public function __construct(
         private Connection $connection,
     ) {
@@ -25,28 +29,47 @@ public function __construct(
     #[Setup]
     public function create(): void
     {
-        $this->connection->executeStatement('CREATE TABLE IF NOT EXISTS projection_profile (id VARCHAR PRIMARY KEY, name VARCHAR);');
+        $this->connection->executeStatement("CREATE TABLE IF NOT EXISTS {$this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR);");
     }
 
     #[Teardown]
     public function drop(): void
     {
-        $this->connection->executeStatement('DROP TABLE IF EXISTS projection_profile;');
+        $this->connection->executeStatement("DROP TABLE IF EXISTS {$this->table()};");
     }
 
     #[Subscribe(ProfileCreated::class)]
-    public function handleProfileCreated(Message $message): void
+    public function onProfileCreated(Message $message): void
     {
         $profileCreated = $message->event();
 
         assert($profileCreated instanceof ProfileCreated);
 
-        $this->connection->executeStatement(
-            'INSERT INTO projection_profile (`id`, `name`) VALUES(:id, :name);',
+        $this->connection->insert(
+            $this->table(),
             [
                 'id' => $profileCreated->profileId->toString(),
                 'name' => $profileCreated->name,
             ],
         );
     }
+
+    #[Subscribe(NameChanged::class)]
+    public function onNameChanged(Message $message): void
+    {
+        $nameChanged = $message->event();
+
+        assert($nameChanged instanceof NameChanged);
+
+        $this->connection->update(
+            $this->table(),
+            ['name' => $nameChanged->name],
+            ['id' => $message->aggregateId()],
+        );
+    }
+
+    public function table(): string
+    {
+        return 'projection_' . $this->projectorId();
+    }
 }
diff --git a/tests/Benchmark/ProjectionistBench.php b/tests/Benchmark/ProjectionistBench.php
@@ -0,0 +1,105 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Benchmark;
+
+use Doctrine\DBAL\Driver\PDO\SQLite\Driver;
+use Doctrine\DBAL\DriverManager;
+use Patchlevel\EventSourcing\Aggregate\AggregateRootId;
+use Patchlevel\EventSourcing\EventBus\DefaultEventBus;
+use Patchlevel\EventSourcing\EventBus\EventBus;
+use Patchlevel\EventSourcing\Projection\Projection\Store\DoctrineStore;
+use Patchlevel\EventSourcing\Projection\Projectionist\DefaultProjectionist;
+use Patchlevel\EventSourcing\Projection\Projectionist\Projectionist;
+use Patchlevel\EventSourcing\Repository\DefaultRepository;
+use Patchlevel\EventSourcing\Repository\Repository;
+use Patchlevel\EventSourcing\Schema\ChainSchemaConfigurator;
+use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
+use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
+use Patchlevel\EventSourcing\Store\DoctrineDbalStore;
+use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Aggregate\Profile;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Processor\SendEmailProcessor;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\ProfileId;
+use Patchlevel\EventSourcing\Tests\Benchmark\BasicImplementation\Projection\ProfileProjector;
+use PhpBench\Attributes as Bench;
+
+use function file_exists;
+use function unlink;
+
+#[Bench\BeforeMethods('setUp')]
+final class ProjectionistBench
+{
+    private const DB_PATH = __DIR__ . '/BasicImplementation/data/db.sqlite3';
+
+    private Store $store;
+    private EventBus $bus;
+    private Repository $repository;
+
+    private Projectionist $projectionist;
+
+    private AggregateRootId $id;
+
+    public function setUp(): void
+    {
+        if (file_exists(self::DB_PATH)) {
+            unlink(self::DB_PATH);
+        }
+
+        $connection = DriverManager::getConnection([
+            'driverClass' => Driver::class,
+            'path' => self::DB_PATH,
+        ]);
+
+        $this->bus = DefaultEventBus::create();
+
+        $this->store = new DoctrineDbalStore(
+            $connection,
+            DefaultEventSerializer::createFromPaths([__DIR__ . '/BasicImplementation/Events']),
+            'eventstore',
+        );
+
+        $this->repository = new DefaultRepository($this->store, $this->bus, Profile::metadata());
+
+        $projectionStore = new DoctrineStore(
+            $connection,
+        );
+
+        $schemaDirector = new DoctrineSchemaDirector(
+            $connection,
+            new ChainSchemaConfigurator([
+                $this->store,
+                $projectionStore,
+            ]),
+        );
+
+        $schemaDirector->create();
+
+        $this->id = ProfileId::v7();
+
+        $profile = Profile::create($this->id, 'Peter');
+
+        for ($i = 1; $i < 10_000; $i++) {
+            $profile->changeName('Peter ' . $i);
+        }
+
+        $this->repository->save($profile);
+
+        $this->projectionist = new DefaultProjectionist(
+            $this->store,
+            $projectionStore,
+            [
+                new ProfileProjector($connection),
+                new SendEmailProcessor(),
+            ],
+        );
+    }
+
+    #[Bench\Revs(20)]
+    public function benchHandle10000Events(): void
+    {
+        $this->projectionist->boot();
+        $this->projectionist->remove();
+    }
+}