add queue create payload hook

laravel · Oct 1, 2018 · 3f68cbe · 3f68cbe
1 parent db6d5cb
commit 3f68cbe
Show file tree

Hide file tree

Showing 9 changed files with 102 additions and 28 deletions.
diff --git a/src/Illuminate/Queue/BeanstalkdQueue.php b/src/Illuminate/Queue/BeanstalkdQueue.php
@@ -68,7 +68,7 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        return $this->pushRaw($this->createPayload($job, $data), $queue);
+        return $this->pushRaw($this->createPayload($job, $this->getQueue($queue), $data), $queue);
     }
 
     /**
@@ -100,7 +100,7 @@ public function later($delay, $job, $data = '', $queue = null)
         $pheanstalk = $this->pheanstalk->useTube($this->getQueue($queue));
 
         return $pheanstalk->put(
-            $this->createPayload($job, $data),
+            $this->createPayload($job, $this->getQueue($queue), $data),
             Pheanstalk::DEFAULT_PRIORITY,
             $this->secondsUntil($delay),
             $this->timeToRun

diff --git a/src/Illuminate/Queue/DatabaseQueue.php b/src/Illuminate/Queue/DatabaseQueue.php
@@ -78,7 +78,9 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        return $this->pushToDatabase($queue, $this->createPayload($job, $data));
+        return $this->pushToDatabase($queue, $this->createPayload(
+            $job, $this->getQueue($queue), $data
+        ));
     }
 
     /**
@@ -105,7 +107,9 @@ public function pushRaw($payload, $queue = null, array $options = [])
      */
     public function later($delay, $job, $data = '', $queue = null)
     {
-        return $this->pushToDatabase($queue, $this->createPayload($job, $data), $delay);
+        return $this->pushToDatabase($queue, $this->createPayload(
+            $job, $this->getQueue($queue), $data
+        ), $delay);
     }
 
     /**
@@ -124,7 +128,7 @@ public function bulk($jobs, $data = '', $queue = null)
 
         return $this->database->table($this->table)->insert(collect((array) $jobs)->map(
             function ($job) use ($queue, $data, $availableAt) {
-                return $this->buildDatabaseRecord($queue, $this->createPayload($job, $data), $availableAt);
+                return $this->buildDatabaseRecord($queue, $this->createPayload($job, $this->getQueue($queue), $data), $availableAt);
             }
         )->all());
     }

diff --git a/src/Illuminate/Queue/Queue.php b/src/Illuminate/Queue/Queue.php
@@ -24,6 +24,13 @@ abstract class Queue
      */
     protected $connectionName;
 
+    /**
+     * The create payload callback.
+     *
+     * @var callable|null
+     */
+    protected static $createPayloadCallback;
+
     /**
      * Push a new job onto the queue.
      *
@@ -70,14 +77,15 @@ public function bulk($jobs, $data = '', $queue = null)
      * Create a payload string from the given job and data.
      *
      * @param  string  $job
+     * @param  string  $queue
      * @param  mixed   $data
      * @return string
      *
      * @throws \Illuminate\Queue\InvalidPayloadException
      */
-    protected function createPayload($job, $data = '')
+    protected function createPayload($job, $queue, $data = '')
     {
-        $payload = json_encode($this->createPayloadArray($job, $data));
+        $payload = json_encode($this->createPayloadArray($job, $queue, $data));
 
         if (JSON_ERROR_NONE !== json_last_error()) {
             throw new InvalidPayloadException(
@@ -92,35 +100,44 @@ protected function createPayload($job, $data = '')
      * Create a payload array from the given job and data.
      *
      * @param  mixed  $job
+     * @param  string  $queue
      * @param  mixed  $data
      * @return array
      */
-    protected function createPayloadArray($job, $data = '')
+    protected function createPayloadArray($job, $queue, $data = '')
     {
         return is_object($job)
-                    ? $this->createObjectPayload($job)
-                    : $this->createStringPayload($job, $data);
+                    ? $this->createObjectPayload($job, $queue)
+                    : $this->createStringPayload($job, $queue, $data);
     }
 
     /**
      * Create a payload for an object-based queue handler.
      *
      * @param  mixed  $job
+     * @param  string  $queue
      * @return array
      */
-    protected function createObjectPayload($job)
+    protected function createObjectPayload($job, $queue)
     {
-        return [
+        $payload = $this->withCreatePayloadHooks($queue, [
             'displayName' => $this->getDisplayName($job),
             'job' => 'Illuminate\Queue\CallQueuedHandler@call',
             'maxTries' => $job->tries ?? null,
             'timeout' => $job->timeout ?? null,
             'timeoutAt' => $this->getJobExpiration($job),
+            'data' => [
+                'commandName' => $job,
+                'command' => $job,
+            ],
+        ]);
+
+        return array_merge($payload, [
             'data' => [
                 'commandName' => get_class($job),
                 'command' => serialize(clone $job),
             ],
-        ];
+        ]);
     }
 
     /**
@@ -157,16 +174,48 @@ public function getJobExpiration($job)
      * Create a typical, string based queue payload array.
      *
      * @param  string  $job
+     * @param  string  $queue
      * @param  mixed  $data
      * @return array
      */
-    protected function createStringPayload($job, $data)
+    protected function createStringPayload($job, $queue, $data)
     {
-        return [
+        return $this->withCreatePayloadHooks($queue, [
             'displayName' => is_string($job) ? explode('@', $job)[0] : null,
-            'job' => $job, 'maxTries' => null,
-            'timeout' => null, 'data' => $data,
-        ];
+            'job' => $job,
+            'maxTries' => null,
+            'timeout' => null,
+            'data' => $data,
+        ]);
+    }
+
+    /**
+     * Register a callback to be executed when creating job payloads.
+     *
+     * @param  callable  $callback
+     * @return void
+     */
+    public static function createPayloadUsing($callback)
+    {
+        static::$createPayloadCallback = $callback;
+    }
+
+    /**
+     * Create the given payload using any registered payload hooks.
+     *
+     * @param  string  $queue
+     * @param  array  $payload
+     * @return array
+     */
+    protected function withCreatePayloadHooks($queue, array $payload)
+    {
+        if (static::$createPayloadCallback) {
+            return array_merge($payload, call_user_func(
+                static::$createPayloadCallback, $this->getConnectionName(), $queue, $payload
+            ));
+        }
+
+        return $payload;
     }
 
     /**

diff --git a/src/Illuminate/Queue/RedisQueue.php b/src/Illuminate/Queue/RedisQueue.php
@@ -88,7 +88,7 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        return $this->pushRaw($this->createPayload($job, $data), $queue);
+        return $this->pushRaw($this->createPayload($job, $this->getQueue($queue), $data), $queue);
     }
 
     /**
@@ -117,7 +117,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
      */
     public function later($delay, $job, $data = '', $queue = null)
     {
-        return $this->laterRaw($delay, $this->createPayload($job, $data), $queue);
+        return $this->laterRaw($delay, $this->createPayload($job, $this->getQueue($queue), $data), $queue);
     }
 
     /**
@@ -141,12 +141,13 @@ protected function laterRaw($delay, $payload, $queue = null)
      * Create a payload string from the given job and data.
      *
      * @param  string  $job
+     * @param  string   $queue
      * @param  mixed   $data
      * @return string
      */
-    protected function createPayloadArray($job, $data = '')
+    protected function createPayloadArray($job, $queue, $data = '')
     {
-        return array_merge(parent::createPayloadArray($job, $data), [
+        return array_merge(parent::createPayloadArray($job, $queue, $data), [
             'id' => $this->getRandomId(),
             'attempts' => 0,
         ]);

diff --git a/src/Illuminate/Queue/SqsQueue.php b/src/Illuminate/Queue/SqsQueue.php
@@ -72,7 +72,7 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        return $this->pushRaw($this->createPayload($job, $data), $queue);
+        return $this->pushRaw($this->createPayload($job, $queue ?: $this->default, $data), $queue);
     }
 
     /**
@@ -103,7 +103,7 @@ public function later($delay, $job, $data = '', $queue = null)
     {
         return $this->sqs->sendMessage([
             'QueueUrl' => $this->getQueue($queue),
-            'MessageBody' => $this->createPayload($job, $data),
+            'MessageBody' => $this->createPayload($job, $queue ?: $this->default, $data),
             'DelaySeconds' => $this->secondsUntil($delay),
         ])->get('MessageId');
     }

diff --git a/src/Illuminate/Queue/SyncQueue.php b/src/Illuminate/Queue/SyncQueue.php
@@ -34,7 +34,7 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        $queueJob = $this->resolveJob($this->createPayload($job, $data), $queue);
+        $queueJob = $this->resolveJob($this->createPayload($job, $queue, $data), $queue);
 
         try {
             $this->raiseBeforeJobEvent($queueJob);

diff --git a/tests/Queue/QueueDatabaseQueueUnitTest.php b/tests/Queue/QueueDatabaseQueueUnitTest.php
@@ -64,6 +64,7 @@ public function testFailureToCreatePayloadFromObject()
         $createPayload->setAccessible(true);
         $createPayload->invokeArgs($queue, [
             $job,
+            'queue-name',
         ]);
     }
 
@@ -78,6 +79,7 @@ public function testFailureToCreatePayloadFromArray()
         $createPayload->setAccessible(true);
         $createPayload->invokeArgs($queue, [
             ["\xc3\x28"],
+            'queue-name',
         ]);
     }
 

diff --git a/tests/Queue/QueueRedisQueueTest.php b/tests/Queue/QueueRedisQueueTest.php
@@ -3,6 +3,7 @@
 namespace Illuminate\Tests\Queue;
 
 use Mockery as m;
+use Illuminate\Queue\Queue;
 use PHPUnit\Framework\TestCase;
 
 class QueueRedisQueueTest extends TestCase
@@ -23,6 +24,23 @@ public function testPushProperlyPushesJobOntoRedis()
         $this->assertEquals('foo', $id);
     }
 
+    public function testPushProperlyPushesJobOntoRedisWithCustomPayloadHook()
+    {
+        $queue = $this->getMockBuilder('Illuminate\Queue\RedisQueue')->setMethods(['getRandomId'])->setConstructorArgs([$redis = m::mock('Illuminate\Contracts\Redis\Factory'), 'default'])->getMock();
+        $queue->expects($this->once())->method('getRandomId')->will($this->returnValue('foo'));
+        $redis->shouldReceive('connection')->once()->andReturn($redis);
+        $redis->shouldReceive('rpush')->once()->with('queues:default', json_encode(['displayName' => 'foo', 'job' => 'foo', 'maxTries' => null, 'timeout' => null, 'data' => ['data'], 'custom' => 'taylor', 'id' => 'foo', 'attempts' => 0]));
+
+        Queue::createPayloadUsing(function ($connection, $queue, $payload) {
+            return ['custom' => 'taylor'];
+        });
+
+        $id = $queue->push('foo', ['data']);
+        $this->assertEquals('foo', $id);
+
+        Queue::createPayloadUsing(null);
+    }
+
     public function testDelayedPushProperlyPushesJobOntoRedis()
     {
         $queue = $this->getMockBuilder('Illuminate\Queue\RedisQueue')->setMethods(['availableAt', 'getRandomId'])->setConstructorArgs([$redis = m::mock('Illuminate\Contracts\Redis\Factory'), 'default'])->getMock();

diff --git a/tests/Queue/QueueSqsQueueTest.php b/tests/Queue/QueueSqsQueueTest.php
@@ -87,7 +87,7 @@ public function testDelayedPushWithDateTimeProperlyPushesJobOntoSqs()
     {
         $now = \Illuminate\Support\Carbon::now();
         $queue = $this->getMockBuilder('Illuminate\Queue\SqsQueue')->setMethods(['createPayload', 'secondsUntil', 'getQueue'])->setConstructorArgs([$this->sqs, $this->queueName, $this->account])->getMock();
-        $queue->expects($this->once())->method('createPayload')->with($this->mockedJob, $this->mockedData)->will($this->returnValue($this->mockedPayload));
+        $queue->expects($this->once())->method('createPayload')->with($this->mockedJob, $this->queueName, $this->mockedData)->will($this->returnValue($this->mockedPayload));
         $queue->expects($this->once())->method('secondsUntil')->with($now)->will($this->returnValue(5));
         $queue->expects($this->once())->method('getQueue')->with($this->queueName)->will($this->returnValue($this->queueUrl));
         $this->sqs->shouldReceive('sendMessage')->once()->with(['QueueUrl' => $this->queueUrl, 'MessageBody' => $this->mockedPayload, 'DelaySeconds' => 5])->andReturn($this->mockedSendMessageResponseModel);
@@ -98,7 +98,7 @@ public function testDelayedPushWithDateTimeProperlyPushesJobOntoSqs()
     public function testDelayedPushProperlyPushesJobOntoSqs()
     {
         $queue = $this->getMockBuilder('Illuminate\Queue\SqsQueue')->setMethods(['createPayload', 'secondsUntil', 'getQueue'])->setConstructorArgs([$this->sqs, $this->queueName, $this->account])->getMock();
-        $queue->expects($this->once())->method('createPayload')->with($this->mockedJob, $this->mockedData)->will($this->returnValue($this->mockedPayload));
+        $queue->expects($this->once())->method('createPayload')->with($this->mockedJob, $this->queueName, $this->mockedData)->will($this->returnValue($this->mockedPayload));
         $queue->expects($this->once())->method('secondsUntil')->with($this->mockedDelay)->will($this->returnValue($this->mockedDelay));
         $queue->expects($this->once())->method('getQueue')->with($this->queueName)->will($this->returnValue($this->queueUrl));
         $this->sqs->shouldReceive('sendMessage')->once()->with(['QueueUrl' => $this->queueUrl, 'MessageBody' => $this->mockedPayload, 'DelaySeconds' => $this->mockedDelay])->andReturn($this->mockedSendMessageResponseModel);
@@ -109,7 +109,7 @@ public function testDelayedPushProperlyPushesJobOntoSqs()
     public function testPushProperlyPushesJobOntoSqs()
     {
         $queue = $this->getMockBuilder('Illuminate\Queue\SqsQueue')->setMethods(['createPayload', 'getQueue'])->setConstructorArgs([$this->sqs, $this->queueName, $this->account])->getMock();
-        $queue->expects($this->once())->method('createPayload')->with($this->mockedJob, $this->mockedData)->will($this->returnValue($this->mockedPayload));
+        $queue->expects($this->once())->method('createPayload')->with($this->mockedJob, $this->queueName, $this->mockedData)->will($this->returnValue($this->mockedPayload));
         $queue->expects($this->once())->method('getQueue')->with($this->queueName)->will($this->returnValue($this->queueUrl));
         $this->sqs->shouldReceive('sendMessage')->once()->with(['QueueUrl' => $this->queueUrl, 'MessageBody' => $this->mockedPayload])->andReturn($this->mockedSendMessageResponseModel);
         $id = $queue->push($this->mockedJob, $this->mockedData, $this->queueName);