From 6f0d2ef44b17f65b8733240defbe96ae2fc0eae5 Mon Sep 17 00:00:00 2001 From: Frank de Jonge Date: Sat, 8 Mar 2025 17:06:12 +0100 Subject: [PATCH] Iterable message dispatching --- src/AggregateRootRepository.php | 3 ++- .../AntiCorruptionMessageDispatcher.php | 7 +++++-- .../AntiCorruptionMessageDispatcherTest.php | 10 +++++----- .../AntiCorruptionMessageRelay.php | 2 +- src/CollectingMessageDispatcher.php | 4 +++- src/CollectingMessageDispatcherTest.php | 8 ++++---- src/DecoratingMessageDispatcher.php | 8 ++++++-- src/EventDispatcher.php | 10 ++++++++-- src/EventSourcedAggregateRootRepository.php | 13 +++++++++---- src/InMemoryMessageRepository.php | 2 +- src/InMemoryMessageRepositoryTest.php | 8 ++++---- .../AggregateRootConstructionTest.php | 2 +- .../SynchronousMessageDispatcherTest.php | 2 +- src/MessageDispatcher.php | 3 ++- src/MessageDispatcherChain.php | 4 ++-- src/MessageDispatchingEventDispatcher.php | 9 +++++---- src/MessageDispatchingEventDispatcherTest.php | 2 +- src/MessageRepository.php | 3 ++- src/ReplayingMessages/ReplayMessagesTest.php | 4 ++-- ...ctingAggregateRootRepositoryWithSnapshotting.php | 4 ++-- src/SynchronousMessageDispatcher.php | 7 ++++++- src/TestUtilities/AggregateRootTestCase.php | 2 +- src/TestUtilities/AntiCorruptionLayerTestCase.php | 2 +- src/TestUtilities/EventStager.php | 2 +- 24 files changed, 75 insertions(+), 46 deletions(-) diff --git a/src/AggregateRootRepository.php b/src/AggregateRootRepository.php index f69d9d0f..8a316883 100644 --- a/src/AggregateRootRepository.php +++ b/src/AggregateRootRepository.php @@ -28,6 +28,7 @@ public function persist(object $aggregateRoot): void; /** * @throws UnableToPersistMessages * @throws UnableToDispatchMessages + * @param iterable $events */ - public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, object ...$events): void; + public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, iterable $events): void; } diff --git a/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcher.php b/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcher.php index 5ecaeabb..8e3e3505 100644 --- a/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcher.php +++ b/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcher.php @@ -6,6 +6,7 @@ use EventSauce\EventSourcing\Message; use EventSauce\EventSourcing\MessageDispatcher; +use function is_iterable; class AntiCorruptionMessageDispatcher implements MessageDispatcher { @@ -24,9 +25,11 @@ public function __construct( $this->filterAfter = $filterAfter ?? new AllowAllMessages(); } - public function dispatch(Message ...$messages): void + public function dispatch(iterable|Message $messages): void { $forwarded = []; + /** @var iterable $messages */ + $messages = is_iterable($messages) ? $messages : [$messages]; foreach ($messages as $message) { if ( ! $this->filterBefore->allows($message)) { @@ -42,6 +45,6 @@ public function dispatch(Message ...$messages): void $forwarded[] = $message; } - $this->dispatcher->dispatch(...$forwarded); + $this->dispatcher->dispatch($forwarded); } } diff --git a/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcherTest.php b/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcherTest.php index 5406ed20..dfd3d061 100644 --- a/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcherTest.php +++ b/src/AntiCorruptionLayer/AntiCorruptionMessageDispatcherTest.php @@ -36,7 +36,7 @@ public function when_no_filters_and_transformation_happens_messages_are_always_r $dispatcher = $this->messageDispatcher(); $messages = array_map(fn (object $o) => new Message($o), $incoming); - $dispatcher->dispatch(...$messages); + $dispatcher->dispatch($messages); $dispatchedEvents = $this->dispatchedPayloads(); $this->assertEquals($expected, $dispatchedEvents); @@ -62,7 +62,7 @@ public function no_transformation_filter_out_excluded_payloads_before_transforma $dispatcher = $this->messageDispatcher(); $messages = array_map(fn (object $o) => new Message($o), $incoming); - $dispatcher->dispatch(...$messages); + $dispatcher->dispatch($messages); $dispatchedEvents = $this->dispatchedPayloads(); $this->assertEquals($expected, $dispatchedEvents); @@ -80,7 +80,7 @@ public function no_transformation_filter_out_excluded_payloads_after_transformat $dispatcher = $this->messageDispatcher(); $messages = array_map(fn (object $o) => new Message($o), $incoming); - $dispatcher->dispatch(...$messages); + $dispatcher->dispatch($messages); $dispatchedEvents = $this->dispatchedPayloads(); $this->assertEquals($expected, $dispatchedEvents); @@ -105,7 +105,7 @@ public function no_transformation_filter_all_but_public_and_private_payloads_aft $dispatcher = $this->messageDispatcher(); $messages = array_map(fn (object $o) => new Message($o), $incoming); - $dispatcher->dispatch(...$messages); + $dispatcher->dispatch($messages); $dispatchedEvents = $this->dispatchedPayloads(); $this->assertEquals($expected, $dispatchedEvents); @@ -132,7 +132,7 @@ public function transformation_private_to_public( $dispatcher = $this->messageDispatcher(); $messages = array_map(fn (object $o) => new Message($o), $incoming); - $dispatcher->dispatch(...$messages); + $dispatcher->dispatch($messages); $dispatchedEvents = $this->dispatchedPayloads(); $this->assertEquals($expected, $dispatchedEvents); diff --git a/src/AntiCorruptionLayer/AntiCorruptionMessageRelay.php b/src/AntiCorruptionLayer/AntiCorruptionMessageRelay.php index 43b257fc..502c3df8 100644 --- a/src/AntiCorruptionLayer/AntiCorruptionMessageRelay.php +++ b/src/AntiCorruptionLayer/AntiCorruptionMessageRelay.php @@ -35,6 +35,6 @@ public function handle(Message $message): void return; } - $this->dispatcher->dispatch($message); + $this->dispatcher->dispatch([$message]); } } diff --git a/src/CollectingMessageDispatcher.php b/src/CollectingMessageDispatcher.php index b7e647c2..a3f76da7 100644 --- a/src/CollectingMessageDispatcher.php +++ b/src/CollectingMessageDispatcher.php @@ -14,8 +14,10 @@ class CollectingMessageDispatcher implements MessageDispatcher */ private array $collectedMessages = []; - public function dispatch(Message ...$messages): void + public function dispatch(iterable|Message $messages): void { + $messages = is_iterable($messages) ? $messages : [$messages]; + array_push($this->collectedMessages, ...$messages); } diff --git a/src/CollectingMessageDispatcherTest.php b/src/CollectingMessageDispatcherTest.php index 00cca409..e7e159e1 100644 --- a/src/CollectingMessageDispatcherTest.php +++ b/src/CollectingMessageDispatcherTest.php @@ -15,11 +15,11 @@ public function collecting_messages(): void { $dispatcher = new CollectingMessageDispatcher(); - $dispatcher->dispatch( + $dispatcher->dispatch([ new Message(new EventStub('what')), new Message(new EventStub('is')), new Message(new EventStub('up')), - ); + ]); $this->assertEquals( [ @@ -38,11 +38,11 @@ public function exposing_collected_payloads(): void { $dispatcher = new CollectingMessageDispatcher(); - $dispatcher->dispatch( + $dispatcher->dispatch([ new Message(new EventStub('what')), new Message(new EventStub('is')), new Message(new EventStub('up')), - ); + ]); $this->assertEquals( [ diff --git a/src/DecoratingMessageDispatcher.php b/src/DecoratingMessageDispatcher.php index bc07b9f0..cd38b78b 100644 --- a/src/DecoratingMessageDispatcher.php +++ b/src/DecoratingMessageDispatcher.php @@ -4,16 +4,20 @@ namespace EventSauce\EventSourcing; +use function is_iterable; + class DecoratingMessageDispatcher implements MessageDispatcher { public function __construct(private MessageDispatcher $dispatcher, private MessageDecorator $decorator) { } - public function dispatch(Message ...$messages): void + public function dispatch(iterable|Message $messages): void { + $messages = is_iterable($messages) ? $messages : [$messages]; + $this->dispatcher->dispatch( - ...array_map(fn (Message $message) => $this->decorator->decorate($message), $messages) + array_map(fn (Message $message) => $this->decorator->decorate($message), (array) $messages) ); } } diff --git a/src/EventDispatcher.php b/src/EventDispatcher.php index 5a978847..81d52a41 100644 --- a/src/EventDispatcher.php +++ b/src/EventDispatcher.php @@ -6,7 +6,13 @@ interface EventDispatcher { - public function dispatch(object ...$events): void; + /** + * @param iterable|object $events + */ + public function dispatch(iterable|object $events): void; - public function dispatchWithHeaders(array $headers, object ...$events): void; + /** + * @param iterable|object $events + */ + public function dispatchWithHeaders(array $headers, iterable|object $events): void; } diff --git a/src/EventSourcedAggregateRootRepository.php b/src/EventSourcedAggregateRootRepository.php index 8d50732f..04d8107f 100644 --- a/src/EventSourcedAggregateRootRepository.php +++ b/src/EventSourcedAggregateRootRepository.php @@ -77,12 +77,17 @@ public function persist(object $aggregateRoot): void $this->persistEvents( $aggregateRoot->aggregateRootId(), $aggregateRoot->aggregateRootVersion(), - ...$aggregateRoot->releaseEvents() + $aggregateRoot->releaseEvents() ); } - public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, object ...$events): void + /** + * @param iterable $events + */ + public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, iterable $events): void { + $events = [...$events]; + if (count($events) === 0) { return; } @@ -102,7 +107,7 @@ public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRo )); }, $events); - $this->messages->persist(...$messages); - $this->dispatcher->dispatch(...$messages); + $this->messages->persist($messages); + $this->dispatcher->dispatch($messages); } } diff --git a/src/InMemoryMessageRepository.php b/src/InMemoryMessageRepository.php index ffd9187f..1b565fa2 100644 --- a/src/InMemoryMessageRepository.php +++ b/src/InMemoryMessageRepository.php @@ -33,7 +33,7 @@ public function purgeLastCommit(): void $this->lastCommit = []; } - public function persist(Message ...$messages): void + public function persist(iterable $messages): void { $this->lastCommit = []; diff --git a/src/InMemoryMessageRepositoryTest.php b/src/InMemoryMessageRepositoryTest.php index 4c214352..54355e5e 100644 --- a/src/InMemoryMessageRepositoryTest.php +++ b/src/InMemoryMessageRepositoryTest.php @@ -17,7 +17,7 @@ public function getting_the_first_page_for_pagination(): void { $repository = new InMemoryMessageRepository(); - $repository->persist( + $repository->persist([ new Message(new EventStub('1')), new Message(new EventStub('2')), new Message(new EventStub('3')), @@ -28,7 +28,7 @@ public function getting_the_first_page_for_pagination(): void new Message(new EventStub('8')), new Message(new EventStub('9')), new Message(new EventStub('10')), - ); + ]); $messages = $repository->paginate(OffsetCursor::fromStart(limit: 5)); $actualMessages = iterator_to_array($messages, false); @@ -52,7 +52,7 @@ public function getting_a_subsequent_page(): void { $repository = new InMemoryMessageRepository(); - $repository->persist( + $repository->persist([ new Message(new EventStub('1')), new Message(new EventStub('2')), new Message(new EventStub('3')), @@ -63,7 +63,7 @@ public function getting_a_subsequent_page(): void new Message(new EventStub('8')), new Message(new EventStub('9')), new Message(new EventStub('10')), - ); + ]); $messages = $repository->paginate(OffsetCursor::fromStart(limit: 5)); // consume messages to ensure return value is generated diff --git a/src/LibraryConsumptionTests/RequiringHistoryWithAggregateRootConstruction/AggregateRootConstructionTest.php b/src/LibraryConsumptionTests/RequiringHistoryWithAggregateRootConstruction/AggregateRootConstructionTest.php index d774d74f..a8ec7cb2 100644 --- a/src/LibraryConsumptionTests/RequiringHistoryWithAggregateRootConstruction/AggregateRootConstructionTest.php +++ b/src/LibraryConsumptionTests/RequiringHistoryWithAggregateRootConstruction/AggregateRootConstructionTest.php @@ -37,7 +37,7 @@ public function expecting_an_aggregate_when_there_is_history(): void new InMemoryMessageRepository() ); $id = DummyAggregateRootId::fromString('nope'); - $repository->persistEvents($id, 1, new DummyInternalEvent()); + $repository->persistEvents($id, 1, [new DummyInternalEvent()]); $aggregateRoot = $repository->retrieve($id); $this->assertInstanceOf(AggregateThatRequiredHistoryForReconstitutionStub::class, $aggregateRoot); } diff --git a/src/LibraryConsumptionTests/SynchronousDispatching/SynchronousMessageDispatcherTest.php b/src/LibraryConsumptionTests/SynchronousDispatching/SynchronousMessageDispatcherTest.php index cefe3451..d00dd5af 100644 --- a/src/LibraryConsumptionTests/SynchronousDispatching/SynchronousMessageDispatcherTest.php +++ b/src/LibraryConsumptionTests/SynchronousDispatching/SynchronousMessageDispatcherTest.php @@ -20,7 +20,7 @@ public function dispatching_messages_synchronously(): void $stubConsumer = new SynchronousMessageConsumerStub(); $syncDispatcher = new SynchronousMessageDispatcher($stubConsumer, $stubConsumer); $message = new Message(new EventStub('value')); - $syncDispatcher->dispatch($message, $message); + $syncDispatcher->dispatch([$message, $message]); $this->assertEquals([$message, $message, $message, $message], $stubConsumer->handled); } diff --git a/src/MessageDispatcher.php b/src/MessageDispatcher.php index 1b36c04a..6f5042da 100644 --- a/src/MessageDispatcher.php +++ b/src/MessageDispatcher.php @@ -8,6 +8,7 @@ interface MessageDispatcher { /** * @throws UnableToDispatchMessages + * @param iterable|Message $messages */ - public function dispatch(Message ...$messages): void; + public function dispatch(iterable|Message $messages): void; } diff --git a/src/MessageDispatcherChain.php b/src/MessageDispatcherChain.php index 517e374a..beab571b 100644 --- a/src/MessageDispatcherChain.php +++ b/src/MessageDispatcherChain.php @@ -16,10 +16,10 @@ public function __construct(MessageDispatcher ...$dispatchers) $this->dispatchers = $dispatchers; } - public function dispatch(Message ...$messages): void + public function dispatch(iterable|Message $messages): void { foreach ($this->dispatchers as $dispatcher) { - $dispatcher->dispatch(...$messages); + $dispatcher->dispatch($messages); } } } diff --git a/src/MessageDispatchingEventDispatcher.php b/src/MessageDispatchingEventDispatcher.php index 4bad584d..30b2f7a1 100644 --- a/src/MessageDispatchingEventDispatcher.php +++ b/src/MessageDispatchingEventDispatcher.php @@ -15,19 +15,20 @@ public function __construct(MessageDispatcher $dispatcher, ?MessageDecorator $de $this->decorator = $decorator ?: new DefaultHeadersDecorator(); } - public function dispatch(object ...$events): void + public function dispatch(iterable|object $events): void { - $this->dispatchWithHeaders([], ...$events); + $this->dispatchWithHeaders([], $events); } - public function dispatchWithHeaders(array $headers, object ...$events): void + public function dispatchWithHeaders(array $headers, iterable|object $events): void { + $events = is_iterable($events) ? $events : [$events]; $messages = []; foreach ($events as $event) { $messages[] = $this->decorator->decorate(new Message($event, $headers)); } - $this->dispatcher->dispatch(...$messages); + $this->dispatcher->dispatch($messages); } } diff --git a/src/MessageDispatchingEventDispatcherTest.php b/src/MessageDispatchingEventDispatcherTest.php index db935728..75380e2a 100644 --- a/src/MessageDispatchingEventDispatcherTest.php +++ b/src/MessageDispatchingEventDispatcherTest.php @@ -16,7 +16,7 @@ public function dispatching_messages_plainly(): void $subdispatcher = new CollectingMessageDispatcher(); $eventDispatcher = new MessageDispatchingEventDispatcher($subdispatcher); $event = new EventStub('value'); - $eventDispatcher->dispatch($event); + $eventDispatcher->dispatch([$event]); $collectedMessages = $subdispatcher->collectedMessages(); $this->assertEquals($event, $collectedMessages[0]->payload()); } diff --git a/src/MessageRepository.php b/src/MessageRepository.php index 6ce68878..79729a09 100644 --- a/src/MessageRepository.php +++ b/src/MessageRepository.php @@ -10,8 +10,9 @@ interface MessageRepository { /** * @throws UnableToPersistMessages + * @param iterable $messages */ - public function persist(Message ...$messages): void; + public function persist(iterable $messages): void; /** * @return Generator diff --git a/src/ReplayingMessages/ReplayMessagesTest.php b/src/ReplayingMessages/ReplayMessagesTest.php index cd904812..2fa8a730 100644 --- a/src/ReplayingMessages/ReplayMessagesTest.php +++ b/src/ReplayingMessages/ReplayMessagesTest.php @@ -18,7 +18,7 @@ class ReplayMessagesTest extends TestCase public function replaying_multiple_pages(): void { $messageRepository = new InMemoryMessageRepository(); - $messageRepository->persist( + $messageRepository->persist([ new Message(new EventStub('1')), new Message(new EventStub('2')), new Message(new EventStub('3')), @@ -29,7 +29,7 @@ public function replaying_multiple_pages(): void new Message(new EventStub('8')), new Message(new EventStub('9')), new Message(new EventStub('10')), - ); + ]); $replayer = new ReplayMessages( $messageRepository, diff --git a/src/Snapshotting/ConstructingAggregateRootRepositoryWithSnapshotting.php b/src/Snapshotting/ConstructingAggregateRootRepositoryWithSnapshotting.php index 16e15d3f..ae051040 100644 --- a/src/Snapshotting/ConstructingAggregateRootRepositoryWithSnapshotting.php +++ b/src/Snapshotting/ConstructingAggregateRootRepositoryWithSnapshotting.php @@ -73,8 +73,8 @@ public function persist(object $aggregateRoot): void $this->regularRepository->persist($aggregateRoot); } - public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, object ...$events): void + public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, iterable $events): void { - $this->regularRepository->persistEvents($aggregateRootId, $aggregateRootVersion, ...$events); + $this->regularRepository->persistEvents($aggregateRootId, $aggregateRootVersion, $events); } } diff --git a/src/SynchronousMessageDispatcher.php b/src/SynchronousMessageDispatcher.php index 644ccadf..a6e8ee32 100644 --- a/src/SynchronousMessageDispatcher.php +++ b/src/SynchronousMessageDispatcher.php @@ -4,6 +4,8 @@ namespace EventSauce\EventSourcing; +use function is_iterable; + final class SynchronousMessageDispatcher implements MessageDispatcher { /** @@ -16,8 +18,11 @@ public function __construct(MessageConsumer ...$consumers) $this->consumers = $consumers; } - public function dispatch(Message ...$messages): void + public function dispatch(iterable|Message $messages): void { + /** @var iterable $messages */ + $messages = is_iterable($messages) ? $messages : [$messages]; + foreach ($messages as $message) { foreach ($this->consumers as $consumer) { $consumer->handle($message); diff --git a/src/TestUtilities/AggregateRootTestCase.php b/src/TestUtilities/AggregateRootTestCase.php index 81248978..7cadc4a7 100644 --- a/src/TestUtilities/AggregateRootTestCase.php +++ b/src/TestUtilities/AggregateRootTestCase.php @@ -159,7 +159,7 @@ abstract protected function aggregateRootClassName(): string; */ public function given(object ...$events) { - $this->repository->persistEvents($this->aggregateRootId(), count($events), ...$events); + $this->repository->persistEvents($this->aggregateRootId(), count($events), $events); $this->messageRepository->purgeLastCommit(); return $this; diff --git a/src/TestUtilities/AntiCorruptionLayerTestCase.php b/src/TestUtilities/AntiCorruptionLayerTestCase.php index 4348acb9..0f526a36 100644 --- a/src/TestUtilities/AntiCorruptionLayerTestCase.php +++ b/src/TestUtilities/AntiCorruptionLayerTestCase.php @@ -81,7 +81,7 @@ protected function expectMessages(Message ...$messages): void $dispatcher = new SynchronousMessageDispatcher($aclConsumer); $aclDispatcher = ($this->dispatcher)($dispatcher); - $aclDispatcher->dispatch(...$this->messagesToDispatch); + $aclDispatcher->dispatch($this->messagesToDispatch); $messagesDispatched = $consumer->collectedMessages(); $this->assertCount(count($messages), $messagesDispatched); diff --git a/src/TestUtilities/EventStager.php b/src/TestUtilities/EventStager.php index 0ea51def..53bfddbb 100644 --- a/src/TestUtilities/EventStager.php +++ b/src/TestUtilities/EventStager.php @@ -24,7 +24,7 @@ public function __construct( public function stage(object ...$events): AggregateRootTestCase { - $this->repository->persistEvents($this->id, count($events), ...$events); + $this->repository->persistEvents($this->id, count($events), $events); $this->messageRepository->purgeLastCommit(); return $this->testCase;