Speed up Psalm analysis of activity and workflow jobs

src/Activity.php

@@ -57,19 +57,12 @@ public function __construct(
         $options = $this->storedWorkflow->workflowOptions();
         $connection = $options->connection;
 
-        if ($connection !== null) {
-            $this->onConnection($connection);
-        } elseif (property_exists($this, 'connection')) {
-            $this->onConnection($this->connection);
-        }
+        // Give Psalm a base-typed receiver so Queueable is analyzed once per job base class.
+        self::initializeConnection($this, $connection);
 
         $queue = $options->queue;
 
-        if ($queue !== null) {
-            $this->onQueue($queue);
-        } elseif (property_exists($this, 'queue')) {
-            $this->onQueue($this->queue);
-        }
+        self::initializeQueue($this, $queue);
 
         $this->afterCommit = true;
     }
@@ -182,4 +175,22 @@ public function heartbeat(): void
             Cache::put($this->key, 1, $this->timeout);
         }
     }
+
+    private static function initializeConnection(self $activity, $connection): void
+    {
+        if ($connection !== null) {
+            $activity->onConnection($connection);
+        } elseif (property_exists($activity, 'connection')) {
+            $activity->onConnection($activity->connection);
+        }
+    }
+
+    private static function initializeQueue(self $activity, $queue): void
+    {
+        if ($queue !== null) {
+            $activity->onQueue($queue);
+        } elseif (property_exists($activity, 'queue')) {
+            $activity->onQueue($activity->queue);
+        }
+    }
 }

src/Workflow.php

@@ -79,19 +79,12 @@ public function __construct(
 
         $connection = $this->storedWorkflow->effectiveConnection();
 
-        if ($connection !== null) {
-            $this->onConnection($connection);
-        } elseif (property_exists($this, 'connection')) {
-            $this->onConnection($this->connection);
-        }
+        // Give Psalm a base-typed receiver so Queueable is analyzed once per job base class.
+        self::initializeConnection($this, $connection);
 
         $queue = $this->storedWorkflow->effectiveQueue();
 
-        if ($queue !== null) {
-            $this->onQueue($queue);
-        } elseif (property_exists($this, 'queue')) {
-            $this->onQueue($this->queue);
-        }
+        self::initializeQueue($this, $queue);
 
         $this->afterCommit = true;
     }
@@ -314,6 +307,24 @@ public function handle(): void
         }
     }
 
+    private static function initializeConnection(self $workflow, $connection): void
+    {
+        if ($connection !== null) {
+            $workflow->onConnection($connection);
+        } elseif (property_exists($workflow, 'connection')) {
+            $workflow->onConnection($workflow->connection);
+        }
+    }
+
+    private static function initializeQueue(self $workflow, $queue): void
+    {
+        if ($queue !== null) {
+            $workflow->onQueue($queue);
+        } elseif (property_exists($workflow, 'queue')) {
+            $workflow->onQueue($workflow->queue);
+        }
+    }
+
     private function setContext(array $context): void
     {
         $existingContext = WorkflowStub::getContext();

tests/Unit/ActivityTest.php

@@ -6,19 +6,22 @@
 
 use BadMethodCallException;
 use Exception;
+use Mockery;
 use Tests\Fixtures\TestExceptionActivity;
 use Tests\Fixtures\TestInvalidActivity;
 use Tests\Fixtures\TestNonRetryableExceptionActivity;
 use Tests\Fixtures\TestNullableArgumentsActivity;
 use Tests\Fixtures\TestOtherActivity;
 use Tests\Fixtures\TestWorkflow;
 use Tests\TestCase;
+use Workflow\Activity;
 use Workflow\Exceptions\NonRetryableException;
 use Workflow\Models\StoredWorkflow;
 use Workflow\Serializers\Serializer;
 use Workflow\States\WorkflowCreatedStatus;
 use Workflow\States\WorkflowFailedStatus;
 use Workflow\Webhooks;
+use Workflow\WorkflowOptions;
 use Workflow\WorkflowStub;
 
 final class ActivityTest extends TestCase
@@ -40,22 +43,61 @@ public function testActivity(): void
         $this->assertSame($activity->timeout, pcntl_alarm(0));
     }
 
-    public function testActivityUsesWorkflowOptionConnection(): void
+    public function testActivityUsesWorkflowConnectionAndQueueOptions(): void
     {
         $workflow = WorkflowStub::load(WorkflowStub::make(TestWorkflow::class)->id());
         $storedWorkflow = StoredWorkflow::findOrFail($workflow->id());
         $storedWorkflow->arguments = Serializer::serialize([
             'arguments' => [],
             'options' => [
                 'connection' => 'sync',
-                'queue' => null,
+                'queue' => 'high',
             ],
         ]);
         $storedWorkflow->save();
 
         $activity = new TestOtherActivity(0, now()->toDateTimeString(), $storedWorkflow, ['other']);
 
         $this->assertSame('sync', $activity->connection);
+        $this->assertSame('high', $activity->queue);
+    }
+
+    public function testActivityPreservesDefaultConnectionAndQueueWhenWorkflowOptionsAreNull(): void
+    {
+        $storedWorkflow = Mockery::mock(StoredWorkflow::class);
+        $storedWorkflow->shouldReceive('workflowOptions')
+            ->once()
+            ->andReturn(new WorkflowOptions());
+
+        $activity = new class(0, now()->toDateTimeString(), $storedWorkflow) extends Activity {
+            public $connection = 'redis';
+
+            public $queue = 'default';
+
+            public array $routingCalls = [];
+
+            public function onConnection($connection)
+            {
+                $this->routingCalls[] = ['connection', $connection];
+
+                return parent::onConnection($connection);
+            }
+
+            public function onQueue($queue)
+            {
+                $this->routingCalls[] = ['queue', $queue];
+
+                return parent::onQueue($queue);
+            }
+
+            public function execute(): void
+            {
+            }
+        };
+
+        $this->assertSame('redis', $activity->connection);
+        $this->assertSame('default', $activity->queue);
+        $this->assertSame([['connection', 'redis'], ['queue', 'default']], $activity->routingCalls);
     }
 
     public function testInvalidActivity(): void

tests/Unit/WorkflowTest.php

@@ -334,20 +334,66 @@ public function testParentPending(): void
         $this->assertSame('other', $childWorkflow->output());
     }
 
-    public function testConstructorUsesQueuePropertyWhenEffectiveQueueIsNull(): void
-    {
+    /**
+     * @dataProvider routingProvider
+     */
+    public function testConstructorInitializesRouting(
+        ?string $effectiveConnection,
+        ?string $effectiveQueue,
+        ?string $classConnection,
+        ?string $classQueue,
+        ?string $expectedConnection,
+        ?string $expectedQueue
+    ): void {
         $storedWorkflow = Mockery::mock(StoredWorkflow::class);
         $storedWorkflow->shouldReceive('effectiveConnection')
             ->once()
-            ->andReturn('sync');
+            ->andReturn($effectiveConnection);
         $storedWorkflow->shouldReceive('effectiveQueue')
             ->once()
-            ->andReturn(null);
+            ->andReturn($effectiveQueue);
 
-        $workflow = new Workflow($storedWorkflow);
+        $workflow = new class($storedWorkflow, $classConnection, $classQueue) extends Workflow {
+            public array $routingCalls = [];
+
+            public function __construct(StoredWorkflow $storedWorkflow, ?string $connection, ?string $queue)
+            {
+                $this->connection = $connection;
+                $this->queue = $queue;
+
+                parent::__construct($storedWorkflow);
+            }
+
+            public function onConnection($connection)
+            {
+                $this->routingCalls[] = ['connection', $connection];
+
+                return parent::onConnection($connection);
+            }
 
-        $this->assertSame('sync', $workflow->connection);
-        $this->assertNull($workflow->queue);
+            public function onQueue($queue)
+            {
+                $this->routingCalls[] = ['queue', $queue];
+
+                return parent::onQueue($queue);
+            }
+        };
+
+        $this->assertSame($expectedConnection, $workflow->connection);
+        $this->assertSame($expectedQueue, $workflow->queue);
+        $this->assertSame([
+            ['connection', $expectedConnection],
+            ['queue', $expectedQueue],
+        ], $workflow->routingCalls);
+    }
+
+    public static function routingProvider(): array
+    {
+        return [
+            'effective routing' => ['sync', 'high', null, null, 'sync', 'high'],
+            'mixed routing' => ['sync', null, null, null, 'sync', null],
+            'class defaults' => [null, null, 'redis', 'default', 'redis', 'default'],
+        ];
     }
 
     public function testThrowsWhenExecuteMethodIsMissing(): void