AMQP keepalive support

.travis.yml

@@ -1,5 +1,7 @@
 language: php
 
+dist: trusty
+
 php:
   - 5.6
   - 7.0

CHANGELOG.md

@@ -4,9 +4,11 @@ Yii2 Queue Extension Change Log
 2.3.1 under development
 -----------------------
 
+- Enh #372: Add ability to configure keepalive and heartbeat for Amqp (vyachin)
 - Bug #380: Fixed amqp-interop queue/listen signal handling (tarinu)
 - Enh #388: `symfony/process 5.0` compatibility (leandrogehlen)
 
+
 2.3.0 June 04, 2019
 -------------------
 

composer.json

@@ -23,7 +23,7 @@
     "require-dev": {
         "yiisoft/yii2-redis": "*",
         "php-amqplib/php-amqplib": "*",
-        "enqueue/amqp-lib": "^0.8||^0.9.10",
+        "enqueue/amqp-lib": "^0.8",
         "pda/pheanstalk": "v3.*",
         "jeremeamia/superclosure": "*",
         "yiisoft/yii2-debug": "*",

src/drivers/amqp/Queue.php

@@ -31,6 +31,16 @@ class Queue extends CliQueue
     public $queueName = 'queue';
     public $exchangeName = 'exchange';
     public $vhost = '/';
+    /**
+     * @var int The periods of time PHP pings the broker in order to prolong the connection timeout. In seconds.
+     * @since 2.3.1
+     */
+    public $heartbeat = 0;
+    /**
+     * @var bool send keep-alive packets for a socket connection
+     * @since 2.3.1
+     */
+    public $keepalive = false;
     /**
      * @var string command class name
      */
@@ -118,7 +128,24 @@ protected function open()
         if ($this->channel) {
             return;
         }
-        $this->connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password, $this->vhost);
+        $this->connection = new AMQPStreamConnection(
+            $this->host,
+            $this->port,
+            $this->user,
+            $this->password,
+            $this->vhost,
+            false,
+            'AMQPLAIN',
+            null,
+            'en_US',
+            3.0,
+            3.0,
+            null,
+            $this->keepalive,
+            $this->heartbeat,
+            0.0,
+            null
+        );
         $this->channel = $this->connection->channel();
         $this->channel->queue_declare($this->queueName, false, true, false, false);
         $this->channel->exchange_declare($this->exchangeName, 'direct', false, true, false);

src/drivers/amqp_interop/Queue.php

@@ -107,6 +107,12 @@ class Queue extends CliQueue
      * @var bool|null
      */
     public $persisted;
+    /**
+     *
+     * @var bool send keep-alive packets for a socket connection
+     * @since 2.3.1
+     */
+    public $keepalive;
     /**
      * The connection will be established as later as possible if set true.
      *
@@ -373,6 +379,7 @@ protected function open()
             'connection_timeout' => $this->connectionTimeout,
             'heartbeat' => $this->heartbeat,
             'persisted' => $this->persisted,
+            'keepalive' => $this->keepalive,
             'lazy' => $this->lazy,
             'qos_global' => $this->qosGlobal,
             'qos_prefetch_size' => $this->qosPrefetchSize,