php-enqueue · makasim · Nov 26, 2018 · Nov 26, 2018 · Nov 26, 2018 · Tobion
diff --git a/pkg/amqp-lib/AmqpConnectionFactory.php b/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()
        ;


diff --git a/pkg/amqp-lib/AmqpContext.php b/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']);
    }

    /**

diff --git a/pkg/amqp-lib/AmqpSubscriptionConsumer.php b/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);
        }
    }


diff --git a/pkg/amqp-lib/Tests/AmqpContextTest.php b/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());
    }

diff --git a/pkg/amqp-lib/Tests/AmqpSubscriptionConsumerTest.php b/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);
    }

    /**

diff --git a/pkg/amqp-lib/Tests/Functional/AmqpSubscriptionConsumerWithHeartbeatOnTickTest.php b/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;
+    }
+}
diff --git a/pkg/amqp-lib/composer.json b/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"
    },
-Original file line number
+Diff line change
@@ -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
             Copy link

  
      
    
  

  
      



      

  

  
    
      

      
            Tobion
  

      

      

      


        Nov 29, 2018


      
    

  


        
      
  
  
  
    

    There was a problem hiding this comment.


  

 
  
    

    Choose a reason for hiding this comment

    
      The reason will be displayed to describe this comment to others. Learn more.
    

    
      
      


  


  
    
      I don't see where this test actually tests anything related to ticks. The test would also pass if no tick handler was registered. Or am I missing something?
    
  
  


    

        
      
  
  
    
  
  
  
    
    Sorry, something went wrong.
  

  
    
  
    
      

              Uh oh!

              
There was an error while loading. Please reload this page.


  
  


          
      
  
    
    
      
        
            
    All reactions
  


          
          
        
      
    

    



  
        
  
    
        
    
  


      
          
  
      
            Copy link

  
      
    
  

  
      

  
  Member


      

  Author


  

  
    
      

      
            makasim
  

      

      

      


        Nov 30, 2018


      
    

  


        
      
  
  
  
    

    There was a problem hiding this comment.


  

 
  
    

    Choose a reason for hiding this comment

    
      The reason will be displayed to describe this comment to others. Learn more.
    

    
      
      


  


  
    
      It contains decalre tricks which triggers tick callback and therfor heartbeat logic.
Is there a better way to test it?
    
  
  


    

        
      
  
  
    
  
  
  
    
    Sorry, something went wrong.
  

  
    
  
    
      

              Uh oh!

              
There was an error while loading. Please reload this page.


  
  


          
      
  
    
    
      
        
            
    All reactions
+    {
+        /**
+         * @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;
+        }
+    }