[8.x] Job Batching

composer.json

@@ -26,7 +26,7 @@
         "league/flysystem": "^1.0.8",
         "monolog/monolog": "^2.0",
         "nesbot/carbon": "^2.17",
-        "opis/closure": "^3.1",
+        "opis/closure": "^3.5",
         "psr/container": "^1.0",
         "psr/simple-cache": "^1.0",
         "ramsey/uuid": "^4.0",

src/Illuminate/Bus/Batch.php

@@ -0,0 +1,380 @@
+<?php
+
+namespace Illuminate\Bus;
+
+use Carbon\CarbonImmutable;
+use Illuminate\Collections\Arr;
+use Illuminate\Collections\Collection;
+use Illuminate\Contracts\Queue\Factory as QueueFactory;
+use JsonSerializable;
+
+class Batch implements JsonSerializable
+{
+    /**
+     * The queue factory implementation.
+     *
+     * @var \Illuminate\Contracts\Queue\Factory
+     */
+    protected $queue;
+
+    /**
+     * The repository implementation.
+     *
+     * @var \Illuminate\Bus\BatchRepository
+     */
+    protected $repository;
+
+    /**
+     * The batch ID.
+     *
+     * @var string
+     */
+    public $id;
+
+    /**
+     * The total number of jobs that belong to the batch.
+     *
+     * @var int
+     */
+    public $totalJobs;
+
+    /**
+     * The total number of jobs that are still pending.
+     *
+     * @var int
+     */
+    public $pendingJobs;
+
+    /**
+     * The total number of jobs that have failed.
+     *
+     * @var int
+     */
+    public $failedJobs;
+
+    /**
+     * The IDs of the jobs that have failed.
+     *
+     * @var array
+     */
+    public $failedJobIds;
+
+    /**
+     * The batch options.
+     *
+     * @var array
+     */
+    public $options;
+
+    /**
+     * The date indicating when the batch was created.
+     *
+     * @var \Illuminate\Support\CarbonImmutable
+     */
+    public $createdAt;
+
+    /**
+     * The date indicating when the batch was cancelled.
+     *
+     * @var \Illuminate\Support\CarbonImmutable|null
+     */
+    public $cancelledAt;
+
+    /**
+     * The date indicating when the batch was finished.
+     *
+     * @var \Illuminate\Support\CarbonImmutable|null
+     */
+    public $finishedAt;
+
+    /**
+     * Create a new batch instance.
+     *
+     * @param  \Illuminate\Contracts\Queue\Factory  $queue
+     * @param  \Illuminate\Bus\BatchRepository  $repository
+     * @param  string  $id
+     * @param  int  $totalJobs
+     * @param  int  $pendingJobs
+     * @param  int  $failedJobs
+     * @param  array  $failedJobIds
+     * @param  array  $options
+     * @param  \Illuminate\Support\CarbonImmutable  $createdAt
+     * @param  \Illuminate\Support\CarbonImmutable|null  $cancelledAt
+     * @param  \Illuminate\Support\CarbonImmutable|null  $finishedAt
+     * @return void
+     */
+    public function __construct(QueueFactory $queue,
+                                BatchRepository $repository,
+                                string $id,
+                                int $totalJobs,
+                                int $pendingJobs,
+                                int $failedJobs,
+                                array $failedJobIds,
+                                array $options,
+                                CarbonImmutable $createdAt,
+                                ?CarbonImmutable $cancelledAt = null,
+                                ?CarbonImmutable $finishedAt = null)
+    {
+        $this->queue = $queue;
+        $this->repository = $repository;
+        $this->id = $id;
+        $this->totalJobs = $totalJobs;
+        $this->pendingJobs = $pendingJobs;
+        $this->failedJobs = $failedJobs;
+        $this->failedJobIds = $failedJobIds;
+        $this->options = $options;
+        $this->createdAt = $createdAt;
+        $this->cancelledAt = $cancelledAt;
+        $this->finishedAt = $finishedAt;
+    }
+
+    /**
+     * Get a fresh instance of the batch represented by this ID.
+     *
+     * @return self
+     */
+    public function fresh()
+    {
+        return $this->repository->find($this->id);
+    }
+
+    /**
+     * Add additional jobs to the batch.
+     *
+     * @param  \Illuminate\Collections\Collection|array  $jobs
+     * @return self
+     */
+    public function add($jobs)
+    {
+        $jobs = Collection::wrap($jobs);
+
+        $jobs->each->withBatchId($this->id);
+
+        $this->repository->transaction(function () use ($jobs) {
+            $this->repository->incrementTotalJobs($this->id, count($jobs));
+
+            $this->queue->connection($this->options['connection'] ?? null)->bulk(
+                $jobs->all(),
+                $data = '',
+                $this->options['queue'] ?? null
+            );
+        });
+
+        return $this->fresh();
+    }
+
+    /**
+     * Get the total number of jobs that have been processed by the batch thus far.
+     */
+    public function processedJobs()
+    {
+        return $this->totalJobs - $this->pendingJobs;
+    }
+
+    /**
+     * Get the percentage of jobs that have been processed.
+     *
+     * @return float
+     */
+    public function progress()
+    {
+        return $this->totalJobs > 0 ? round($this->processedJobs() / $this->totalJobs, 2) : 0;
+    }
+
+    /**
+     * Record that a job within the batch finished successfully, executing any callbacks if necessary.
+     *
+     * @param  string  $jobId
+     * @return void
+     */
+    public function recordSuccessfulJob(string $jobId)
+    {
+        $counts = $this->decrementPendingJobs($jobId);
+
+        if ($counts->pendingJobs === 0) {
+            $this->repository->markAsFinished($this->id);
+        }
+
+        if ($counts->pendingJobs === 0 && $this->hasSuccessCallbacks()) {
+            $batch = $this->fresh();
+
+            collect($this->options['success'])->each->__invoke($batch);
+        }
+
+        if ($counts->allJobsHaveRanExactlyOnce() && $this->hasThenCallbacks()) {
+            $batch = $this->fresh();
+
+            collect($this->options['then'])->each->__invoke($batch);
+        }
+    }
+
+    /**
+     * Decrement the pending jobs for the batch.
+     *
+     * @param  string  $jobId
+     * @return int
+     */
+    public function decrementPendingJobs(string $jobId)
+    {
+        return $this->repository->decrementPendingJobs($this->id, $jobId);
+    }
+
+    /**
+     * Determine if the batch has finished executing.
+     *
+     * @return bool
+     */
+    public function finished()
+    {
+        return ! is_null($this->finishedAt);
+    }
+
+    /**
+     * Determine if the batch has "success" callbacks.
+     *
+     * @return bool
+     */
+    public function hasSuccessCallbacks()
+    {
+        return isset($this->options['success']) && ! empty($this->options['success']);
+    }
+
+    /**
+     * Determine if the batch allows jobs to fail without cancelling the batch.
+     *
+     * @return bool
+     */
+    public function allowsFailures()
+    {
+        return Arr::get($this->options, 'allowFailures', false) === true;
+    }
+
+    /**
+     * Determine if the batch has job failures.
+     *
+     * @return bool
+     */
+    public function hasFailures()
+    {
+        return $this->failedJobs > 0;
+    }
+
+    /**
+     * Record that a job within the batch failed to finish successfully, executing any callbacks if necessary.
+     *
+     * @param  string  $jobId
+     * @param  \Throwable  $e
+     * @return void
+     */
+    public function recordFailedJob(string $jobId, $e)
+    {
+        $counts = $this->incrementFailedJobs($jobId);
+
+        if ($counts->failedJobs === 1 && ! $this->allowsFailures()) {
+            $this->cancel();
+        }
+
+        if ($counts->failedJobs === 1 && $this->hasCatchCallbacks()) {
+            $batch = $this->fresh();
+
+            collect($this->options['catch'])->each->__invoke($batch, $e);
+        }
+
+        if ($counts->allJobsHaveRanExactlyOnce() && $this->hasThenCallbacks()) {
+            $batch = $this->fresh();
+
+            collect($this->options['then'])->each->__invoke($batch, $e);
+        }
+    }
+
+    /**
+     * Increment the failed jobs for the batch.
+     *
+     * @param  string  $jobId
+     * @return int
+     */
+    public function incrementFailedJobs(string $jobId)
+    {
+        return $this->repository->incrementFailedJobs($this->id, $jobId);
+    }
+
+    /**
+     * Determine if the batch has "catch" callbacks.
+     *
+     * @return bool
+     */
+    public function hasCatchCallbacks()
+    {
+        return isset($this->options['catch']) && ! empty($this->options['catch']);
+    }
+
+    /**
+     * Determine if the batch has "then" callbacks.
+     *
+     * @return bool
+     */
+    public function hasThenCallbacks()
+    {
+        return isset($this->options['then']) && ! empty($this->options['then']);
+    }
+
+    /**
+     * Cancel the batch.
+     *
+     * @return void
+     */
+    public function cancel()
+    {
+        $this->repository->cancel($this->id);
+    }
+
+    /**
+     * Determine if the batch has been cancelled.
+     *
+     * @return bool
+     */
+    public function canceled()
+    {
+        return $this->cancelled();
+    }
+
+    /**
+     * Determine if the batch has been cancelled.
+     *
+     * @return bool
+     */
+    public function cancelled()
+    {
+        return ! is_null($this->cancelledAt);
+    }
+
+    /**
+     * Delete the batch from storage.
+     *
+     * @return void
+     */
+    public function delete()
+    {
+        $this->repository->delete($this->id);
+    }
+
+    /**
+     * Get the JSON serializable representation of the object.
+     *
+     * @return array
+     */
+    public function jsonSerialize()
+    {
+        return [
+            'id' => $this->id,
+            'totalJobs' => $this->totalJobs,
+            'pendingJobs' => $this->pendingJobs,
+            'processedJobs' => $this->processedJobs(),
+            'progress' => $this->progress(),
+            'failedJobs' => $this->failedJobs,
+            'createdAt' => $this->createdAt,
+            'cancelledAt' => $this->cancelledAt,
+            'finishedAt' => $this->finishedAt,
+        ];
+    }
+}