Using latest PHPUnit 6.1 and Orchestra TestBench 3.4, feature to skip declaring queues added (hopefully addresses #1), size() implemented for queues

FractalizeR · FractalizeR · commit 9359ba3b3332 · 2017-06-01T23:36:49.000+03:00
diff --git a/README.md b/README.md
@@ -35,6 +35,7 @@ In your ```config/queue.php``` file you have to provide the following:
         'vhost' => '/',
         'queue' => null,
         'queue_flags' => ['durable' => true, 'routing_key' => null], //Durable queue (survives server crash)
+        'declare_queues' => true, //If we need to declare queues each time before sending a message. If not, you will have to declare them manually elsewhere
         'message_properties' => ['delivery_mode' => 2], //Persistent messages (survives server crash)
         'channel_id' => null,
         'exchange_name' => null,
diff --git a/composer.json b/composer.json
@@ -10,8 +10,8 @@
   "require-dev": {
     "laravel/framework": ">=5.3.0",
     "squizlabs/php_codesniffer": "1.*",
-    "orchestra/testbench": "3.3.*",
-    "phpunit/phpunit": "4.3.*"
+    "orchestra/testbench": "~3.4",
+    "phpunit/phpunit": "~6.1"
   },
   "autoload": {
     "psr-4": {
diff --git a/src/Connectors/AmqpConnector.php b/src/Connectors/AmqpConnector.php
@@ -48,14 +48,9 @@ public function connect(array $config)
         }
 
         return new AMQPQueue(
-            $connection,
-            $config['queue'],
-            $config['queue_flags'],
-            $config['message_properties'],
-            $config['channel_id'],
-            $config['exchange_name'],
-            $config['exchange_type'],
-            $config['exchange_flags']
+            $connection, $config['queue'], $config['queue_flags'], $config['declare_queues'],
+            $config['message_properties'], $config['channel_id'],
+            $config['exchange_name'], $config['exchange_type'], $config['exchange_flags']
         );
     }
 }
diff --git a/src/Queue/AMQPQueue.php b/src/Queue/AMQPQueue.php
@@ -52,32 +52,42 @@ class AMQPQueue extends Queue implements QueueContract
      * @var string Default channel id if needed
      */
     private $defaultChannelId;
+
     /**
      * @var array
      */
     private $queueFlags;
+
     /**
      * @var array
      */
     private $messageProperties;
 
+    /**
+     * @var bool
+     */
+    private $declareQueues;
+
     /**
      * @param AMQPStreamConnection $connection
-     * @param string         $defaultQueueName  Default queue name
-     * @param array          $queueFlags        Queue flags See a list of parameters to
-     *                                          \PhpAmqpLib\Channel\AMQPChannel::queue_declare. Parameters should be
-     *                                          passed like for call_user_func_array in this parameter
-     * @param array          $messageProperties This is passed as a second parameter to \PhpAmqpLib\Message\AMQPMessage
-     *                                          constructor
-     * @param string         $defaultChannelId  Default channel id
-     * @param string         $exchangeName      Exchange name
-     * @param mixed          $exchangeType      Exchange type
-     * @param mixed          $exchangeFlags     Exchange flags
+     * @param string               $defaultQueueName  Default queue name
+     * @param array                $queueFlags        Queue flags See a list of parameters to
+     *                                                \PhpAmqpLib\Channel\AMQPChannel::queue_declare. Parameters should
+     *                                                be passed like for call_user_func_array in this parameter
+     * @param bool                 $declareQueues     If we should declare queues before actually trying to send a
+     *                                                message
+     * @param array                $messageProperties This is passed as a second parameter to
+     *                                                \PhpAmqpLib\Message\AMQPMessage constructor
+     * @param string               $defaultChannelId  Default channel id
+     * @param string               $exchangeName      Exchange name
+     * @param mixed                $exchangeType      Exchange type
+     * @param mixed                $exchangeFlags     Exchange flags
      */
     public function __construct(
         AMQPStreamConnection $connection,
         $defaultQueueName = null,
         $queueFlags = [],
+        $declareQueues = true,
         $messageProperties = [],
         $defaultChannelId = null,
         $exchangeName = '',
@@ -87,6 +97,7 @@ public function __construct(
         $this->connection = $connection;
         $this->defaultQueueName = $defaultQueueName ?: 'default';
         $this->queueFlags = $queueFlags;
+        $this->declareQueues = $declareQueues;
         $this->messageProperties = $messageProperties;
         $this->defaultChannelId = $defaultChannelId;
         $this->exchangeName = $exchangeName;
@@ -98,9 +109,9 @@ public function __construct(
     }
 
     /**
-     * @param string $exchangeName The name of the exchange. For example, 'logs'
-     * @param string $exchangeType The type of the exchange. See EXCHANGE_TYPE_* constants for details
-     * @param array $exchangeFlags The flags of the exchange. See \PhpAmqpLib\Channel\AMQPChannel::exchange_declare
+     * @param string $exchangeName  The name of the exchange. For example, 'logs'
+     * @param string $exchangeType  The type of the exchange. See EXCHANGE_TYPE_* constants for details
+     * @param array  $exchangeFlags The flags of the exchange. See \PhpAmqpLib\Channel\AMQPChannel::exchange_declare
      *                              (from third parameter onwards). Must be an assoc array. Default flags can be omitted
      *
      * @see \PhpAmqpLib\Channel\AMQPChannel::exchange_declare
@@ -109,15 +120,15 @@ public function __construct(
     protected function declareExchange($exchangeName, $exchangeType, array $exchangeFlags = [])
     {
         $flags = array_replace([
-            'exchange' => $exchangeName,
-            'type' => $exchangeType,
-            'passive' => false,
-            'durable' => false,
+            'exchange'    => $exchangeName,
+            'type'        => $exchangeType,
+            'passive'     => false,
+            'durable'     => false,
             'auto_delete' => true,
-            'internal' => false,
-            'nowait' => false,
-            'arguments' => null,
-            'ticket' => null,
+            'internal'    => false,
+            'nowait'      => false,
+            'arguments'   => null,
+            'ticket'      => null,
         ], $exchangeFlags);
 
         call_user_func_array([$this->channel, 'exchange_declare'], $flags);
@@ -136,7 +147,9 @@ protected function declareExchange($exchangeName, $exchangeType, array $exchange
     public function push($job, $data = '', $queue = null)
     {
         $queue = $this->getQueueName($queue);
-        $this->declareQueue($queue);
+        if ($this->declareQueues) {
+            $this->declareQueue($queue);
+        }
         $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
         $this->channel->basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
         return true;
@@ -164,29 +177,29 @@ protected function getQueueName($name)
      *
      * @param string $name The name of the queue to declare
      *
-     * @return void
+     * @return QueueInfo
      */
     public function declareQueue($name)
     {
         $queue = $this->getQueueName($name);
         $flags = array_replace_recursive([
-            'queue' => $queue,
-            'passive' => false,
-            'durable' => false,
-            'exclusive' => false,
+            'queue'       => $queue,
+            'passive'     => false,
+            'durable'     => false,
+            'exclusive'   => false,
             'auto_delete' => true,
-            'nowait' => false,
-            'arguments' => null,
-            'ticket' => null,
+            'nowait'      => false,
+            'arguments'   => null,
+            'ticket'      => null,
         ], $this->getQueueFlags($name));
 
-        call_user_func_array([$this->channel, 'queue_declare'], $flags);
+        return QueueInfo::createFromDeclareOk(call_user_func_array([$this->channel, 'queue_declare'], $flags));
     }
 
     /**
-     * @param string $queueName
+     * @param string      $queueName
      * @param null|string $deferredQueueName
-     * @param null|int $deferredQueueDelay
+     * @param null|int    $deferredQueueDelay
      *
      * @return array
      */
@@ -206,6 +219,7 @@ protected function getQueueFlags($queueName, $deferredQueueName = null, $deferre
      *  Get routing key from config or use default one (queue name)
      *
      * @param $queue string
+     *
      * @return string Routing key name
      */
     protected function getRoutingKey($queue)
@@ -225,7 +239,9 @@ protected function getRoutingKey($queue)
     public function pushRaw($payload, $queue = null, array $options = [])
     {
         $queue = $this->getQueueName($queue);
-        $this->declareQueue($queue);
+        if ($this->declareQueues) {
+            $this->declareQueue($queue);
+        }
         $payload = new AMQPMessage($payload, $this->messageProperties);
         $this->channel->basic_publish($payload, $this->exchangeName, $queue);
         return true;
@@ -248,7 +264,9 @@ public function later($delay, $job, $data = '', $queue = null)
         }
 
         $queue = $this->getQueueName($queue);
-        $this->declareQueue($queue);
+        if ($this->declareQueues) {
+            $this->declareQueue($queue);
+        }
         $delayedQueueName = $this->declareDelayedQueue($queue, $delay);
 
         $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
@@ -270,21 +288,21 @@ public function declareDelayedQueue($destinationQueueName, $delay)
         $deferredQueueName = $destinationQueueName . '_deferred_' . $delay;
 
         $flags = array_replace_recursive([
-            'queue' => '',
-            'passive' => false,
-            'durable' => false,
-            'exclusive' => false,
+            'queue'       => '',
+            'passive'     => false,
+            'durable'     => false,
+            'exclusive'   => false,
             'auto_delete' => true,
-            'nowait' => false,
-            'arguments' => null,
-            'ticket' => null,
+            'nowait'      => false,
+            'arguments'   => null,
+            'ticket'      => null,
         ], $this->getQueueFlags($destinationQueueName, $deferredQueueName, $delay), [
-            'queue' => $deferredQueueName,
-            'durable' => true,
+            'queue'     => $deferredQueueName,
+            'durable'   => true,
             'arguments' => new AMQPTable([
-                'x-dead-letter-exchange' => '',
+                'x-dead-letter-exchange'    => '',
                 'x-dead-letter-routing-key' => $destinationQueueName,
-                'x-message-ttl' => $delay * 1000,
+                'x-message-ttl'             => $delay * 1000,
             ]),
         ]);
 
@@ -302,7 +320,9 @@ public function declareDelayedQueue($destinationQueueName, $delay)
     public function pop($queue = null)
     {
         $queue = $this->getQueueName($queue);
-        $this->declareQueue($queue);
+        if ($this->declareQueues) {
+            $this->declareQueue($queue);
+        }
         $envelope = $this->channel->basic_get($queue);
 
         if ($envelope instanceof AMQPMessage) {
@@ -316,9 +336,12 @@ public function pop($queue = null)
      * Get the size of the queue.
      *
      * @param  string $queue
+     *
      * @return int
      */
     public function size($queue = null)
     {
+        $data = $this->declareQueue($this->getQueueName($queue));
+        return $data->getJobs();
     }
 }
diff --git a/src/Queue/QueueInfo.php b/src/Queue/QueueInfo.php
@@ -0,0 +1,68 @@
+<?php
+
+namespace Forumhouse\LaravelAmqp\Queue;
+
+/**
+ * Class providing information about a queue
+ */
+class QueueInfo
+{
+    /**
+     * @var string The name of the queue
+     */
+    private $name;
+
+    /**
+     * @var int The number of ready messages in the queue
+     */
+    private $jobs;
+
+    /**
+     * @var int The number of consumers currently connected to read from queue
+     */
+    private $consumers;
+
+    public function __construct($name, $jobs, $consumers)
+    {
+        $this->name = $name;
+        $this->jobs = $jobs;
+        $this->consumers = $consumers;
+    }
+
+    /**
+     * Creates an instance from the data returned from queue_declare call
+     *
+     * @param array $declareOkData An array of data returned by queue_declare
+     *
+     * @return static
+     */
+    public static function createFromDeclareOk(array $declareOkData)
+    {
+        list($name, $jobs, $consumers) = $declareOkData;
+        return new static($name, $jobs, $consumers);
+    }
+
+    /**
+     * @return string
+     */
+    public function getName(): string
+    {
+        return $this->name;
+    }
+
+    /**
+     * @return int
+     */
+    public function getJobs(): int
+    {
+        return $this->jobs;
+    }
+
+    /**
+     * @return int
+     */
+    public function getConsumers(): int
+    {
+        return $this->consumers;
+    }
+}
diff --git a/tests/BaseFeaturesTest.php b/tests/BaseFeaturesTest.php
@@ -2,6 +2,7 @@
 
 namespace Forumhouse\LaravelAmqp\Tests;
 
+use Forumhouse\LaravelAmqp\Jobs\AMQPJob;
 use Illuminate\Queue\Jobs\Job;
 
 /**
@@ -32,7 +33,7 @@ public function setUp()
     public function testSimpleDelete()
     {
         $this->app['queue']->push(self::TEST_JOB_CLASS, ['test1' => 1, 'test2' => 2, 'delete' => true]);
-        /** @var Job $job */
+        /** @var AMQPJob $job */
         $job = $this->app['queue']->pop();
         $this->assertInstanceOf('Illuminate\Queue\Jobs\Job', $job);
         $job->fire();
@@ -51,7 +52,7 @@ public function testSimpleRelease()
     {
         $this->app['queue']->push(self::TEST_JOB_CLASS, ['test1' => 1, 'test2' => 2, 'release' => true]);
 
-        /** @var Job $job */
+        /** @var AMQPJob $job */
         $job = $this->app['queue']->pop();
         $this->assertInstanceOf('Illuminate\Queue\Jobs\Job', $job);
         $this->assertEquals(0, $job->attempts());
diff --git a/tests/LaravelAmqpTestBase.php b/tests/LaravelAmqpTestBase.php

Original file line number	Diff line number	Diff line change
`@@ -48,14 +48,9 @@ public function connect(array $config)`
`48`	`48`	`}`
`49`	`49`
`50`	`50`	`return new AMQPQueue(`
`51`		`- $connection,`
`52`		`- $config['queue'],`
`53`		`- $config['queue_flags'],`
`54`		`- $config['message_properties'],`
`55`		`- $config['channel_id'],`
`56`		`- $config['exchange_name'],`
`57`		`- $config['exchange_type'],`
`58`		`- $config['exchange_flags']`
	`51`	`+ $connection, $config['queue'], $config['queue_flags'], $config['declare_queues'],`
	`52`	`+ $config['message_properties'], $config['channel_id'],`
	`53`	`+ $config['exchange_name'], $config['exchange_type'], $config['exchange_flags']`
`59`	`54`	`);`
`60`	`55`	`}`
`61`	`56`	`}`