Process Manager example with Symfony Messenger Command / Event Bus and ProophOS

[webdevilopers]

Came from:

do you have any example of the process manager pattern in PHP?

• https://twitter.com/iosifch/status/1329190494153347078

/cc @iosifch

Example coming soon.

[webdevilopers]

In general a process manager subscribes to (Domain or Application) Events. Each event can fire one or more new command(s). The manager can decide which command to fire based on the provided event. But there should not be added deeper business logic to make these decisions.

A Saga AFAIK is a process manager that persists the state of the process - also known as state machine. This can help to log the steps of the process and view the current state over a longer period of time.

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 MessageSubscriberInterface is part of the Symfony Messenger in order to subscribe to Events (Messages) from the event bus.

# 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. In earlier versions Prooph offered a Service Bus and Event Publisher. But the package was deprecated.

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

[webdevilopers]

Does it help @iosifch?

[iosifch]

I didn't have time to look at it, but I want to do this as soon as possible!

[iosifch]

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?

[webdevilopers]

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