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

要将元数据添加到您的消息中，you can use stamps。

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

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

class LoopCount implements StampInterface {


    private int $count; 

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

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

然后create your own middleware检查这个邮票并在处理后重新发送：

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'

我写的很匆忙，目前无法测试，但是基础应该可以帮助您朝着正确的方向前进。测试和调试，你会得到它的工作。稍后我会回来尝试看看是否缺少任何东西