README

🧰 提供了 Symfony Messenger 中间件，用于基本的按总线管道。它允许对消息处理程序创建的消息进行流畅的链式处理。例如，当处理程序（hdl1）处理消息（msg1）时，它创建后续消息（msg2），触发下一个处理程序（hdl2）的调用，该处理程序可能反过来又产生另一个新消息，此循环继续。

安装 📥

1. 使用 composer 安装

   composer require phphd/pipeline-bundle

2. 在 bundles.php 中启用该包

   PhPhD\PipelineBundle\PhdPipelineBundle::class => ['all' => true],

配置 ⚒️

要利用管道处理程序链为您的命令/查询总线，应将 phd_pipeline.forward_chain 中间件添加到列表中

framework:
    messenger:
        buses:
            command.bus:
                middleware:
                    - doctrine_transaction
+                   - phd_pipeline.forward_chain
                    - validation
            query.bus:
                middleware:
+                   - phd_pipeline.forward_chain
                    - validation

使用 🚀

考虑这个最初被派发到消息总线的 原始消息

final readonly class CreateVacationRequestCommandDto
{
    public function __construct(
        public int $userId,
        public int $vacationTypeId,
        #[Assert\DateTime]
        public string $startDate,
        #[Assert\DateTime]
        public string $endDate,
    ) {
    }
}

前置消息处理程序 返回一个新消息，该消息将被用于后续重新派发

#[AsMessageHandler(bus: 'command.bus')]
final readonly class ConvertVacationRequestCommandHandler
{
    public function __invoke(CreateVacationRequestCommandDto $dto): CreateVacationRequestCommand
    {
        $employee = $this->employeeRepository->find($dto->userId);
        $vacationType = $this->vacationTypeRepository->find($dto->vacationTypeId);

        $vacationPeriod = VacationPeriod::fromStringDates($dto->startDate, $dto->endDate);

        return new CreateVacationRequestCommand($employee, $vacationType, $vacationPeriod);
    }
}

新创建的消息传达基本上相同的企业概念，但比最初抽象层次更高。因此，它不是标量类型，而是业务对象（例如，VacationType 实体而不是 $vacationTypeId 标量）。基本上，新类不再仅代表 DTO。现在它体现了一个完整的领域对象。

您应该添加 #[NextForwarded] 属性以启用转发 此新消息到下一个处理程序

use PhPhD\Pipeline\NextForwarded;

#[NextForwarded]
final readonly class CreateVacationRequestCommand
{
    public function __construct(
        public Employee $employee,
        public VacationType $vacationType,
        public VacationPeriod $vacationPeriod,
    ) {
    }
}

缺少 #[NextForwarded] 属性的消息将不会被转发。此属性必须放在每个预期重新派发的消息上。

最后，一个 最终处理程序 必须实现核心业务逻辑。它可能或可能不向调用代码返回结果。

#[AsMessageHandler(bus: 'command.bus')]
final readonly class CreateVacationRequestHandler
{
    public function __invoke(CreateVacationRequestCommand $command)
    {
        // The core business logic that deals with domain entities rather than primitives...
    }
}

您可以通过返回原始消息类的实例（只要启用了转发属性）来按需链式连接任意数量的消息处理程序，甚至以递归方式，只要最终处理程序实现核心业务逻辑。

扩展转发

如果您不想在消息类上使用属性，也可以不使用。可能有一些情况，您想为 NextForwarded 实例应用一些动态配置。在这种情况下，您可以直接从处理程序方法返回 NextForwarded 类的实例。

#[AsMessageHandler(bus: 'command.bus')]
final readonly class ConvertVacationRequestCommandHandler
{
    /** @return NextForwarded<CreateVacationRequestCommand> */
    public function __invoke(CreateVacationRequestCommandDto $dto): NextForwarded
    {
        return new NextForwarded($this->createCommandFromDto($dto));
    }
}

上面的代码与之前显示的代码没有区别。