[Symfony] sendCommand / sendEvent for delayed message have different behaviour

[amenophis]

Hi,
we are trying to use RabbitMqDlxDelayStrategy with the rabbitmq_amqp transport in a symfony project. Our config looks like that:

enqueue:
    transport:
        default:
            alias: rabbitmq_amqp
        rabbitmq_amqp:
            driver: bunny
            host: '%env(RABBITMQ_HOST)%'
            user: '%env(RABBITMQ_USER)%'
            pass: '%env(RABBITMQ_PASSWORD)%'
            delay_strategy: dlx
    client:
        prefix: company
        app_name: webhook
        router_queue: router # Used id queue name (${prefix}.{app_name}.{router_queue})
        router_topic: router # Used id exchange name (${prefix}.{router_topic})

With sendEvent, the delayed message is pushed in enqueue.company.webhook.job_execute.120000.delayed for 120s, and then delivered to the company.webhook.job_execute queue.

With sendCommand, the delayed message is pushed in enqueue.company.webhook.router.120000.delayed for 120s. After the 120s it is sent to enqueue.company.webhook.job_execute.120000.delayed for 120s, and then delivered to the company.webhook.job_execute queue.

In both cases, we use the autowired ProducerInterface to sendEvent/sendCommand.

We are waiting for the same behaviour with Event or Command (The actual Event behaviour)

Is it a normal behaviour ? Is it a bug ?
Thanks

[makasim]

@amenophis the message is being delayed twice right? I think that a router should reset any priority, delaying, and expiration while sending the message to its final destination.

[amenophis]

@makasim Yes, the message is delayed first for 120s in the router.delayed queue, and then a second time for 120s in the job_execute.delayed queue.

This behaviour appear only while sending command, not event.

[amenophis]

I don't understand why the router.delayed queue is created, what's it's purpose ?

[makasim]

rabbitmq does not support message delaying out of the box. We have to two options: a workaround based on dead letter queue and time to live features and the plugin.

You are using a kind of "work around" which needs a queue to be created

[amenophis]

I understand how the delayed feature work, but i don't understand why 2 queues are needed with Command and just 1 queue with Event

[makasim]

Could you post a code where you register an event and command processors?

I guess it should the other way around.

[amenophis]

@makasim You can see code below.

This is the Processors

<?php

namespace AppBundle\Async\Processor;

use AppBundle\Entity\WebhookJob;
use AppBundle\WebhookJobRunner;
use Enqueue\Client\CommandSubscriberInterface;
use Enqueue\Client\TopicSubscriberInterface;
use Interop\Queue\PsrContext;
use Interop\Queue\PsrMessage;
use Interop\Queue\PsrProcessor;

class WebhookExecuteProcessor implements PsrProcessor, TopicSubscriberInterface
{
    public static function getSubscribedTopics()
    {
        return [
            'webhook.job_execute' => ['queueName' => 'job_execute']
        ];
    }

    protected $runner;

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

    public function process(PsrMessage $message, PsrContext $context)
    {
        $this->runner->run($message->getBody());

        return self::ACK;
    }
}

class WebhookExecuteProcessor2 implements PsrProcessor, CommandSubscriberInterface
{
    public static function getSubscribedCommand()
    {
        return [
            'processorName' => 'webhook.job_execute',
            'queueName' => 'job_execute',
        ];
    }

    protected $runner;

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

    public function process(PsrMessage $message, PsrContext $context)
    {
        $this->runner->run($message->getBody());

        return self::ACK;
    }
}

And this is the runner used in the Processors

<?php

namespace AppBundle;

use Enqueue\Client\Message;
use Enqueue\Client\ProducerInterface;

class WebhookJobRunner
{
    protected $producer;

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

    public function run(string $jobId): void
    {
        // ...
        $message = new Message($jobId);
        $message->setDelay(120);
        
        $this->producer->sendEvent('webhook_job.execute', $message);
        $this->producer->sendCommand('webhook_job.execute', $message);
    }
}

[amenophis]

@makasim any updates on this issue ?

[amenophis]

Thanks @makasim. When do you plan the next release ?

[makasim]

I was almost ready to release a stable version (3-5 days left). Now I am involved in a project development. Not sure when it happens.