[Messenger] Fix amqp socket lost

Co-authored-by: Miguel Fernandez <[email protected]>

Author: 2 people

Transport/Connection.php

@@ -102,6 +102,11 @@ class Connection
      */
     private $amqpDelayExchange;
 
+    /**
+     * @var int
+     */
+    private $lastActivityTime = 0;
+
     public function __construct(array $connectionOptions, array $exchangeOptions, array $queuesOptions, AmqpFactory $amqpFactory = null)
     {
         if (!\extension_loaded('amqp')) {
@@ -347,6 +352,8 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
         $attributes['delivery_mode'] = $attributes['delivery_mode'] ?? 2;
         $attributes['timestamp'] = $attributes['timestamp'] ?? time();
 
+        $this->lastActivityTime = time();
+
         $exchange->publish(
             $body,
             $routingKey,
@@ -510,6 +517,11 @@ static function (): bool {
                     }
                 );
             }
+
+            $this->lastActivityTime = time();
+        } elseif (0 < ($this->connectionOptions['heartbeat'] ?? 0) && time() > $this->lastActivityTime + 2 * $this->connectionOptions['heartbeat']) {
+            $disconnectMethod = 'true' === ($this->connectionOptions['persistent'] ?? 'false') ? 'pdisconnect' : 'disconnect';
+            $this->amqpChannel->getConnection()->{$disconnectMethod}();
         }
 
         return $this->amqpChannel;