[amqp][lib] Improve heartbeat handling. Introduce heartbeat on tick.

Author: makasim

pkg/amqp-lib/AmqpConnectionFactory.php

@@ -47,6 +47,7 @@ public function __construct($config = 'amqp:')
             ->addDefaultOption('login_response', null)
             ->addDefaultOption('locale', 'en_US')
             ->addDefaultOption('keepalive', false)
+            ->addDefaultOption('heartbeat_on_tick', true)
             ->parse()
         ;
 

pkg/amqp-lib/AmqpContext.php

@@ -110,7 +110,7 @@ public function createConsumer(Destination $destination): Consumer
      */
     public function createSubscriptionConsumer(): SubscriptionConsumer
     {
-        return new AmqpSubscriptionConsumer($this);
+        return new AmqpSubscriptionConsumer($this, (bool) $this->config['heartbeat_on_tick']);
     }
 
     /**

pkg/amqp-lib/AmqpSubscriptionConsumer.php

@@ -27,9 +27,15 @@ class AmqpSubscriptionConsumer implements InteropAmqpSubscriptionConsumer
      */
     private $subscribers;
 
-    public function __construct(AmqpContext $context)
+    /**
+     * @var bool
+     */
+    private $heartbeatOnTick;
+
+    public function __construct(AmqpContext $context, bool $heartbeatOnTick)
     {
         $this->context = $context;
+        $this->heartbeatOnTick = $heartbeatOnTick;
     }
 
     public function consume(int $timeout = 0): void
@@ -41,6 +47,12 @@ public function consume(int $timeout = 0): void
         $signalHandler = new SignalSocketHelper();
         $signalHandler->beforeSocket();
 
+        $heartbeatOnTick = function (AmqpContext $context) {
+            $context->getLibChannel()->getConnection()->getIO()->check_heartbeat();
+        };
+
+        $this->heartbeatOnTick && register_tick_function($heartbeatOnTick);
+
         try {
             while (true) {
                 $start = microtime(true);
@@ -69,6 +81,8 @@ public function consume(int $timeout = 0): void
             throw $e;
         } finally {
             $signalHandler->afterSocket();
+
+            $this->heartbeatOnTick && unregister_tick_function($heartbeatOnTick);
         }
     }
 

pkg/amqp-lib/Tests/AmqpContextTest.php

@@ -344,7 +344,7 @@ public function testShouldSetQos()
 
     public function testShouldReturnExpectedSubscriptionConsumerInstance()
     {
-        $context = new AmqpContext($this->createConnectionMock(), []);
+        $context = new AmqpContext($this->createConnectionMock(), ['heartbeat_on_tick' => true]);
 
         $this->assertInstanceOf(AmqpSubscriptionConsumer::class, $context->createSubscriptionConsumer());
     }

pkg/amqp-lib/Tests/AmqpSubscriptionConsumerTest.php

@@ -16,9 +16,9 @@ public function testShouldImplementSubscriptionConsumerInterface()
         $this->assertTrue($rc->implementsInterface(SubscriptionConsumer::class));
     }
 
-    public function testCouldBeConstructedWithAmqpContextAsFirstArgument()
+    public function testCouldBeConstructedWithAmqpContextAndHeartbeatOnTickAsArguments()
     {
-        new AmqpSubscriptionConsumer($this->createAmqpContextMock());
+        new AmqpSubscriptionConsumer($this->createAmqpContextMock(), $heartbeatOnTick = true);
     }
 
     /**

pkg/amqp-lib/composer.json

@@ -7,7 +7,7 @@
     "license": "MIT",
     "require": {
         "php": "^7.1.3",
-        "php-amqplib/php-amqplib": "^2.7",
+        "php-amqplib/php-amqplib": "^2.8",
         "queue-interop/amqp-interop": "^0.8",
         "enqueue/amqp-tools": "0.9.x-dev"
     },