[Symfony] delayed message is being received twice

[hadesy]

[debug] Set context's logger "Symfony\Component\Console\Logger\ConsoleLogger"
[debug] [AmqpDriver] Declare router exchange: enqueue.default
[debug] [AmqpDriver] Declare router queue: enqueue.app.default
[debug] [AmqpDriver] Bind router queue to exchange: enqueue.app.default -> enqueue.default
[debug] [AmqpDriver] Declare processor queue: enqueue.app.default
[debug] [AmqpDriver] Declare processor queue: symfony_events
[info] Start consuming
[info] Message received from the queue: enqueue.app.default
[debug] Headers: array (
  'message_id' => '15d58069-b7fc-4c10-a863-4ac9c9a7f19a',
  'correlation_id' => '',
  'app_id' => '',
  'type' => '',
  'content_encoding' => '',
  'content_type' => 'text/plain',
  'expiration' => '',
  'priority' => 2,
  'reply_to' => '',
  'timestamp' => 1537262335,
  'user_id' => '',
)
[debug] Properties: array (
  'enqueue.topic_name' => 'order.close',
  'enqueue-delay' => 1000,
)
[debug] Payload: 'close order #1'
[info] Message processed: enqueue.ack
[info] Message received from the queue: enqueue.app.default
[debug] Headers: array (
  'message_id' => '15d58069-b7fc-4c10-a863-4ac9c9a7f19a',
  'correlation_id' => '',
  'app_id' => '',
  'type' => '',
  'content_encoding' => '',
  'content_type' => 'text/plain',
  'expiration' => '',
  'priority' => 2,
  'reply_to' => '',
  'timestamp' => 1537262335,
  'user_id' => '',
)
[debug] Properties: array (
  'enqueue.topic_name' => 'order.close',
  'enqueue-delay' => 1000,
  'enqueue.processor_name' => 'App\\Processor\\Order\\OrderCloseProcessor',
  'enqueue.processor_queue_name' => 'default',
  'x-delay' => 1000,
)
[debug] Payload: 'close order #1'
[2018-09-18 09:18:56]process #15d58069-b7fc-4c10-a863-4ac9c9a7f19a
[info] Message processed: enqueue.ack

Why?

[hadesy]

It seems that

vendor/enqueue/amqp-ext/AmqpProducer.php 154

        if ($this->deliveryDelay) {
            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
        } elseif ($destination instanceof AmqpTopic) {
            $amqpExchange = new \AMQPExchange($this->amqpChannel);
            $amqpExchange->setType($destination->getType());
            $amqpExchange->setName($destination->getTopicName());
            $amqpExchange->setFlags(Flags::convertTopicFlags($destination->getFlags()));
            $amqpExchange->setArguments($destination->getArguments());

            $amqpExchange->publish(
                $message->getBody(),
                $message->getRoutingKey(),
                Flags::convertMessageFlags($message->getFlags()),
                $amqpAttributes
            );
        } else {
            /** @var AmqpQueue $destination */
            $amqpExchange = new \AMQPExchange($this->amqpChannel);
            $amqpExchange->setType(AMQP_EX_TYPE_DIRECT);
            $amqpExchange->setName('');

            $amqpExchange->publish(
                $message->getBody(),
                $destination->getQueueName(),
                Flags::convertMessageFlags($message->getFlags()),
                $amqpAttributes
            );
        }

Why not set $this->deliveryDelay when sending messages in the first time?

[makasim]

The message first goes to a router processor. Then, the router process re-sends it to final destinations.

See the difference in set properties.

[hadesy]

Can i understand like that?

[makasim]

@hadesy that's correct understanding.

[makasim]

Fixed in 0.9