Refactor how connection names are set so they are always set.

This fixes problem where connection name would only be set if running through queue:listen and queue:work commands but not if you manually called pop() somewhere in your own code.
laravel · Dec 30, 2016 · 4c600fb · 4c600fb
1 parent feb52bf
commit 4c600fb
Show file tree

Hide file tree

Showing 18 changed files with 88 additions and 20 deletions.
diff --git a/src/Illuminate/Contracts/Queue/Queue.php b/src/Illuminate/Contracts/Queue/Queue.php
@@ -81,4 +81,19 @@ public function bulk($jobs, $data = '', $queue = null);
      * @return \Illuminate\Contracts\Queue\Job|null
      */
     public function pop($queue = null);
+
+    /**
+     * Get the connection name for the queue.
+     *
+     * @return string
+     */
+    public function getConnectionName();
+
+    /**
+     * Set the connection name for the queue.
+     *
+     * @param  string  $name
+     * @return $this
+     */
+    public function setConnectionName($name);
 }
diff --git a/src/Illuminate/Queue/BeanstalkdQueue.php b/src/Illuminate/Queue/BeanstalkdQueue.php
@@ -120,7 +120,9 @@ public function pop($queue = null)
         $job = $this->pheanstalk->watchOnly($queue)->reserve(0);
 
         if ($job instanceof PheanstalkJob) {
-            return new BeanstalkdJob($this->container, $this->pheanstalk, $job, $queue);
+            return new BeanstalkdJob(
+                $this->container, $this->pheanstalk, $job, $this->connectionName, $queue
+            );
         }
     }
 

diff --git a/src/Illuminate/Queue/DatabaseQueue.php b/src/Illuminate/Queue/DatabaseQueue.php
@@ -262,7 +262,7 @@ protected function marshalJob($queue, $job)
         $this->database->commit();
 
         return new DatabaseJob(
-            $this->container, $this, $job, $queue
+            $this->container, $this, $job, $this->connectionName, $queue
         );
     }
 

diff --git a/src/Illuminate/Queue/Jobs/BeanstalkdJob.php b/src/Illuminate/Queue/Jobs/BeanstalkdJob.php
@@ -29,15 +29,17 @@ class BeanstalkdJob extends Job implements JobContract
      * @param  \Illuminate\Container\Container  $container
      * @param  \Pheanstalk\Pheanstalk  $pheanstalk
      * @param  \Pheanstalk\Job  $job
+     * @param  string  $connectionName
      * @param  string  $queue
      * @return void
      */
-    public function __construct(Container $container, Pheanstalk $pheanstalk, PheanstalkJob $job, $queue)
+    public function __construct(Container $container, Pheanstalk $pheanstalk, PheanstalkJob $job, $connectionName, $queue)
     {
         $this->job = $job;
         $this->queue = $queue;
         $this->container = $container;
         $this->pheanstalk = $pheanstalk;
+        $this->connectionName = $connectionName;
     }
 
     /**

diff --git a/src/Illuminate/Queue/Jobs/DatabaseJob.php b/src/Illuminate/Queue/Jobs/DatabaseJob.php
@@ -28,15 +28,17 @@ class DatabaseJob extends Job implements JobContract
      * @param  \Illuminate\Container\Container  $container
      * @param  \Illuminate\Queue\DatabaseQueue  $database
      * @param  \StdClass  $job
+     * @param  string  $connectionName
      * @param  string  $queue
      * @return void
      */
-    public function __construct(Container $container, DatabaseQueue $database, $job, $queue)
+    public function __construct(Container $container, DatabaseQueue $database, $job, $connectionName, $queue)
     {
         $this->job = $job;
         $this->queue = $queue;
         $this->database = $database;
         $this->container = $container;
+        $this->connectionName = $connectionName;
     }
 
     /**

diff --git a/src/Illuminate/Queue/Jobs/RedisJob.php b/src/Illuminate/Queue/Jobs/RedisJob.php
@@ -44,9 +44,11 @@ class RedisJob extends Job implements JobContract
      * @param  \Illuminate\Queue\RedisQueue  $redis
      * @param  string  $job
      * @param  string  $reserved
+     * @param  string  $connectionName
      * @param  string  $queue
+     * @return void
      */
-    public function __construct(Container $container, RedisQueue $redis, $job, $reserved, $queue)
+    public function __construct(Container $container, RedisQueue $redis, $job, $reserved, $connectionName, $queue)
     {
         // The $job variable is the original job JSON as it existed in the ready queue while
         // the $reserved variable is the raw JSON in the reserved queue. The exact format
@@ -56,6 +58,8 @@ public function __construct(Container $container, RedisQueue $redis, $job, $rese
         $this->queue = $queue;
         $this->reserved = $reserved;
         $this->container = $container;
+        $this->connectionName = $connectionName;
+
         $this->decoded = $this->payload();
     }
 

diff --git a/src/Illuminate/Queue/Jobs/SqsJob.php b/src/Illuminate/Queue/Jobs/SqsJob.php
@@ -27,16 +27,18 @@ class SqsJob extends Job implements JobContract
      *
      * @param  \Illuminate\Container\Container  $container
      * @param  \Aws\Sqs\SqsClient  $sqs
-     * @param  string  $queue
      * @param  array   $job
+     * @param  string  $connectionName
+     * @param  string  $queue
      * @return void
      */
-    public function __construct(Container $container, SqsClient $sqs, $queue, array $job)
+    public function __construct(Container $container, SqsClient $sqs, array $job, $connectionName, $queue)
     {
         $this->sqs = $sqs;
         $this->job = $job;
         $this->queue = $queue;
         $this->container = $container;
+        $this->connectionName = $connectionName;
     }
 
     /**

diff --git a/src/Illuminate/Queue/Jobs/SyncJob.php b/src/Illuminate/Queue/Jobs/SyncJob.php
@@ -26,14 +26,16 @@ class SyncJob extends Job implements JobContract
      *
      * @param  \Illuminate\Container\Container  $container
      * @param  string  $payload
+     * @param  string  $connectionName
      * @param  string  $queue
      * @return void
      */
-    public function __construct(Container $container, $payload, $queue)
+    public function __construct(Container $container, $payload, $connectionName, $queue)
     {
         $this->queue = $queue;
         $this->payload = $payload;
         $this->container = $container;
+        $this->connectionName = $connectionName;
     }
 
     /**

diff --git a/src/Illuminate/Queue/Queue.php b/src/Illuminate/Queue/Queue.php
@@ -22,6 +22,13 @@ abstract class Queue
      */
     protected $encrypter;
 
+    /**
+     * The connection name for the queue.
+     *
+     * @var string
+     */
+    protected $connectionName;
+
     /**
      * Push a new job onto the queue.
      *
@@ -131,6 +138,29 @@ protected function createStringPayload($job, $data)
         return ['job' => $job, 'data' => $data];
     }
 
+    /**
+     * Get the connection name for the queue.
+     *
+     * @return string
+     */
+    public function getConnectionName()
+    {
+        return $this->connectionName;
+    }
+
+    /**
+     * Set the connection name for the queue.
+     *
+     * @param  string  $name
+     * @return $this
+     */
+    public function setConnectionName($name)
+    {
+        $this->connectionName = $name;
+
+        return $this;
+    }
+
     /**
      * Set the IoC container instance.
      *

diff --git a/src/Illuminate/Queue/QueueManager.php b/src/Illuminate/Queue/QueueManager.php
@@ -150,7 +150,9 @@ protected function resolve($name)
     {
         $config = $this->getConfig($name);
 
-        return $this->getConnector($config['driver'])->connect($config);
+        return $this->getConnector($config['driver'])
+                        ->connect($config)
+                        ->setConnectionName($name);
     }
 
     /**

diff --git a/src/Illuminate/Queue/RedisQueue.php b/src/Illuminate/Queue/RedisQueue.php
@@ -149,7 +149,7 @@ public function pop($queue = null)
         if ($reserved) {
             return new RedisJob(
                 $this->container, $this, $job,
-                $reserved, $queue ?: $this->default
+                $reserved, $this->connectionName, $queue ?: $this->default
             );
         }
     }

diff --git a/src/Illuminate/Queue/SqsQueue.php b/src/Illuminate/Queue/SqsQueue.php
@@ -117,7 +117,10 @@ public function pop($queue = null)
         ]);
 
         if (count($response['Messages']) > 0) {
-            return new SqsJob($this->container, $this->sqs, $queue, $response['Messages'][0]);
+            return new SqsJob(
+                $this->container, $this->sqs, $response['Messages'][0],
+                $this->connectionName, $queue
+            );
         }
     }
 

diff --git a/src/Illuminate/Queue/SyncQueue.php b/src/Illuminate/Queue/SyncQueue.php
@@ -60,7 +60,7 @@ public function push($job, $data = '', $queue = null)
      */
     protected function resolveJob($payload, $queue)
     {
-        return new SyncJob($this->container, $payload, $queue);
+        return new SyncJob($this->container, $payload, $this->connectionName, $queue);
     }
 
     /**
@@ -72,7 +72,7 @@ protected function resolveJob($payload, $queue)
     protected function raiseBeforeJobEvent(Job $job)
     {
         if ($this->container->bound('events')) {
-            $this->container['events']->fire(new Events\JobProcessing('sync', $job));
+            $this->container['events']->fire(new Events\JobProcessing($this->connectionName, $job));
         }
     }
 
@@ -85,7 +85,7 @@ protected function raiseBeforeJobEvent(Job $job)
     protected function raiseAfterJobEvent(Job $job)
     {
         if ($this->container->bound('events')) {
-            $this->container['events']->fire(new Events\JobProcessed('sync', $job));
+            $this->container['events']->fire(new Events\JobProcessed($this->connectionName, $job));
         }
     }
 
@@ -99,7 +99,7 @@ protected function raiseAfterJobEvent(Job $job)
     protected function raiseExceptionOccurredJobEvent(Job $job, $e)
     {
         if ($this->container->bound('events')) {
-            $this->container['events']->fire(new Events\JobExceptionOccurred('sync', $job, $e));
+            $this->container['events']->fire(new Events\JobExceptionOccurred($this->connectionName, $job, $e));
         }
     }
 
@@ -116,7 +116,7 @@ protected function handleException($queueJob, $e)
     {
         $this->raiseExceptionOccurredJobEvent($queueJob, $e);
 
-        FailingJob::handle('sync', $queueJob, $e);
+        FailingJob::handle($this->connectionName, $queueJob, $e);
 
         throw $e;
     }

diff --git a/src/Illuminate/Queue/Worker.php b/src/Illuminate/Queue/Worker.php
@@ -172,8 +172,6 @@ public function runNextJob($connectionName, $queue, WorkerOptions $options)
         // from this method. If there is no job on the queue, we will "sleep" the worker
         // for the specified number of seconds, then keep processing jobs after sleep.
         if ($job) {
-            $job->setConnectionName($connectionName);
-
             return $this->runJob($job, $connectionName, $options);
         }
 

diff --git a/tests/Queue/QueueBeanstalkdJobTest.php b/tests/Queue/QueueBeanstalkdJobTest.php
@@ -60,6 +60,7 @@ protected function getJob()
             m::mock('Illuminate\Container\Container'),
             m::mock('Pheanstalk\Pheanstalk'),
             m::mock('Pheanstalk\Job'),
+            'connection-name',
             'default'
         );
     }

diff --git a/tests/Queue/QueueManagerTest.php b/tests/Queue/QueueManagerTest.php
@@ -24,6 +24,7 @@ public function testDefaultConnectionCanBeResolved()
         $manager = new QueueManager($app);
         $connector = m::mock('StdClass');
         $queue = m::mock('StdClass');
+        $queue->shouldReceive('setConnectionName')->once()->with('sync')->andReturnSelf();
         $connector->shouldReceive('connect')->once()->with(['driver' => 'sync'])->andReturn($queue);
         $manager->addConnector('sync', function () use ($connector) {
             return $connector;
@@ -46,6 +47,7 @@ public function testOtherConnectionCanBeResolved()
         $manager = new QueueManager($app);
         $connector = m::mock('StdClass');
         $queue = m::mock('StdClass');
+        $queue->shouldReceive('setConnectionName')->once()->with('foo')->andReturnSelf();
         $connector->shouldReceive('connect')->once()->with(['driver' => 'bar'])->andReturn($queue);
         $manager->addConnector('bar', function () use ($connector) {
             return $connector;
@@ -67,6 +69,7 @@ public function testNullConnectionCanBeResolved()
         $manager = new QueueManager($app);
         $connector = m::mock('StdClass');
         $queue = m::mock('StdClass');
+        $queue->shouldReceive('setConnectionName')->once()->with('null')->andReturnSelf();
         $connector->shouldReceive('connect')->once()->with(['driver' => 'null'])->andReturn($queue);
         $manager->addConnector('null', function () use ($connector) {
             return $connector;

diff --git a/tests/Queue/QueueRedisJobTest.php b/tests/Queue/QueueRedisJobTest.php
@@ -44,6 +44,7 @@ protected function getJob()
             m::mock(Illuminate\Queue\RedisQueue::class),
             json_encode(['job' => 'foo', 'data' => ['data'], 'attempts' => 1]),
             json_encode(['job' => 'foo', 'data' => ['data'], 'attempts' => 2]),
+            'connection-name',
             'default'
         );
     }

diff --git a/tests/Queue/QueueSqsJobTest.php b/tests/Queue/QueueSqsJobTest.php
@@ -87,8 +87,9 @@ protected function getJob()
         return new Illuminate\Queue\Jobs\SqsJob(
             $this->mockedContainer,
             $this->mockedSqsClient,
-            $this->queueUrl,
-            $this->mockedJobData
+            $this->mockedJobData,
+            'connection-name',
+            $this->queueUrl
         );
     }
 }