Skip to content

Commit 6043a4e

Browse files
DavidBaduraDavidBadura
committed
add stateful subscriber feature
1 parent d70ac89 commit 6043a4e

8 files changed

Lines changed: 307 additions & 2 deletions

File tree

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@
3434
"php": "~8.2.0 || ~8.3.0 || ~8.4.0 || ~8.5.0",
3535
"doctrine/dbal": "^4.4.0",
3636
"doctrine/migrations": "^3.3.2",
37-
"patchlevel/hydrator": "^1.8.0",
37+
"patchlevel/hydrator": "^1.19.0",
3838
"patchlevel/worker": "^1.4.0",
3939
"psr/cache": "^2.0.0 || ^3.0.0",
4040
"psr/clock": "^1.0",

‎composer.lock‎

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎phpstan-baseline.neon‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,18 @@ parameters:
222222
count: 1
223223
path: src/Subscription/RetryStrategy/ClockBasedRetryStrategy.php
224224

225+
-
226+
message: '#^Parameter \#1 \$json of function json_decode expects string, mixed given\.$#'
227+
identifier: argument.type
228+
count: 1
229+
path: src/Subscription/StatefulSubscriber/DoctrineStatefulSubscriberStore.php
230+
231+
-
232+
message: '#^Parameter \#2 \$data of method Patchlevel\\Hydrator\\HydratorWithContext\:\:hydrate\(\) expects array\<string, mixed\>, mixed given\.$#'
233+
identifier: argument.type
234+
count: 1
235+
path: src/Subscription/StatefulSubscriber/DoctrineStatefulSubscriberStore.php
236+
225237
-
226238
message: '#^Parameter \#3 \$errorContext of class Patchlevel\\EventSourcing\\Subscription\\SubscriptionError constructor expects list\<array\{namespace\: string, short_name\: string, class\: class\-string, message\: string, code\: int\|string, file\: string, line\: int, trace\: list\<array\{file\?\: string, line\?\: int, function\?\: string, class\?\: string, type\?\: string, args\?\: array\<mixed\>\}\>\}\>\|null, mixed given\.$#'
227239
identifier: argument.type

‎src/Subscription/StatefulSubscriber/DoctrineStatefulSubscriberStore.php‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Subscription\StatefulSubscriber;
6+
7+
use Doctrine\DBAL\Connection;
8+
use Doctrine\DBAL\Schema\Schema;
9+
use Doctrine\DBAL\Types\Types;
10+
use Patchlevel\EventSourcing\Metadata\Subscriber\AttributeSubscriberMetadataFactory;
11+
use Patchlevel\EventSourcing\Metadata\Subscriber\SubscriberMetadataFactory;
12+
use Patchlevel\EventSourcing\Schema\DoctrineHelper;
13+
use Patchlevel\EventSourcing\Schema\DoctrineSchemaConfigurator;
14+
use Patchlevel\Hydrator\HydratorWithContext;
15+
use Patchlevel\Hydrator\MetadataHydrator;
16+
17+
use function json_decode;
18+
use function json_encode;
19+
20+
final readonly class DoctrineStatefulSubscriberStore implements StatefulSubscriberStore, DoctrineSchemaConfigurator
21+
{
22+
public function __construct(
23+
private Connection $connection,
24+
private HydratorWithContext $hydrator = new MetadataHydrator(),
25+
private SubscriberMetadataFactory $subscriberMetadataFactory = new AttributeSubscriberMetadataFactory(),
26+
private string $tableName = 'stateful_subscriber_state',
27+
) {
28+
}
29+
30+
public function store(StatefulSubscriber $subscriber): void
31+
{
32+
$subscriberId = $this->subscriberId($subscriber);
33+
$data = $this->hydrator->extract($subscriber);
34+
35+
$this->connection->insert(
36+
$this->tableName,
37+
[
38+
'id' => $subscriberId,
39+
'state' => json_encode($data),
40+
],
41+
);
42+
}
43+
44+
public function load(StatefulSubscriber $subscriber): void
45+
{
46+
$subscriberId = $this->subscriberId($subscriber);
47+
48+
$data = $this->connection->fetchAssociative(
49+
'SELECT * FROM ' . $this->tableName . ' WHERE id = ?',
50+
[$subscriberId],
51+
);
52+
53+
if (!$data) {
54+
return;
55+
}
56+
57+
$this->hydrator->hydrate(
58+
$subscriber::class,
59+
json_decode($data['state'], true),
60+
[HydratorWithContext::OBJECT_TO_POPULATE => $subscriber],
61+
);
62+
}
63+
64+
public function configureSchema(Schema $schema, Connection $connection): void
65+
{
66+
if (!DoctrineHelper::sameDatabase($this->connection, $connection)) {
67+
return;
68+
}
69+
70+
$table = $schema->createTable($this->tableName);
71+
72+
$table->addColumn('id', Types::STRING)
73+
->setLength(255)
74+
->setNotnull(true);
75+
$table->addColumn('state', Types::JSON)
76+
->setNotnull(true);
77+
78+
$table->setPrimaryKey(['id']);
79+
}
80+
81+
public function subscriberId(StatefulSubscriber $projection): string
82+
{
83+
return $this->subscriberMetadataFactory->metadata($projection::class)->id;
84+
}
85+
}

‎src/Subscription/StatefulSubscriber/StatefulSubscriber.php‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Subscription\StatefulSubscriber;
6+
7+
use Patchlevel\EventSourcing\Subscription\Subscriber\BatchableSubscriber;
8+
use Patchlevel\Hydrator\Attribute\Ignore;
9+
10+
abstract class StatefulSubscriber implements BatchableSubscriber
11+
{
12+
public function __construct(
13+
#[Ignore]
14+
private readonly StatefulSubscriberStore $store,
15+
) {
16+
$this->store->load($this);
17+
}
18+
19+
public function beginBatch(): void
20+
{
21+
// do nothing
22+
}
23+
24+
public function commitBatch(): void
25+
{
26+
$this->store->store($this);
27+
}
28+
29+
public function rollbackBatch(): void
30+
{
31+
$this->store->load($this);
32+
}
33+
34+
public function forceCommit(): bool
35+
{
36+
return false;
37+
}
38+
}

‎src/Subscription/StatefulSubscriber/StatefulSubscriberStore.php‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Subscription\StatefulSubscriber;
6+
7+
interface StatefulSubscriberStore
8+
{
9+
public function store(StatefulSubscriber $subscriber): void;
10+
11+
public function load(StatefulSubscriber $subscriber): void;
12+
}

‎tests/Integration/Subscription/Subscriber/ProfileInlineStatefulSubscriber.php‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber;
6+
7+
use Patchlevel\EventSourcing\Attribute\Projector;
8+
use Patchlevel\EventSourcing\Attribute\Subscribe;
9+
use Patchlevel\EventSourcing\Subscription\StatefulSubscriber\StatefulSubscriber;
10+
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Events\ProfileCreated;
11+
use Patchlevel\EventSourcing\Tests\Integration\Subscription\ProfileId;
12+
13+
#[Projector('profile_inline')]
14+
final class ProfileInlineStatefulSubscriber extends StatefulSubscriber
15+
{
16+
/** @var array<string, string> */
17+
public array $profiles = [];
18+
19+
#[Subscribe(ProfileCreated::class)]
20+
public function handleProfileCreated(ProfileCreated $profileCreated): void
21+
{
22+
$this->profiles[$profileCreated->profileId->toString()] = $profileCreated->name;
23+
}
24+
25+
public function findById(ProfileId $id): string|null
26+
{
27+
return $this->profiles[$id->toString()] ?? null;
28+
}
29+
}

‎tests/Integration/Subscription/SubscriptionTest.php‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngineCriteria;
3434
use Patchlevel\EventSourcing\Subscription\RetryStrategy\ClockBasedRetryStrategy;
3535
use Patchlevel\EventSourcing\Subscription\RunMode;
36+
use Patchlevel\EventSourcing\Subscription\StatefulSubscriber\DoctrineStatefulSubscriberStore;
3637
use Patchlevel\EventSourcing\Subscription\Status;
3738
use Patchlevel\EventSourcing\Subscription\Store\DoctrineSubscriptionStore;
3839
use Patchlevel\EventSourcing\Subscription\Subscriber\ArgumentResolver\LookupResolver;
@@ -43,6 +44,7 @@
4344
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\ErrorProducerWithSelfRecoverySubscriber;
4445
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\LookupSubscriber;
4546
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\MigrateAggregateToStreamStoreSubscriber;
47+
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\ProfileInlineStatefulSubscriber;
4648
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\ProfileNewProjection;
4749
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\ProfileProcessor;
4850
use Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber\ProfileProjection;
@@ -53,6 +55,7 @@
5355

5456
use function gc_collect_cycles;
5557
use function iterator_to_array;
58+
use function json_decode;
5659
use function sprintf;
5760

5861
#[CoversNothing]
@@ -1632,6 +1635,132 @@ class {
16321635
self::assertEquals(RunMode::FromNow, $subscriptions[0]->runMode());
16331636
}
16341637

1638+
public function testStatefulSubscriber(): void
1639+
{
1640+
$store = new DoctrineDbalStore(
1641+
$this->connection,
1642+
DefaultEventSerializer::createFromPaths([__DIR__ . '/Events']),
1643+
);
1644+
1645+
$clock = new FrozenClock(new DateTimeImmutable('2021-01-01T00:00:00'));
1646+
1647+
$subscriptionStore = new DoctrineSubscriptionStore(
1648+
$this->connection,
1649+
$clock,
1650+
);
1651+
1652+
$manager = new DefaultRepositoryManager(
1653+
new AggregateRootRegistry(['profile' => Profile::class]),
1654+
$store,
1655+
);
1656+
1657+
$repository = $manager->get(Profile::class);
1658+
1659+
$stateStore = new DoctrineStatefulSubscriberStore($this->connection);
1660+
1661+
$schemaDirector = new DoctrineSchemaDirector(
1662+
$this->connection,
1663+
new ChainDoctrineSchemaConfigurator([
1664+
$store,
1665+
$subscriptionStore,
1666+
$stateStore,
1667+
]),
1668+
);
1669+
1670+
$schemaDirector->create();
1671+
1672+
$subscriberRepository = new MetadataSubscriberAccessorRepository([
1673+
new ProfileInlineStatefulSubscriber(
1674+
$stateStore,
1675+
),
1676+
]);
1677+
1678+
$engine = new DefaultSubscriptionEngine(
1679+
new EventFilteredStoreMessageLoader($store, new AttributeEventMetadataFactory(), $subscriberRepository),
1680+
$subscriptionStore,
1681+
$subscriberRepository,
1682+
);
1683+
1684+
self::assertEquals(
1685+
[
1686+
new Subscription(
1687+
'profile_inline',
1688+
'projector',
1689+
lastSavedAt: new DateTimeImmutable('2021-01-01T00:00:00'),
1690+
),
1691+
],
1692+
$engine->subscriptions(),
1693+
);
1694+
1695+
$result = $engine->setup();
1696+
1697+
self::assertEquals([], $result->errors);
1698+
1699+
$result = $engine->boot();
1700+
1701+
self::assertEquals(0, $result->processedMessages);
1702+
self::assertEquals([], $result->errors);
1703+
1704+
self::assertEquals(
1705+
[
1706+
new Subscription(
1707+
'profile_inline',
1708+
'projector',
1709+
RunMode::FromBeginning,
1710+
Status::Active,
1711+
lastSavedAt: new DateTimeImmutable('2021-01-01T00:00:00'),
1712+
),
1713+
],
1714+
$engine->subscriptions(),
1715+
);
1716+
1717+
$profileId = ProfileId::fromString('019d3991-e575-73b2-a18c-7da3fd7f5d70');
1718+
$profile = Profile::create($profileId, 'John');
1719+
$repository->save($profile);
1720+
1721+
$result = $engine->run();
1722+
1723+
self::assertEquals(1, $result->processedMessages);
1724+
self::assertEquals([], $result->errors);
1725+
1726+
self::assertEquals(
1727+
[
1728+
new Subscription(
1729+
'profile_inline',
1730+
'projector',
1731+
RunMode::FromBeginning,
1732+
Status::Active,
1733+
1,
1734+
lastSavedAt: new DateTimeImmutable('2021-01-01T00:00:00'),
1735+
),
1736+
],
1737+
$engine->subscriptions(),
1738+
);
1739+
1740+
$result = $this->connection->fetchAssociative(
1741+
'SELECT * FROM stateful_subscriber_state WHERE id = ?',
1742+
['profile_inline'],
1743+
);
1744+
1745+
self::assertIsArray($result);
1746+
self::assertArrayHasKey('id', $result);
1747+
self::assertEquals('profile_inline', $result['id']);
1748+
1749+
self::assertArrayHasKey('state', $result);
1750+
self::assertIsString($result['state']);
1751+
1752+
self::assertEquals(
1753+
['profiles' => ['019d3991-e575-73b2-a18c-7da3fd7f5d70' => 'John']],
1754+
json_decode($result['state'], true),
1755+
);
1756+
1757+
$projection = new ProfileInlineStatefulSubscriber(
1758+
$stateStore,
1759+
);
1760+
1761+
self::assertEquals(['019d3991-e575-73b2-a18c-7da3fd7f5d70' => 'John'], $projection->profiles);
1762+
}
1763+
16351764
/** @param list<Subscription> $subscriptions */
16361765
private static function findSubscription(array $subscriptions, string $id): Subscription
16371766
{

0 commit comments

Comments
 (0)