[8.x] Allow including a closure in a queued batch (#34333)

* allow including a closure in a queued batch

* fix style

* fix style

src/Illuminate/Bus/Batch.php

@@ -3,8 +3,10 @@
 namespace Illuminate\Bus;
 
 use Carbon\CarbonImmutable;
+use Closure;
 use Illuminate\Contracts\Queue\Factory as QueueFactory;
 use Illuminate\Contracts\Support\Arrayable;
+use Illuminate\Queue\CallQueuedClosure;
 use Illuminate\Queue\SerializableClosure;
 use Illuminate\Support\Arr;
 use Illuminate\Support\Collection;
@@ -159,9 +161,15 @@ public function fresh()
      */
     public function add($jobs)
     {
-        $jobs = Collection::wrap($jobs);
+        $jobs = Collection::wrap($jobs)->map(function ($job) {
+            if ($job instanceof Closure) {
+                $job = CallQueuedClosure::create($job);
+            }
 
-        $jobs->each->withBatchId($this->id);
+            $job->withBatchId($this->id);
+
+            return $job;
+        });
 
         $this->repository->transaction(function () use ($jobs) {
             $this->repository->incrementTotalJobs($this->id, count($jobs));

src/Illuminate/Queue/CallQueuedClosure.php

@@ -4,6 +4,7 @@
 
 use Closure;
 use Exception;
+use Illuminate\Bus\Batchable;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Container\Container;
 use Illuminate\Contracts\Queue\ShouldQueue;
@@ -12,7 +13,7 @@
 
 class CallQueuedClosure implements ShouldQueue
 {
-    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+    use Batchable, Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
 
     /**
      * The serializable Closure instance.

tests/Bus/BusBatchTest.php

@@ -12,6 +12,7 @@
 use Illuminate\Contracts\Queue\Factory;
 use Illuminate\Database\Capsule\Manager as DB;
 use Illuminate\Database\Eloquent\Model;
+use Illuminate\Queue\CallQueuedClosure;
 use Mockery as m;
 use PHPUnit\Framework\TestCase;
 use RuntimeException;
@@ -89,16 +90,25 @@ public function test_jobs_can_be_added_to_the_batch()
             use Batchable;
         };
 
+        $thirdJob = function () {
+        };
+
         $queue->shouldReceive('connection')->once()
                         ->with('test-connection')
                         ->andReturn($connection = m::mock(stdClass::class));
 
-        $connection->shouldReceive('bulk')->once()->with([$job, $secondJob], '', 'test-queue');
+        $connection->shouldReceive('bulk')->once()->with(\Mockery::on(function ($args) use ($job, $secondJob) {
+            return
+                $args[0] == $job &&
+                $args[1] == $secondJob &&
+                $args[2] instanceof CallQueuedClosure
+                && is_string($args[2]->batchId);
+        }), '', 'test-queue');
 
-        $batch = $batch->add([$job, $secondJob]);
+        $batch = $batch->add([$job, $secondJob, $thirdJob]);
 
-        $this->assertEquals(2, $batch->totalJobs);
-        $this->assertEquals(2, $batch->pendingJobs);
+        $this->assertEquals(3, $batch->totalJobs);
+        $this->assertEquals(3, $batch->pendingJobs);
         $this->assertTrue(is_string($job->batchId));
         $this->assertInstanceOf(CarbonImmutable::class, $batch->createdAt);
     }