Symfony MessageHandler 计算消息已发送的次数

Question

我正在使用 Symfony Messenger，我想继续在处理程序中发送消息，直到它被发送多次。

我怎样才能跟踪它？

这是到目前为止我的处理程序类的代码：

class RetryTestHandler implements MessageHandlerInterface
{
    /**
    * @var EntityManagerInterface
    */
    private $entityManager;
    /**
     * @var MessageBusInterface
     */
    private $bus;

    public function __construct(MessageBusInterface $bus, EntityManagerInterface $entityManager)
    {
        $this->entityManager = $entityManager;
        $this->bus = $bus;
    }

    public function __invoke(RetryTest $message)
    {
        // TODO: Keep dispatching message until it has been dispatched 10 times?
        $this->bus->dispatch(new RetryTest("This is a test!"), [
            new DelayStamp(5000)
        ]);
    }
}

Answer 1

要将元数据添加到消息中，您可以使用 stamp。

您稍后可以在自己的自定义中间件中使用它。

例如对于这个自定义StampInterface实现类：

class LoopCount implements StampInterface {


    private int $count; 

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

    public function getCount(): int {
        return $this->count;
    } 
}

然后创建您自己的中间件来检查此标记并在处理后重新分派：

class ResendingMiddleware implements MiddlewareInterface
{
     private $bus;

     public function __construct(MessageBusInterface $bus) {
           $this->bus = $bus;
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {

        $envelope = $stack->next()->handle($envelope, $stack);

        if (null !== $stamp = $envelope->last(LoopCount::class)) {
            $count = $stamp->getCount();
        } else {
            return $envelope;
        }

        // Stop dispatching
        if ($count > 9) {
            return $envelope;
        }

        $this->bus->dispatch(new RetryTest("Dit is een test"), [
            new DelayStamp(5000),
            new LoopCount($count + 1)
        ]);

        return $envelope;
    }

如果处理次数超过 9 次，则消费该消息而不执行任何操作。

您还需要将中间件添加到配置中：

framework:
    messenger:
        buses:
            messenger.bus.default:
                middleware:
                    # service ids that implement Symfony\Component\Messenger\Middleware\MiddlewareInterface
                    - 'App\Middleware\ResendingMiddleware'

我匆忙写了这篇文章，现在无法测试它，但基础应该可以帮助你走向正确的方向。测试和调试，你就会让它工作。我稍后会再回来尝试看看是否有任何遗漏