Event Bus inside Doctrine Entities in PHP

Solution for a custom event system inside business entities — no hacking, pure object-oriented design. A case study of an imaginary Learning Management System.

Photo by Mia Baker on Unsplash

How often have you seen code written this way?

public function assignTeacherAction(
    SubjectInterface $subject,
    TeacherInterface $teacher
): JsonResponse
{
    $subject->assignTeacher($teacher);
    $this->entityManager->flush();
    $this->logTeacherAssignedToSubject($subject, $teacher);
    $this->notifyTeacher($subject, $teacher);
    $this->notifyStudents($subject, $teacher);
    $this->notifyAccounting($subject, $teacher);
    // … list goes on
    return new JsonResponse();
}

I’ve seen this too many times. Usually it is either a service or a controller action. What’s wrong with this approach? Beside violating some letters from the SOLID principles, it’s hard to maintain and test. We’re also introducing coupling between domain code and infrastructure details (logging drivers, notification systems, etc.).

Also you don’t have guarantee if someone assigns a teacher to the subject, all the underlying processes dependent will be invoked.

We could refactor this using the pub-sub pattern:

public function assignTeacherAction(
    SubjectInterface $subject,
    TeacherInterface $teacher
): JsonResponse
{
    $subject->assignTeacher($teacher);
    $this->entityManager->flush();
    $this->eventBus->publish(
        TeacherAssignedEvent::create($subject, $teacher)
    ));
    return new JsonResponse();
}

Using an Event Bus we can move unrelated code to the subscribers, which is nice, but still we have no guarantee that other programmers will follow the same approach.

I’d propose moving events closer to the domain and publishing them directly inside our business models. Let’s create three interfaces that will be re-used across our domain:

• An interface to define a meaningful business event:

namespace App\SharedKernel\Events;
interface EventInterface
{
}

• An interface to define a collection of events to which events can be added inside domain models:

namespace App\SharedKernel\Events;
interface EventCollectionInterface
{
    public function record(EventInterface $event): void;
    /** @return iterable<EventInterface> */
    public function popEvents(): iterable;
}

The implementation can be as simple as:

namespace App\SharedKernel\Events;
final class ArrayEventCollection implements EventCollectionInterface
{
    /** @var EventInterface[] */
    private array $events = [];
    public function publish(EventInterface $event): void
    {
        $this->events[] = $event;
    }
    /** {@inheritDoc} */
    public function popEvents(): iterable
    {
        try {
            return $this->events;
        } finally {
            $this->events = [];
        }
    }
}

• Finally an interface for the domain models to indicate that it can have events to publish:

namespace App\SharedKernel\Events;
interface EventAwareInterface
{
    /** @return iterable<EventInterface> */
    public function popEvents(): iterable;
}

Now we can begin on re-working our business entity:

class Subject implements SubjectInterface, EventAwareInterface
{
    // other class properties and methods skipped for brevity
    private EventCollectionInterface $events;
    public function __construct()
    {
        $this->events = new ArrayEventCollection();
    }
    public function assignTeacher(TeacherInterface $teacher): void
    {
        $this->teachers[] = $teacher;
        $this->events->record(
            TeacherAssignedEvent::create($subject, $teacher)
        ));
    }
}

Unit tests are now simple and fun:

class SubjectTest extends TestCase
{
    public function testAssignTeacher(): void
    {
        $teacher = new Teacher();
        
        $this->subject->assignTeacher($teacher);
        $this->assertEventRecorded(
            $this->subject,
            TeacherAssignedEvent::class
        );
    }
    /**
     * $param class-string<EventInterface> $expectedEvent
     */
    private function assertEventRecorded(
        EventAwareInterface $eventAware,
        string $expectedEvent
    ): void {
        $recordedEvents = [];
        foreach ($eventAware->popEvents() as $event) {
            $recordedEvents[] = \get_class($event);
        }
        self::assertContains($expectedEvent, $recordedEvents);
    }
}

Quick question: is this story of any value to you? Please support my work by leaving a clap as a token of appreciation. Thank you.

Now we need a way to publish events on the event bus during the flush. In Doctrine it’s quite easy thanks to it’s own event system. Since an event indicates something that had happened in the past, let’s publish events on the postFlush Doctrine event (in other framework this can be accomplished by doing something similar or introducing a service that has to be called after the entity was persisted).

final class AggregatePersistEventDispatcher
{
    private EventBus $eventBus;

    /** @var array<int, EventAwareInterface> */
    private array $events = [];

    public function __construct(EventBus $eventBus)
    {
        $this->eventBus = $eventBus;
    }

    public function onFlush(OnFlushEventArgs $event): void
    {
        $unitOfWork = $event->getEntityManager()->getUnitOfWork();

        $this->gather($unitOfWork->getScheduledEntityUpdates());
        $this->gather($unitOfWork->getScheduledEntityInsertions());
        $this->gather($unitOfWork->getScheduledEntityDeletions());
    }

    public function postFlush(): void
    {
        foreach ($this->events as $idx => $event) {
            unset($this->events[$idx]);
            $this->eventBus->dispatch($event);
        }
    }
    /** @param object[] $entities */
    private function gather(array $entities): void
    {
        foreach ($entities as $entity) {
            if ($entity instanceof EventAwareInterface) {
                foreach ($entity->popEvents() as $event) {
                    $this->events[] = $event;
                }
            }
        }
    }
}

I deliberately skipped the EventBus class as you could use any implementation you wish. Symfony Messenger, Prooph to name the few, or even your simple custom implementation.

Our controller now looks extremely lean:

public function assignTeacherAction(
    SubjectInterface $subject,
    TeacherInterface $teacher
): JsonResponse
{
    $subject->assignTeacher($teacher);
    $this->entityManager->flush();
    return new JsonResponse();
}

I’m using the Symfony Messenger as the event bus of my choice, it’s extremely easy to subscribe to the published event:

class NotifyTeacherAssignedToSubjectSubscriber
{
    private TeacherRepository $teachers;
    private SubjectRepository $subjects;
    private MailerInterface $mailer;
    public function __invoke(TeacherAssignedEvent $event): void
    {
        $teacher = $this->teachers->byId($event->getTeacherId());
        $subject = $this->subjects->byId($event->getSubjectId());
        $this->mailer->send(new TemplatedEmail(
            'Email/Subject/teacher-assigned-notification.html.twig',
            [
                'teacher' => $teacher,
                'subject' => $subject,
            ]
        ));
    }
}

Also if you choose to publish the message to an asynchronous queue system (like RabbitMq), you will be guaranteed to eventually process it — eg. in case there’s an issue with the e-mail provider service during the notification delivery.

Have fun coding. Any questions? Hit me up in the comments.

Photo by Kenny Eliason on Unsplash

I’d also recommend reading the following story. Meaningful, lean business models are the heart of your application

Improve your application with domain-rich models 📝