support heartbeat via configuration (#21)

Author: TomK

src/Provider/Amqp/AmqpQueueProvider.php

@@ -9,7 +9,9 @@
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
+use PhpAmqpLib\Connection\Heartbeat\PCNTLHeartbeatSender;
 use PhpAmqpLib\Exception\AMQPProtocolChannelException;
+use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Exception\AMQPTimeoutException;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
@@ -83,6 +85,11 @@ class AmqpQueueProvider extends AbstractQueueProvider
    */
   private $_slowPushThreshold = 0;
 
+  /**
+   * @var ?PCNTLHeartbeatSender
+   */
+  protected $_heartbeatSender;
+
   protected function _construct()
   {
     $this->_fixedConsumerCallback = [$this, 'consumerCallback'];
@@ -116,9 +123,8 @@ public function pushBatch(array $batch, $persistent = null)
     $returnCallback = null;
     if($mandatory)
     {
-      $returnCallback = function ($replyCode, $replyText, $exchange, $routingKey) use
-      (&$needRetry, &$needDeclare, &$autoDeclare, $declareAttempts, $declareRetryLimit)
-      {
+      $returnCallback = function ($replyCode, $replyText, $exchange, $routingKey)
+      use (&$needRetry, &$needDeclare, &$autoDeclare, $declareAttempts, $declareRetryLimit) {
         if($autoDeclare && ($declareAttempts < $declareRetryLimit) && ($replyCode == 312))
         {
           $needDeclare = true;
@@ -520,7 +526,17 @@ protected function _getConnection($connectionMode)
           $host,
           $config->getItem('port', 5672),
           $config->getItem('username', 'guest'),
-          $config->getItem('password', 'guest')
+          $config->getItem('password', 'guest'),
+          $config->getItem('vhost', '/'),
+          false,
+          'AMQPLAIN',
+          null,
+          'en_US',
+          $config->getItem('connection_timeout', 3),
+          $config->getItem('read_write_timeout', 3),
+          null,
+          (bool)$config->getItem('keepalive', false),
+          $config->getItem('heartbeat', 0)
         );
       }
       catch(Exception $e)
@@ -534,6 +550,20 @@ protected function _getConnection($connectionMode)
       );
       $this->_lastConnectTimes[$connectionMode] = time();
     }
+
+    try
+    {
+      if($this->_heartbeatSender)
+      {
+        $this->_heartbeatSender->unregister();
+      }
+      $this->_heartbeatSender = new PCNTLHeartbeatSender($this->_connections[$connectionMode]);
+      $this->_heartbeatSender->register();
+    }
+    catch(AMQPRuntimeException $e)
+    {
+    }
+
     return $this->_connections[$connectionMode];
   }
 
@@ -630,6 +660,12 @@ private function _disconnect($connectionMode)
     {
     }
     $this->_channels[$connectionMode] = null;
+
+    if($this->_heartbeatSender)
+    {
+      $this->_heartbeatSender->unregister();
+      $this->_heartbeatSender = null;
+    }
     try
     {
       if((!empty($this->_connections[$connectionMode]))

tests/Provider/AmqpTest.php

@@ -7,10 +7,17 @@
 
 class AmqpTest extends \PHPUnit_Framework_TestCase
 {
+  protected function _getProvider(string $queue, ?string $exchange = null)
+  {
+    $q = AmqpQueueProvider::create($queue, $exchange);
+    $q->configure(new ConfigSection('', ['heartbeat' => 2]));
+    return $q;
+  }
+
   public function testAmqp()
   {
-    $q = AmqpQueueProvider::create('test', 'testexchange')
-      ->declareExchange()
+    $q = $this->_getProvider('test', 'testexchange');
+    $q->declareExchange()
       ->declareQueue()
       ->bindQueue()
       ->push('this is a test');
@@ -25,7 +32,7 @@ function ($message, $deliveryTag) use ($q) {
 
   public function testQueueExists()
   {
-    $q = AmqpQueueProvider::create('new_queue');
+    $q = $this->_getProvider('new_queue');
     $this->assertFalse($q->queueExists());
     $q->declareQueue();
     $this->assertTrue($q->queueExists());
@@ -35,7 +42,7 @@ public function testQueueExists()
 
   public function testExchangeExists()
   {
-    $q = AmqpQueueProvider::create('new_queue_e', 'new_exchange');
+    $q = $this->_getProvider('new_queue_e', 'new_exchange');
     $this->assertFalse($q->exchangeExists());
     $q->declareExchange();
     $this->assertTrue($q->exchangeExists());
@@ -179,7 +186,7 @@ protected function _getQueue(
     $queueName, ConfigSectionInterface $config = null
   )
   {
-    $q = AmqpQueueProvider::create($queueName);
+    $q = $this->_getProvider($queueName);
     if($config)
     {
       $q->configure($config);
@@ -200,7 +207,7 @@ public function testMandatory(
     $config, $queueName, $createExchange, $createQueue, $createBinding
   )
   {
-    $q = AmqpQueueProvider::create($queueName)->deleteQueueAndExchange();
+    $q = $this->_getProvider($queueName)->deleteQueueAndExchange();
     $q->configure(new ConfigSection('', $config));
 
     if($createExchange)