Process Manager example with Symfony Messenger Command / Event Bus and ProophOS about php-ddd HOT 5 OPEN

<?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'
        ];
    }
}

<?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();
    }
}

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;
    }
}

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());
    }
}

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'
        ];
    }
}