How to 'ack' the message and exit then?

[KIVagant]

The problem:

• If \Doctrine\DBAL\DBALException caught inside a consumer, the Doctrine Entity Manager (EM) will be closed.
• Since EM closed Consumer cannot serve any other messages and should exit. Because no one can guarantee that some of the services hadn't copy a link to the closed EM.
• If Consumer has exited, the "bad" message will be returned to the queue and will kill all consumers one after one.

Solution 1:

• In the \OldSound\RabbitMqBundle\RabbitMq\Consumer::handleProcessMessage add the new signal, something like ConsumerInterface::MSG_ACK_AND_DIE. If it was caught, send "acknowledge" signal and exit then.

Solution 2 (not perfect):

• Consumer should set the flag $this->forceStop = true.
• In the \OldSound\RabbitMqBundle\RabbitMq\Consumer::processMessage, after call_user_func, this flag should be read from a callback object and if it is true, self 'forceStop' must be overridden.

Solution 3:

• Consumers can throw special exceptions like AckAndDieException, RequeueAndDieException or just RuntimeErrorException with one of ConsumerInterface::MSG_* signal in payload.
• If the exception was caught, $this->forceStop will be true, handleProcessMessage should be executed as usual (with flags from Exception payload) and then, maybeStopConsumer will call stopConsuming.

Solution 4:

• Add possibility to override classes from the vendor/php-amqplib/rabbitmq-bundle/Resources/config/rabbitmq.xml.
• For this goal, all old_sound_rabbit_mq.*.class parameters should be exposed in the \OldSound\RabbitMqBundle\DependencyInjection\Configuration::getConfigTreeBuilder.

[skafandri]

I usually try/catch exceptions and ack/nack accordingly to the exception type and situation.
This is not a workable solution for you?

[KIVagant]

@skafandri, yes, I'm (as Consumer) already caught DBALException and EM is already closed. And now I need to notify RMQ that this message is dangerous and need to be removed from queue. And then I need exit because Doctrine can't work anymore.

[KIVagant]

Currently I made a trick that works but with not really good way:

class MyConsumer implements ConsumerInterface
{
    private $dieSignal = false;
    public function execute(AMQPMessage $AMQPMessage)
    {
        if ($this->dieSignal) { // now the new, maybe normal message is processing but I need to exit because of previous error 
            exit;
        }
        try {
// ...
        } catch (DBALException $e) {
                $this->dieSignal = true;
                return true; // the only way to remove message from queue
        } 
    }
}