Skip to content

Iterable message dispatching #237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
frankdejonge wants to merge 1 commit into
base: version/4.x
Choose a base branch
from
Open

Iterable message dispatching #237

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/AggregateRootRepository.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ public function persist(object $aggregateRoot): void;
/**
* @throws UnableToPersistMessages
* @throws UnableToDispatchMessages
* @param iterable<object> $events
*/
public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, object ...$events): void;
public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, iterable $events): void;
}
7 changes: 5 additions & 2 deletions src/AntiCorruptionLayer/AntiCorruptionMessageDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

use EventSauce\EventSourcing\Message;
use EventSauce\EventSourcing\MessageDispatcher;
use function is_iterable;

class AntiCorruptionMessageDispatcher implements MessageDispatcher
{
Expand All @@ -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 = [];

@lgrossi lgrossi Mar 9, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To continue to leverage generators, maybe $forwarded could always be a generator?

Suggested change
$forwarded = [];
$messages = is_iterable($messages) ? $messages : [$messages];
$forwarded = (function () use ($messages): Generator {
foreach ($messages as $message) {
if (! $this->filterBefore->allows($message)) {
continue;
}
$message = $this->translator->translateMessage($message);
if (! $this->filterAfter->allows($message)) {
continue;
}
yield $message;
}
})();

/** @var iterable<Message> $messages */
$messages = is_iterable($messages) ? $messages : [$messages];

foreach ($messages as $message) {
if ( ! $this->filterBefore->allows($message)) {
Expand All @@ -42,6 +45,6 @@ public function dispatch(Message ...$messages): void
$forwarded[] = $message;
}

$this->dispatcher->dispatch(...$forwarded);
$this->dispatcher->dispatch($forwarded);
}
}
10 changes: 5 additions & 5 deletions src/AntiCorruptionLayer/AntiCorruptionMessageDispatcherTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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);
Expand All @@ -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);
Expand All @@ -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);
Expand All @@ -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);
Expand Down
2 changes: 1 addition & 1 deletion src/AntiCorruptionLayer/AntiCorruptionMessageRelay.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,6 @@ public function handle(Message $message): void
return;
}

$this->dispatcher->dispatch($message);
$this->dispatcher->dispatch([$message]);
}
}
4 changes: 3 additions & 1 deletion src/CollectingMessageDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Expand Down
8 changes: 4 additions & 4 deletions src/CollectingMessageDispatcherTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
[
Expand All @@ -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(
[
Expand Down
8 changes: 6 additions & 2 deletions src/DecoratingMessageDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@lgrossi lgrossi Mar 9, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you sure this works for iterable? I have the impression that this will just return an empty array to the map function, no?

Maybe something like:

Suggested change
array_map(fn (Message $message) => $this->decorator->decorate($message), (array) $messages)
$decoratedMessages = (function () use ($messages): Generator {
foreach ($messages as $message) {
yield $this->decorator->decorate($message);
}
})();
$this->dispatcher->dispatch($decoratedMessages);

@lgrossi lgrossi Mar 9, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems that this is also not covered by tests.

);
}
}
10 changes: 8 additions & 2 deletions src/EventDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,13 @@

interface EventDispatcher
{
public function dispatch(object ...$events): void;
/**
* @param iterable<object>|object $events
*/
public function dispatch(iterable|object $events): void;

public function dispatchWithHeaders(array $headers, object ...$events): void;
/**
* @param iterable<object>|object $events
*/
public function dispatchWithHeaders(array $headers, iterable|object $events): void;
}
13 changes: 9 additions & 4 deletions src/EventSourcedAggregateRootRepository.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<object> $events
*/
public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, iterable $events): void
{
$events = [...$events];

@lgrossi lgrossi Mar 9, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will unpack the iterable fully to memory, right? I wonder if there is a way to achieve this by using only generators. I couldn't think on a straightforward way to do so.


if (count($events) === 0) {
return;
}
Expand All @@ -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);
}
}
2 changes: 1 addition & 1 deletion src/InMemoryMessageRepository.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ public function purgeLastCommit(): void
$this->lastCommit = [];
}

public function persist(Message ...$messages): void
public function persist(iterable $messages): void
{
$this->lastCommit = [];

Expand Down
8 changes: 4 additions & 4 deletions src/InMemoryMessageRepositoryTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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')),
Expand All @@ -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);
Expand All @@ -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')),
Expand All @@ -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
Expand Down
2 changes: 1 addition & 1 deletion ...tionTests/RequiringHistoryWithAggregateRootConstruction/AggregateRootConstructionTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
2 changes: 1 addition & 1 deletion src/LibraryConsumptionTests/SynchronousDispatching/SynchronousMessageDispatcherTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Expand Down
3 changes: 2 additions & 1 deletion src/MessageDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ interface MessageDispatcher
{
/**
* @throws UnableToDispatchMessages
* @param iterable<Message>|Message $messages
*/
public function dispatch(Message ...$messages): void;
public function dispatch(iterable|Message $messages): void;
}
4 changes: 2 additions & 2 deletions src/MessageDispatcherChain.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
}
9 changes: 5 additions & 4 deletions src/MessageDispatchingEventDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}
Comment on lines 28 to 30

@lgrossi lgrossi Mar 9, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To keep leveraging generators, maybe this could pass messages as a generator instead?

Suggested change
foreach ($events as $event) {
$messages[] = $this->decorator->decorate(new Message($event, $headers));
}
$messages = (function() use ($events, $headers): Generator {
foreach ($events as $event) {
yield $this->decorator->decorate(new Message($event, $headers));
}
})();


$this->dispatcher->dispatch(...$messages);
$this->dispatcher->dispatch($messages);
}
}
2 changes: 1 addition & 1 deletion src/MessageDispatchingEventDispatcherTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
Expand Down
3 changes: 2 additions & 1 deletion src/MessageRepository.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,9 @@ interface MessageRepository
{
/**
* @throws UnableToPersistMessages
* @param iterable<Message> $messages
*/
public function persist(Message ...$messages): void;
public function persist(iterable $messages): void;

/**
* @return Generator<Message>
Expand Down
4 changes: 2 additions & 2 deletions src/ReplayingMessages/ReplayMessagesTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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')),
Expand All @@ -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,
Expand Down
4 changes: 2 additions & 2 deletions src/Snapshotting/ConstructingAggregateRootRepositoryWithSnapshotting.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
7 changes: 6 additions & 1 deletion src/SynchronousMessageDispatcher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

namespace EventSauce\EventSourcing;

use function is_iterable;

final class SynchronousMessageDispatcher implements MessageDispatcher
{
/**
Expand All @@ -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<Message> $messages */
$messages = is_iterable($messages) ? $messages : [$messages];

foreach ($messages as $message) {
foreach ($this->consumers as $consumer) {
$consumer->handle($message);
Expand Down
2 changes: 1 addition & 1 deletion src/TestUtilities/AggregateRootTestCase.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
2 changes: 1 addition & 1 deletion src/TestUtilities/AntiCorruptionLayerTestCase.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down
Loading