Batch message publishing capabilities added to dev-master version. Some small fixes.

Author: FractalizeR

src/Jobs/AMQPJob.php

@@ -121,6 +121,7 @@ public function getQueue()
      * Get the job identifier.
      *
      * @return string
+     * @throws \OutOfBoundsException
      */
     public function getJobId()
     {

src/Queue/AMQPQueue.php

@@ -147,15 +147,44 @@ protected function declareExchange($exchangeName, $exchangeType, array $exchange
      */
     public function push($job, $data = '', $queue = null)
     {
-        $queue = $this->getQueueName($queue);
-        if ($this->declareQueues) {
-            $this->declareQueue($queue);
-        }
+        $queue = $this->prepareQueue($queue);
         $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
         $this->channel->basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
+
+        return true;
+    }
+
+
+    /**
+     * Adds the message to internal buffer to be sent later in a batch
+     *
+     * @param string $job   Job implementation class name
+     * @param mixed  $data  Job custom data. Usually array
+     * @param string $queue Queue name, if different from the default one
+     *
+     * @return bool
+     * @throws \Illuminate\Queue\InvalidPayloadException
+     * @throws AMQPException
+     */
+    public function addMessageToBatch($job, $data = '', $queue = null)
+    {
+        $queue = $this->prepareQueue($queue);
+        $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
+        $this->channel->batch_basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
+
         return true;
     }
 
+    /**
+     * Publishes and internal buffer of messages, that were added before with {@see addMessageToBatch}
+     *
+     * @return void
+     */
+    public function pushBatch()
+    {
+        $this->channel->publish_batch();
+    }
+
     /**
      * Helper to return a default queue name in case passed param is empty
      *
@@ -241,10 +270,7 @@ protected function getRoutingKey($queue)
      */
     public function pushRaw($payload, $queue = null, array $options = [])
     {
-        $queue = $this->getQueueName($queue);
-        if ($this->declareQueues) {
-            $this->declareQueue($queue);
-        }
+        $queue = $this->prepareQueue($queue);
         $amqpPayload = new AMQPMessage($payload, $this->messageProperties);
         $this->channel->basic_publish($amqpPayload, $this->exchangeName, $queue);
         return true;
@@ -259,6 +285,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
      * @param  string        $queue Queue name, if different from the default one
      *
      * @return bool Always true
+     * @throws \Illuminate\Queue\InvalidPayloadException
      * @throws AMQPException
      */
     public function later($delay, $job, $data = '', $queue = null)
@@ -267,10 +294,7 @@ public function later($delay, $job, $data = '', $queue = null)
             $delay = $delay->getTimestamp() - time();
         }
 
-        $queue = $this->getQueueName($queue);
-        if ($this->declareQueues) {
-            $this->declareQueue($queue);
-        }
+        $queue = $this->prepareQueue($queue);
         $delayedQueueName = $this->declareDelayedQueue($queue, $delay);
 
         $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
@@ -325,10 +349,7 @@ public function declareDelayedQueue($destinationQueueName, $delay)
      */
     public function pop($queue = null)
     {
-        $queue = $this->getQueueName($queue);
-        if ($this->declareQueues) {
-            $this->declareQueue($queue);
-        }
+        $queue = $this->prepareQueue($queue);
         $envelope = $this->channel->basic_get($queue);
 
         if ($envelope instanceof AMQPMessage) {
@@ -371,4 +392,22 @@ public function setMessageProperties(array $messageProperties)
 
         return $this;
     }
+
+    /**
+     * Prepares a queue for later use. Declares it if needed
+     *
+     * @param string $queue
+     *
+     * @return string
+     * @throws AMQPException
+     */
+    private function prepareQueue($queue)
+    {
+        $queue = $this->getQueueName($queue);
+        if ($this->declareQueues) {
+            $this->declareQueue($queue);
+        }
+
+        return $queue;
+    }
 }

src/Queue/QueueInfo.php

@@ -45,23 +45,23 @@ public static function createFromDeclareOk(array $declareOkData)
     /**
      * @return string
      */
-    public function getName(): string
+    public function getName()
     {
         return $this->name;
     }
 
     /**
      * @return int
      */
-    public function getJobs(): int
+    public function getJobs()
     {
         return $this->jobs;
     }
 
     /**
      * @return int
      */
-    public function getConsumers(): int
+    public function getConsumers()
     {
         return $this->consumers;
     }

src/Utility/ArrayUtil.php

@@ -54,7 +54,7 @@ public static function arrayFilterRecursive(array $input, callable $callback)
     public static function removeNullsRecursive(array $arr)
     {
         return static::arrayFilterRecursive($arr, function ($var) {
-            return !is_null($var);
+            return null !== $var;
         });
     }
 }