在 PHP 中，在 Doctrine 实体中使用事件总线

你有多少次看到过这样写的代码呢？

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

我也看到过太多次了。通常这是一个服务或控制器操作。这种方法有什么问题？除了违反 SOLID 原则的一些规定外，它很难维护和测试。我们还引入了领域代码和基础设施细节（日志记录驱动程序、通知系统等）之间的耦合。

此外，如果有人将教师分配给科目，你无法保证所有依赖的底层进程都会被调用。

我们可以使用 发布-订阅 模式对其进行重构：

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

使用事件总线，我们可以将不相关的代码移动到订阅者中，这很好，但我们仍然无法保证其他程序员会遵循相同的方法。

我建议将事件移到更接近领域的位置，并在我们的业务模型中直接发布它们。让我们创建三个接口，它们将在我们的领域中得到重用：

• 一个接口用于定义有意义的业务事件：

namespace App\SharedKernel\Events;interface EventInterface
{
}

• 一个接口用于定义事件集合，可以在领域模型中添加事件：

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

实现可以简单地如下所示：

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 = [];
        }
    }
}

• 最后一个接口是用于领域模型指示其可以发布事件的接口：

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

现在我们可以开始重新设计我们的业务实体：

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

现在单元测试变得简单而有趣：

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

快速问题： 这个故事对你有任何价值吗？请通过留下一个 鼓掌 以表达感激之情来支持我的工作。谢谢。

现在我们需要一种在刷新期间在事件总线上发布事件的方法。在 Doctrine 中，由于它自己的事件系统，这相当容易。由于事件表示过去发生的事情，因此让我们在 Doctrine 事件的 postFlush 上发布事件（在其他框架中，可以通过类似的方式或引入在实体持久化后必须调用的服务来完成）。

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

我故意跳过了 EventBus 类，因为你可以使用任何你喜欢的实现。Symfony Messenger、Prooph 等等，甚至你自己的简单自定义实现。

我们的控制器现在看起来非常简洁：

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

我使用 Symfony Messenger 作为我选择的事件总线，它非常容易订阅已发布的事件：

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,
            ]
        ));
    }
}

此外，如果你选择将消息发布到异步队列系统（如 RabbitMq），你将保证最终处理它——例如，在通知传递期间出现电子邮件提供商服务问题的情况下。

我还建议阅读以下文章。有意义、精简的业务模型是你的应用程序的核心。

使用领域丰富的模型改进你的应用程序 📝

译自：https://medium.com/@dotcom.software/event-bus-inside-doctrine-entities-2a0fe339c425