Feature/support-reconnect

[adm-bome]

Issue: #460
On-top of: #528

proposal branch: https://github.com/vyuldashev/laravel-queue-rabbitmq/compare/octane?expand=1

[matom20]

Laravel: 10.4.1
Octane: 1.5.0
Swoole: 5.0.2

RabbitMQ Server Config:
heartbeat = 10

Laravel queue.php config:

'options' => [
    'connection_timeout' => 0.7,
    'heartbeat' => 10,
],

Octane vs RabbitMQ only worked with this PR.

These exceptions still appear with heartbeat:

PhpAmqpLib\Exception\AMQPConnectionClosedException: CHANNEL_ERROR - expected 'channel.open'(60, 40)

PhpAmqpLib\Exception\AMQPHeartbeatMissedException: Missed server heartbeat in vendor/php-amqplib/php-amqplib/PhpAmqpLib/Wire/IO/AbstractIO.php:167

Simply trace:

#9 vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Queue/RabbitMQQueue.php(410): PhpAmqpLib\Connection\AbstractConnection->channel()
#10 vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Queue/RabbitMQQueue.php(705): VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue->isQueueExists('queue.name')
#11 vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Queue/RabbitMQQueue.php(121): VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue->declareDestination('queue.name', '', 'direct')
#12 vendor/vladimir-yuldashev/laravel-queue-rabbitmq/src/Queue/RabbitMQQueue.php(107): VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue->pushRaw('{"uuid":"ace2d8...', 'queue.name')

The solution was to add a method isQueueExists to ReconnectTrait.

trait ReconnectTrait
{
    protected function publishBasic($msg, $exchange = '', $destination = '', $mandatory = false, $immediate = false, $ticket = null): void
    {
        try {
            parent::publishBasic($msg, $exchange, $destination, $mandatory, $immediate, $ticket);
        } catch (AMQPConnectionClosedException|AMQPChannelClosedException) {
            $this->reconnect();
            parent::publishBasic($msg, $exchange, $destination, $mandatory, $immediate, $ticket);
        }
    }

    protected function publishBatch(): void
    {
        try {
            parent::publishBatch();
        } catch (AMQPConnectionClosedException|AMQPChannelClosedException) {
            $this->reconnect();
            parent::publishBatch();
        }
    }

    public function isQueueExists(string $name = null): bool
    {
        try {
            return parent::isQueueExists($name);
        } catch (AMQPConnectionClosedException|AMQPChannelClosedException) {
            $this->reconnect();
            return parent::isQueueExists($name);
        }
    }
    
}

[adm-bome]

@matom20 Thanks for your input.

Based on your input, the issue is bigger. The same applies for methods: size(), isExchangeExists() and purge().

Every time a channel is requested against a closed connection. Above methods make a temporary channel to request information or to perform an action ontop of RabbitMQ.

SideNote: The temporary channel is, so the main channel won't close when an exception is thrown.

In PR #528, i will extract/refactor a method, for creation of Channel. This can then be overwriten in the trait. All cases will be covered.

[matom20]

@adm-bome Commits 5b8508b b43ac84 worked!
Thanks 👍

[adm-bome]

@johnabil, @matom20 also @khepin and others.....

In this PR, "a" solution to support octane with reconnect.

These are my "2 cents" on how to solve octane with reconnection abbilities.

But, in my humble opinion you don't and never want to reconnect in this way. Workers or Connections should always die. So the managers in place (like: cli workers, horizon or others) can manage these workers/connections and cleanup or handle all logic. 

This is by the; Everything should die principle.

And of course there will be cases you won't want this to happen but that is a choice of a developer, in my opinion.

So, in this pull-request I also made an octane RabbitMQQueue::class with only a trait added. (Note: Not my choice of a solution).

My solution would be to only provide the reconnect Trait. When a developer needs or wants the reconnect functionality. A developer can implement this Trait into his own MyRabbitMQQueue:class. (as Add-on).

As for Octane: we should find a solution, the channel is not requested on a dead connection.

maybe i already did in my previous PR #528

Also see issue: #460

[adm-bome]

@khepin, @M-Porter

Based on my last comment above #531 (comment)

I added:

• a reconnect method, for handling the reconnection within this lib properly.
• an section "Use your own Worker class" to the ReadMe.

Note: even only a trait with overriding methods is a code smell. So I ditched this thought. Its entirely to a developer now.

[johnabil]

@adm-bome
I wasn't available on the last couple of weeks but I read all of your updates and the suggested solutions that you made and it's really great work. things should work proberly but I will read the commits and code once more with checking it on my project to see if the problem still exists.

I think we need to be sure with couple of checks like:

• making sure if we will disconnect the connection and open a new one to close the connection of the original app not the replica instances, you can check this comment to know exactly what I mean original bug.
• checking when the reconnect method will be called exactly.

[johnabil]

@adm-bome @khepin I tested the branch on my project and all the scenarios working well now with Laravel octane but I will continue the test cases and we can release beta tag to test it more and to make all of the package users to test it more.

[khepin]

Thanks!