-
Notifications
You must be signed in to change notification settings - Fork 10
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.
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
Process Manager example with Symfony Messenger Command / Event Bus and ProophOS #53
Comments
In general a process manager subscribes to (Domain or Application) Events. Each event can fire one or more new command(s). A Hope this helps. Here is the current implementation: Process Manager <?php
namespace Acme\Host\Infrastructure\ProcessManager;
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
use Symfony\Component\Messenger\MessageBusInterface;
use Acme\Host\Application\Service\Host\SendVerificationEmail;
use Acme\Host\Domain\Model\Host\Event\HostRegistered;
final class RegistrationManager implements MessageSubscriberInterface
{
private MessageBusInterface $commandBus;
public function __construct(MessageBusInterface $commandBus)
{
$this->commandBus = $commandBus;
}
public function onHostRegistered(HostRegistered $event): void
{
$command = new SendVerificationEmail(
[
'hostId' => $event->hostId()->toString(),
'emailAddress' => $event->emailAddress()->toString(),
'token' => $event->verificationToken()->token(),
]
);
$this->commandBus->dispatch($command);
}
public static function getHandledMessages(): iterable
{
yield HostRegistered::class => [
'method' => 'onHostRegistered'
];
}
} Infrastructure The # config/services.yaml
services:
Acme\Host\Infrastructure\ProcessManager\RegistrationManager:
public: false
tags:
- { name: messenger.message_handler, bus: event.bus }
arguments:
- '@command.bus' # config/packages/messenger.yaml
framework:
messenger:
default_bus: command.bus
buses:
command.bus:
middleware:
- validation
event.bus:
default_middleware: allow_no_handlers In order to make this work with Prooph you have to wire the Messenger Event Bus with the Prooph Event Publisher. # config/packages/prooph_event_store_bus_bridge.yaml
services:
_defaults:
public: false
Prooph\EventStoreBusBridge\EventPublisher:
class: Acme\Common\Infrastructure\Prooph\EventPublisher
arguments:
- '@event.bus'
tags:
- { name: 'prooph_event_store.default.plugin' } <?php
namespace Acme\Common\Infrastructure\Prooph;
use Iterator;
use Prooph\Common\Event\ActionEvent;
use Prooph\EventStore\ActionEventEmitterEventStore;
use Prooph\EventStore\EventStore;
use Prooph\EventStore\Plugin\AbstractPlugin;
use Prooph\EventStore\TransactionalActionEventEmitterEventStore;
use Symfony\Component\Messenger\MessageBusInterface;
final class EventPublisher extends AbstractPlugin
{
private MessageBusInterface $eventBus;
/**
* @var Iterator[]
*/
private array $cachedEventStreams = [];
public function __construct(MessageBusInterface $eventBus)
{
$this->eventBus = $eventBus;
}
public function attachToEventStore(ActionEventEmitterEventStore $eventStore): void
{
$this->listenerHandlers[] = $eventStore->attach(
ActionEventEmitterEventStore::EVENT_APPEND_TO,
function (ActionEvent $event) use ($eventStore): void {
$recordedEvents = $event->getParam('streamEvents', new \ArrayIterator());
if (! $this->inTransaction($eventStore)) {
if ($event->getParam('streamNotFound', false)
|| $event->getParam('concurrencyException', false)
) {
return;
}
foreach ($recordedEvents as $recordedEvent) {
$this->eventBus->dispatch($recordedEvent);
}
} else {
$this->cachedEventStreams[] = $recordedEvents;
}
}
);
$this->listenerHandlers[] = $eventStore->attach(
ActionEventEmitterEventStore::EVENT_CREATE,
function (ActionEvent $event) use ($eventStore): void {
$stream = $event->getParam('stream');
$recordedEvents = $stream->streamEvents();
if (! $this->inTransaction($eventStore)) {
if ($event->getParam('streamExistsAlready', false)) {
return;
}
foreach ($recordedEvents as $recordedEvent) {
$this->eventBus->dispatch($recordedEvent);
}
} else {
$this->cachedEventStreams[] = $recordedEvents;
}
}
);
if ($eventStore instanceof TransactionalActionEventEmitterEventStore) {
$this->listenerHandlers[] = $eventStore->attach(
TransactionalActionEventEmitterEventStore::EVENT_COMMIT,
function (ActionEvent $event): void {
foreach ($this->cachedEventStreams as $stream) {
foreach ($stream as $recordedEvent) {
$this->eventBus->dispatch($recordedEvent);
}
}
$this->cachedEventStreams = [];
}
);
$this->listenerHandlers[] = $eventStore->attach(
TransactionalActionEventEmitterEventStore::EVENT_ROLLBACK,
function (ActionEvent $event): void {
$this->cachedEventStreams = [];
}
);
}
}
private function inTransaction(EventStore $eventStore): bool
{
return $eventStore instanceof TransactionalActionEventEmitterEventStore
&& $eventStore->inTransaction();
}
} Aggregate Root use Prooph\EventSourcing\AggregateChanged;
use Prooph\EventSourcing\AggregateRoot;
final class Host extends AggregateRoot
{
public static function register(
HostId $hostId,
EmailAddress $emailAddress,
EncodedPassword $encodedPassword,
DateTimeImmutable $registeredAt
): Host
{
$verificationToken = VerificationToken::generateWith($hostId, $emailAddress, $registeredAt);
$self = new self();
$self->recordThat(HostRegistered::with($hostId, $emailAddress, $encodedPassword, $verificationToken, $registeredAt));
return $self;
}
} Repository namespace Acme\Host\Infrastructure\Persistence\Pgsql;
use Prooph\EventSourcing\Aggregate\AggregateRepository;
use Acme\Host\Domain\Model\Host\Host;
use Acme\Host\Domain\Model\Host\HostId;
use Acme\Host\Domain\Model\Host\HostRepository;
final class HostEventStoreRepository extends AggregateRepository implements HostRepository
{
public function save(Host $host): void
{
$this->saveAggregateRoot($host);
}
public function get(HostId $id): ?Host
{
return $this->getAggregateRoot($id->toString());
}
} |
Does it help @iosifch? |
I didn't have time to look at it, but I want to do this as soon as possible! |
The Process Manager looks as I imagined. Practically, a Process Manager may group together all those events that are part of the same flow. I am right? |
Exactly. final class RegistrationManager implements MessageSubscriberInterface
{
private MessageBusInterface $commandBus;
public function __construct(MessageBusInterface $commandBus)
{
$this->commandBus = $commandBus;
}
public function onHostRegistered(HostRegistered $event): void
{
$command = new SendVerificationEmail(
[
'hostId' => $event->hostId()->toString(),
'emailAddress' => $event->emailAddress()->toString(),
'token' => $event->verificationToken()->token(),
]
);
$this->commandBus->dispatch($command);
}
public function onHostRegistered(HostRegistered $event): void
{
// Another step of the flow
$this->commandBus->dispatch($command);
}
public static function getHandledMessages(): iterable
{
yield HostRegistered::class => [
'method' => 'onHostRegistered'
];
yield HostEmailVerified::class => [
'method' => 'onHostEmailVerified'
];
}
} |
Came from:
/cc @iosifch
Example coming soon.
The text was updated successfully, but these errors were encountered: