RabbitMQ – queue empty after a restart, why even though it’s “durable”?

State before:

A message in the ‘send-email-queue’

and now restarting the server and the queue is cleared. Message is gone? Why?

You can see that there are two options for a message in “Delivery mode” when publishing the message from the rabbitmq panel – in a “queue” tab you will find this:

This is the panel site, so nothing to do with your producer in the code which should have the proper “delivery mode” when producing the message. So in the class of your producer you should set delivery mode to “2” which is “persistent”.

You can do that using php-amqplib/rabbitmq-bundle by extending your producer class with OldSound\RabbitMqBundle\RabbitMq\Producer which, itself extends BaseAmqp.

Once you’ve extended your producer class with OldSound\RabbitMqBundle\RabbitMq\Producer it’s already by default setting the “persistent” delivery mode and you will see in the panel instead of this messages “In memory”:

you will see it stored both in “Persistent” and “In memory”

Now even after a restart the messages are still int the queue, ready to be consumed.

Sample Producer:

<?php

namespace App\Producer;

use Exception;
use OldSound\RabbitMqBundle\RabbitMq\Producer;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class SendEmailProducer extends Producer
{
    /**
     * @var AMQPStreamConnection
     */
    private $rabbitConn;

    /**
     * SendEmailProducer constructor.
     * @param AMQPStreamConnection $rabbitConn
     */
    public function __construct(AMQPStreamConnection $rabbitConn)
    {
        parent::__construct($rabbitConn);
        $this->rabbitConn = $rabbitConn;
    }

    /**
     * @param string $msgBody (data for email)
     * @param string $routingKey
     * @param array $additionalProperties
     * @param array|null $headers
     * @throws Exception
     */
    public function publishUsingUserPrivacyChanges(string $msgBody, $routingKey = '', $additionalProperties = [], array $headers = null)
    {
        $channel = $this->rabbitConn->channel();

        //TODO: remove the key, and make the exchange a fanout!!!
        $channel->basic_publish(
                new AMQPMessage($msgBody, array_merge($this->getBasicProperties(), $additionalProperties)),
                'send-email-exchange',    //exchange
                $routingKey    //routing-key
            );

        $channel->close();
        $this->rabbitConn->close();
    }
}

Share this:

Related

Leave a comment Cancel reply