qos count configurable

Author: TomK

src/Provider/Amqp/AmqpQueueProvider.php

@@ -44,6 +44,8 @@ class AmqpQueueProvider extends AbstractQueueProvider
   protected $_reconnectInterval = 1800;
   protected $_lastConnectTime = 0;
 
+  protected $_qosCount;
+
   /**
    * @var AMQPMessage[]
    */
@@ -347,6 +349,14 @@ protected function _getChannel()
     if($this->_channel === null)
     {
       $this->_channel = $this->_getConnection()->channel();
+      $config = $this->config();
+      if($config->has('qos_count') || $config->has('qos_size'))
+      {
+        $this->setPrefetch(
+          $config->getItem('qos_count', 0),
+          $config->getItem('qos_size', 0)
+        );
+      }
     }
     return $this->_channel;
   }
@@ -383,8 +393,18 @@ public function disconnect()
     $this->_exchange = null;
   }
 
+  public function batchConsume(callable $callback, $batchSize)
+  {
+    if($this->_qosCount && $batchSize > $this->_qosCount)
+    {
+      throw new \Exception('Cannot consume batches greater than QoS');
+    }
+    parent::batchConsume($callback, $batchSize);
+  }
+
   public function setPrefetch($count, $size = 0)
   {
+    $this->_qosCount = $count;
     $this->_getChannel()->basic_qos($size, $count, false);
     return $this;
   }

tests/Provider/AmqpTest.php

@@ -1,6 +1,8 @@
 <?php
 namespace Packaged\Queue\Tests;
 
+use Packaged\Config\ConfigSectionInterface;
+use Packaged\Config\Provider\ConfigSection;
 use Packaged\Queue\Provider\Amqp\AmqpQueueProvider;
 
 class AmqpTest extends \PHPUnit_Framework_TestCase
@@ -24,12 +26,12 @@ function ($message, $deliveryTag) use ($q)
 
   public function testBatchAck()
   {
-    $q = AmqpQueueProvider::create('test.batch.ack')
+    $config = new ConfigSection('', ['wait_time' => 1, 'qos_count' => 250]);
+    $q = $this->_getQueue('test.batch.ack', $config)
       ->declareExchange()
       ->declareQueue()
       ->bindQueue()
       ->purge();
-    $q->config()->addItem('wait_time', 1);
 
     $total = 1000;
 
@@ -68,12 +70,12 @@ function (array $messages) use ($q, &$c)
 
   public function testBatchNack()
   {
-    $q = AmqpQueueProvider::create('test.batch.nack')
+    $config = new ConfigSection('', ['wait_time' => 1]);
+    $q = $this->_getQueue('test.batch.nack', $config)
       ->declareExchange()
       ->declareQueue()
       ->bindQueue()
       ->purge();
-    $q->config()->addItem('wait_time', 1);
 
     $total = 1000;
 
@@ -112,12 +114,12 @@ function (array $messages) use ($q, &$c)
 
   public function testRequeue()
   {
-    $q = AmqpQueueProvider::create('test.batch.requeue')
+    $config = new ConfigSection('', ['wait_time' => 1]);
+    $q = $this->_getQueue('test.batch.requeue', $config)
       ->declareExchange()
       ->declareQueue()
       ->bindQueue()
       ->purge();
-    $q->config()->addItem('wait_time', 1);
 
     $total = 250;
 
@@ -157,4 +159,16 @@ function (array $messages) use ($q, &$count)
     );
     $this->assertEquals($total, $count);
   }
+
+  protected function _getQueue(
+    $queueName, ConfigSectionInterface $config = null
+  )
+  {
+    $q = AmqpQueueProvider::create($queueName);
+    if($config)
+    {
+      $q->configure($config);
+    }
+    return $q;
+  }
 }