Merge pull request #658 from php-enqueue/amqp-lib-improve-hearbeat

[amqp][lib] Improve heartbeat handling. Introduce heartbeat on tick. Fixes "Invalid frame type 65" and "Broken pipe or closed connection"

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, $this->context);
+
         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/Tests/Functional/AmqpSubscriptionConsumerWithHeartbeatOnTickTest.php

@@ -0,0 +1,83 @@
+<?php
+
+namespace Enqueue\AmqpLib\Tests\Functional;
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Enqueue\AmqpLib\AmqpContext;
+use Interop\Amqp\AmqpQueue;
+use Interop\Queue\Consumer;
+use Interop\Queue\Context;
+use Interop\Queue\Message;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+class AmqpSubscriptionConsumerWithHeartbeatOnTickTest extends TestCase
+{
+    /**
+     * @var Context
+     */
+    private $context;
+
+    protected function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function test()
+    {
+        $this->context = $context = $this->createContext();
+
+        $fooQueue = $this->createQueue($context, 'foo_subscription_consumer_consume_from_all_subscribed_queues_spec');
+
+        $expectedFooBody = 'fooBody';
+
+        $context->createProducer()->send($fooQueue, $context->createMessage($expectedFooBody));
+
+        $fooConsumer = $context->createConsumer($fooQueue);
+
+        $actualBodies = [];
+        $actualQueues = [];
+        $callback = function (Message $message, Consumer $consumer) use (&$actualBodies, &$actualQueues) {
+            declare(ticks=1) {
+                $actualBodies[] = $message->getBody();
+                $actualQueues[] = $consumer->getQueue()->getQueueName();
+
+                $consumer->acknowledge($message);
+
+                return true;
+            }
+        };
+
+        $subscriptionConsumer = $context->createSubscriptionConsumer();
+        $subscriptionConsumer->subscribe($fooConsumer, $callback);
+
+        $subscriptionConsumer->consume(1000);
+
+        $this->assertCount(1, $actualBodies);
+    }
+
+    protected function createContext(): AmqpContext
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        $context = $factory->createContext();
+        $context->setQos(0, 5, false);
+
+        return $context;
+    }
+
+    protected function createQueue(AmqpContext $context, string $queueName): AmqpQueue
+    {
+        $queue = $context->createQueue($queueName);
+        $context->declareQueue($queue);
+        $context->purgeQueue($queue);
+
+        return $queue;
+    }
+}

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"
     },