Merge pull request #53 from Articus/master

Custom tube names in Beanstalkd for queues

config/slm_queue_beanstalkd.local.php.dist

@@ -29,5 +29,12 @@ return array(
                 // 'timeout' => 2
             ),
         ),
+        /**
+         * Beanstalkd tubes that should be used for queues
+         */
+        'queues' => array(
+            // 'my-queue' => array('tube' => 'beanstalkd-tube-for-my-queue'),
+        ),
+
     ),
 );

src/SlmQueueBeanstalkd/Factory/BeanstalkdQueueFactory.php

@@ -2,6 +2,7 @@
 
 namespace SlmQueueBeanstalkd\Factory;
 
+use SlmQueueBeanstalkd\Options\QueueOptions;
 use SlmQueueBeanstalkd\Queue\BeanstalkdQueue;
 use Zend\ServiceManager\FactoryInterface;
 use Zend\ServiceManager\ServiceLocatorInterface;
@@ -20,6 +21,23 @@ public function createService(ServiceLocatorInterface $serviceLocator, $name = '
         $pheanstalk       = $parentLocator->get('SlmQueueBeanstalkd\Service\PheanstalkService');
         $jobPluginManager = $parentLocator->get('SlmQueue\Job\JobPluginManager');
 
-        return new BeanstalkdQueue($pheanstalk, $requestedName, $jobPluginManager);
+        $queueOptions = $this->getQueueOptions($parentLocator, $requestedName);
+
+        return new BeanstalkdQueue($pheanstalk, $requestedName, $jobPluginManager, $queueOptions);
+    }
+
+    /**
+     * Returns custom beanstalkd options for specified queue
+     * @param ServiceLocatorInterface $serviceLocator
+     * @param string $queueName
+     * @return QueueOptions
+     */
+    protected function getQueueOptions(ServiceLocatorInterface $serviceLocator, $queueName)
+    {
+        $config = $serviceLocator->get('Config');
+        $queuesOptions = isset($config['slm_queue']['queues'])? $config['slm_queue']['queues'] : array();
+        $queueOptions = isset($queuesOptions[$queueName])? $queuesOptions[$queueName] : array();
+
+        return new QueueOptions($queueOptions);
     }
 }

src/SlmQueueBeanstalkd/Options/BeanstalkdOptions.php

@@ -14,7 +14,6 @@ class BeanstalkdOptions extends AbstractOptions
      */
     protected $connection;
 
-
     /**
      * Set the connection options
      *

src/SlmQueueBeanstalkd/Options/QueueOptions.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace SlmQueueBeanstalkd\Options;
+
+use Zend\Stdlib\AbstractOptions;
+
+class QueueOptions extends AbstractOptions
+{
+    /**
+     * @var string
+     */
+    protected $tube = '';
+
+    /**
+     * Get beanstalkd tube name for queue
+     * @return string
+     */
+    public function getTube()
+    {
+        return $this->tube;
+    }
+
+    /**
+     * Set beanstalkd tube for queue
+     * @param string $tube
+     */
+    public function setTube($tube)
+    {
+        $this->tube = $tube;
+    }
+}

src/SlmQueueBeanstalkd/Queue/BeanstalkdQueue.php

@@ -7,6 +7,7 @@
 use SlmQueue\Job\JobInterface;
 use SlmQueue\Job\JobPluginManager;
 use SlmQueue\Queue\AbstractQueue;
+use SlmQueueBeanstalkd\Options\QueueOptions;
 
 /**
  * BeanstalkdQueue
@@ -18,6 +19,10 @@ class BeanstalkdQueue extends AbstractQueue implements BeanstalkdQueueInterface
      */
     protected $pheanstalk;
 
+    /**
+     * @var string
+     */
+    protected $tubeName;
 
     /**
      * Constructor
@@ -26,9 +31,17 @@ class BeanstalkdQueue extends AbstractQueue implements BeanstalkdQueueInterface
      * @param string           $name
      * @param JobPluginManager $jobPluginManager
      */
-    public function __construct(Pheanstalk $pheanstalk, $name, JobPluginManager $jobPluginManager)
-    {
+    public function __construct(
+        Pheanstalk $pheanstalk,
+        $name,
+        JobPluginManager $jobPluginManager,
+        QueueOptions $options = null
+    ) {
         $this->pheanstalk = $pheanstalk;
+        $this->tubeName = $name;
+        if (($options !== null) && $options->getTube()) {
+            $this->tubeName = $options->getTube();
+        }
         parent::__construct($name, $jobPluginManager);
     }
 
@@ -43,7 +56,7 @@ public function __construct(Pheanstalk $pheanstalk, $name, JobPluginManager $job
     public function push(JobInterface $job, array $options = array())
     {
         $identifier = $this->pheanstalk->putInTube(
-            $this->getName(),
+            $this->getTubeName(),
             $this->serializeJob($job),
             isset($options['priority']) ? $options['priority'] : Pheanstalk::DEFAULT_PRIORITY,
             isset($options['delay']) ? $options['delay'] : Pheanstalk::DEFAULT_DELAY,
@@ -64,7 +77,7 @@ public function push(JobInterface $job, array $options = array())
     public function pop(array $options = array())
     {
         $job = $this->pheanstalk->reserveFromTube(
-            $this->getName(),
+            $this->getTubeName(),
             isset($options['timeout']) ? $options['timeout'] : null
         );
 
@@ -118,7 +131,16 @@ public function bury(JobInterface $job, array $options = array())
      */
     public function kick($max)
     {
-        $this->pheanstalk->useTube($this->getName());
+        $this->pheanstalk->useTube($this->getTubeName());
         return $this->pheanstalk->kick($max);
     }
+
+    /**
+     * Get the name of the beanstalkd tube that is used for storing queue
+     * @return string
+     */
+    public function getTubeName()
+    {
+        return $this->tubeName;
+    }
 }

tests/SlmQueueBeanstalkdTest/Queue/BeanstalkdQueueTest.php

@@ -30,6 +30,13 @@ public function setUp()
         $this->queue = new BeanstalkdQueue($this->pheanstalk, $this->queueName, $this->pluginManager);
     }
 
+    public function testTubeNameGetter()
+    {
+        $tubeName = $this->queueName;
+        $result = $this->queue->getTubeName();
+        $this->assertEquals($result, $tubeName);
+    }
+
     public function testSuccessfulKickWithSelectedTube()
     {
         $maxKick    = 10;

tests/SlmQueueBeanstalkdTest/Queue/BeanstalkdQueueTubeTest.php

@@ -0,0 +1,108 @@
+<?php
+
+namespace SlmQueueBeanstalkdTest\Queue;
+
+use Pheanstalk\Job as PheanstalkJob;
+use PHPUnit_Framework_TestCase as TestCase;
+use SlmQueueBeanstalkd\Options\QueueOptions;
+use SlmQueueBeanstalkd\Queue\BeanstalkdQueue;
+use SlmQueueBeanstalkdTest\Asset\SimpleJob;
+
+/**
+ * BeanstalkdQueue Test for custom tube name
+ */
+class BeanstalkdQueueTubeTest extends TestCase
+{
+    /**
+     * @var string
+     */
+    protected $queueName;
+    /**
+     * @var string
+     */
+    protected $tubeName;
+    /**
+     * @var BeanstalkdQueue
+     */
+    protected $queue;
+    /**
+     * @var \Pheanstalk\Pheanstalk|\PHPUnit_Framework_MockObject_MockObject
+     */
+    protected $pheanstalk;
+    /**
+     * @var \SlmQueue\Job\JobPluginManager|\PHPUnit_Framework_MockObject_MockObject
+     */
+    protected $pluginManager;
+
+    public function setUp()
+    {
+        $this->queueName  = 'testQueueName';
+        $this->tubeName  = 'testQueueTubeName';
+        $this->pheanstalk = $this->getMockBuilder('Pheanstalk\Pheanstalk')
+                                 ->disableOriginalConstructor()
+                                 ->getMock();
+
+        $this->pluginManager = $this->getMockBuilder('SlmQueue\Job\JobPluginManager')
+                                    ->disableOriginalConstructor()
+                                    ->getMock();
+
+        $queueOptions = new QueueOptions();
+        $queueOptions->setTube($this->tubeName);
+
+        $this->queue = new BeanstalkdQueue($this->pheanstalk, $this->queueName, $this->pluginManager, $queueOptions);
+    }
+
+    public function testTubeNameGetter()
+    {
+        $tubeName = $this->tubeName;
+        $result = $this->queue->getTubeName();
+        $this->assertEquals($result, $tubeName);
+    }
+
+    public function testSuccessfulKickWithSelectedTube()
+    {
+        $maxKick    = 10;
+        $tubeName  = $this->tubeName;
+        $pheanstalk = $this->pheanstalk;
+
+        $pheanstalk->expects($this->once())
+                   ->method('useTube')
+                   ->with($this->equalTo($tubeName))
+                   ->will($this->returnValue($pheanstalk));
+
+        $pheanstalk->expects($this->once())
+                   ->method('kick')
+                   ->with($this->equalTo($maxKick))
+                   ->will($this->returnValue($maxKick));
+
+        $result = $this->queue->kick($maxKick);
+        $this->assertEquals($result, $maxKick);
+    }
+
+    public function testPopPreservesMetadata()
+    {
+        $pheanstalk     = $this->pheanstalk;
+        $tubeName      = $this->tubeName;
+        $pluginManager  = $this->pluginManager;
+
+        $job            = new SimpleJob;
+        $job->setMetadata('foo', 'bar');
+
+        $pheanstalkJob = new PheanstalkJob(1, $this->queue->serializeJob($job));
+
+        $pheanstalk->expects($this->once())
+                   ->method('reserveFromTube')
+                   ->with($this->equalTo($tubeName))
+                   ->will($this->returnValue($pheanstalkJob));
+
+        $pluginManager->expects($this->once())
+                      ->method('get')
+                      ->with(get_class($job))
+                      ->will($this->returnValue($job));
+
+        $result = $this->queue->pop();
+
+        $this->assertEquals($result, $job);
+        $this->assertEquals('bar', $job->getMetadata('foo'));
+    }
+}