Batch.php 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502
  1. <?php
  2. namespace Illuminate\Bus;
  3. use Carbon\CarbonImmutable;
  4. use Closure;
  5. use Illuminate\Bus\Events\BatchCanceled;
  6. use Illuminate\Bus\Events\BatchFinished;
  7. use Illuminate\Container\Container;
  8. use Illuminate\Contracts\Events\Dispatcher;
  9. use Illuminate\Contracts\Queue\Factory as QueueFactory;
  10. use Illuminate\Contracts\Support\Arrayable;
  11. use Illuminate\Queue\CallQueuedClosure;
  12. use Illuminate\Support\Arr;
  13. use Illuminate\Support\Collection;
  14. use JsonSerializable;
  15. use Throwable;
  16. class Batch implements Arrayable, JsonSerializable
  17. {
  18. /**
  19. * The queue factory implementation.
  20. *
  21. * @var \Illuminate\Contracts\Queue\Factory
  22. */
  23. protected $queue;
  24. /**
  25. * The repository implementation.
  26. *
  27. * @var \Illuminate\Bus\BatchRepository
  28. */
  29. protected $repository;
  30. /**
  31. * The batch ID.
  32. *
  33. * @var string
  34. */
  35. public $id;
  36. /**
  37. * The batch name.
  38. *
  39. * @var string
  40. */
  41. public $name;
  42. /**
  43. * The total number of jobs that belong to the batch.
  44. *
  45. * @var int
  46. */
  47. public $totalJobs;
  48. /**
  49. * The total number of jobs that are still pending.
  50. *
  51. * @var int
  52. */
  53. public $pendingJobs;
  54. /**
  55. * The total number of jobs that have failed.
  56. *
  57. * @var int
  58. */
  59. public $failedJobs;
  60. /**
  61. * The IDs of the jobs that have failed.
  62. *
  63. * @var array
  64. */
  65. public $failedJobIds;
  66. /**
  67. * The batch options.
  68. *
  69. * @var array
  70. */
  71. public $options;
  72. /**
  73. * The date indicating when the batch was created.
  74. *
  75. * @var \Carbon\CarbonImmutable
  76. */
  77. public $createdAt;
  78. /**
  79. * The date indicating when the batch was cancelled.
  80. *
  81. * @var \Carbon\CarbonImmutable|null
  82. */
  83. public $cancelledAt;
  84. /**
  85. * The date indicating when the batch was finished.
  86. *
  87. * @var \Carbon\CarbonImmutable|null
  88. */
  89. public $finishedAt;
  90. /**
  91. * Create a new batch instance.
  92. */
  93. public function __construct(
  94. QueueFactory $queue,
  95. BatchRepository $repository,
  96. string $id,
  97. string $name,
  98. int $totalJobs,
  99. int $pendingJobs,
  100. int $failedJobs,
  101. array $failedJobIds,
  102. array $options,
  103. CarbonImmutable $createdAt,
  104. ?CarbonImmutable $cancelledAt = null,
  105. ?CarbonImmutable $finishedAt = null,
  106. ) {
  107. $this->queue = $queue;
  108. $this->repository = $repository;
  109. $this->id = $id;
  110. $this->name = $name;
  111. $this->totalJobs = $totalJobs;
  112. $this->pendingJobs = $pendingJobs;
  113. $this->failedJobs = $failedJobs;
  114. $this->failedJobIds = $failedJobIds;
  115. $this->options = $options;
  116. $this->createdAt = $createdAt;
  117. $this->cancelledAt = $cancelledAt;
  118. $this->finishedAt = $finishedAt;
  119. }
  120. /**
  121. * Get a fresh instance of the batch represented by this ID.
  122. *
  123. * @return self
  124. */
  125. public function fresh()
  126. {
  127. return $this->repository->find($this->id);
  128. }
  129. /**
  130. * Add additional jobs to the batch.
  131. *
  132. * @param \Illuminate\Support\Enumerable|object|array $jobs
  133. * @return self
  134. */
  135. public function add($jobs)
  136. {
  137. $count = 0;
  138. $jobs = Collection::wrap($jobs)->map(function ($job) use (&$count) {
  139. $job = $job instanceof Closure ? CallQueuedClosure::create($job) : $job;
  140. if (is_array($job)) {
  141. $count += count($job);
  142. $chain = $this->prepareBatchedChain($job);
  143. return $chain->first()
  144. ->allOnQueue($this->options['queue'] ?? null)
  145. ->allOnConnection($this->options['connection'] ?? null)
  146. ->chain($chain->slice(1)->values()->all());
  147. } else {
  148. $job->withBatchId($this->id);
  149. $count++;
  150. }
  151. return $job;
  152. });
  153. $this->repository->transaction(function () use ($jobs, $count) {
  154. $this->repository->incrementTotalJobs($this->id, $count);
  155. $this->queue->connection($this->options['connection'] ?? null)->bulk(
  156. $jobs->all(),
  157. $data = '',
  158. $this->options['queue'] ?? null
  159. );
  160. });
  161. return $this->fresh();
  162. }
  163. /**
  164. * Prepare a chain that exists within the jobs being added.
  165. *
  166. * @return \Illuminate\Support\Collection
  167. */
  168. protected function prepareBatchedChain(array $chain)
  169. {
  170. return (new Collection($chain))->map(function ($job) {
  171. $job = $job instanceof Closure ? CallQueuedClosure::create($job) : $job;
  172. return $job->withBatchId($this->id);
  173. });
  174. }
  175. /**
  176. * Get the total number of jobs that have been processed by the batch thus far.
  177. *
  178. * @return int
  179. */
  180. public function processedJobs()
  181. {
  182. return $this->totalJobs - $this->pendingJobs;
  183. }
  184. /**
  185. * Get the percentage of jobs that have been processed (between 0-100).
  186. *
  187. * @return int<0, 100>
  188. */
  189. public function progress()
  190. {
  191. return $this->totalJobs > 0 ? (int) round(($this->processedJobs() / $this->totalJobs) * 100) : 0;
  192. }
  193. /**
  194. * Record that a job within the batch finished successfully, executing any callbacks if necessary.
  195. *
  196. * @return void
  197. */
  198. public function recordSuccessfulJob(string $jobId)
  199. {
  200. $counts = $this->decrementPendingJobs($jobId);
  201. if ($this->hasProgressCallbacks()) {
  202. $this->invokeCallbacks('progress');
  203. }
  204. if ($counts->pendingJobs === 0) {
  205. $this->repository->markAsFinished($this->id);
  206. $container = Container::getInstance();
  207. if ($container->bound(Dispatcher::class)) {
  208. $container->make(Dispatcher::class)->dispatch(new BatchFinished($this));
  209. }
  210. }
  211. if ($counts->pendingJobs === 0 && $this->hasThenCallbacks()) {
  212. $this->invokeCallbacks('then');
  213. }
  214. if ($counts->allJobsHaveRanExactlyOnce() && $this->hasFinallyCallbacks()) {
  215. $this->invokeCallbacks('finally');
  216. }
  217. }
  218. /**
  219. * Decrement the pending jobs for the batch.
  220. *
  221. * @return \Illuminate\Bus\UpdatedBatchJobCounts
  222. */
  223. public function decrementPendingJobs(string $jobId)
  224. {
  225. return $this->repository->decrementPendingJobs($this->id, $jobId);
  226. }
  227. /**
  228. * Invoke the callbacks of the given type.
  229. */
  230. protected function invokeCallbacks(string $type, ?Throwable $e = null): void
  231. {
  232. $batch = $this->fresh();
  233. foreach ($this->options[$type] ?? [] as $handler) {
  234. $this->invokeHandlerCallback($handler, $batch, $e);
  235. }
  236. }
  237. /**
  238. * Determine if the batch has finished executing.
  239. *
  240. * @return bool
  241. */
  242. public function finished()
  243. {
  244. return ! is_null($this->finishedAt);
  245. }
  246. /**
  247. * Determine if the batch has "progress" callbacks.
  248. *
  249. * @return bool
  250. */
  251. public function hasProgressCallbacks()
  252. {
  253. return isset($this->options['progress']) && ! empty($this->options['progress']);
  254. }
  255. /**
  256. * Determine if the batch has "success" callbacks.
  257. *
  258. * @return bool
  259. */
  260. public function hasThenCallbacks()
  261. {
  262. return isset($this->options['then']) && ! empty($this->options['then']);
  263. }
  264. /**
  265. * Determine if the batch allows jobs to fail without cancelling the batch.
  266. *
  267. * @return bool
  268. */
  269. public function allowsFailures()
  270. {
  271. return Arr::get($this->options, 'allowFailures', false) === true;
  272. }
  273. /**
  274. * Determine if the batch has job failures.
  275. *
  276. * @return bool
  277. */
  278. public function hasFailures()
  279. {
  280. return $this->failedJobs > 0;
  281. }
  282. /**
  283. * Record that a job within the batch failed to finish successfully, executing any callbacks if necessary.
  284. *
  285. * @param \Throwable $e
  286. * @return void
  287. */
  288. public function recordFailedJob(string $jobId, $e)
  289. {
  290. $counts = $this->incrementFailedJobs($jobId);
  291. if ($counts->failedJobs === 1 && ! $this->allowsFailures()) {
  292. $this->cancel();
  293. }
  294. if ($this->allowsFailures()) {
  295. if ($this->hasProgressCallbacks()) {
  296. $this->invokeCallbacks('progress', $e);
  297. }
  298. if ($this->hasFailureCallbacks()) {
  299. $this->invokeCallbacks('failure', $e);
  300. }
  301. }
  302. if ($counts->failedJobs === 1 && $this->hasCatchCallbacks()) {
  303. $this->invokeCallbacks('catch', $e);
  304. }
  305. if ($counts->allJobsHaveRanExactlyOnce() && $this->hasFinallyCallbacks()) {
  306. $this->invokeCallbacks('finally');
  307. }
  308. }
  309. /**
  310. * Increment the failed jobs for the batch.
  311. *
  312. * @return \Illuminate\Bus\UpdatedBatchJobCounts
  313. */
  314. public function incrementFailedJobs(string $jobId)
  315. {
  316. return $this->repository->incrementFailedJobs($this->id, $jobId);
  317. }
  318. /**
  319. * Determine if the batch has "catch" callbacks.
  320. *
  321. * @return bool
  322. */
  323. public function hasCatchCallbacks()
  324. {
  325. return isset($this->options['catch']) && ! empty($this->options['catch']);
  326. }
  327. /**
  328. * Determine if the batch has "failure" callbacks.
  329. */
  330. public function hasFailureCallbacks(): bool
  331. {
  332. return isset($this->options['failure']) && ! empty($this->options['failure']);
  333. }
  334. /**
  335. * Determine if the batch has "finally" callbacks.
  336. *
  337. * @return bool
  338. */
  339. public function hasFinallyCallbacks()
  340. {
  341. return isset($this->options['finally']) && ! empty($this->options['finally']);
  342. }
  343. /**
  344. * Cancel the batch.
  345. *
  346. * @return void
  347. */
  348. public function cancel()
  349. {
  350. $this->repository->cancel($this->id);
  351. $container = Container::getInstance();
  352. if ($container->bound(Dispatcher::class)) {
  353. $container->make(Dispatcher::class)->dispatch(new BatchCanceled($this));
  354. }
  355. }
  356. /**
  357. * Determine if the batch has been cancelled.
  358. *
  359. * @return bool
  360. */
  361. public function canceled()
  362. {
  363. return $this->cancelled();
  364. }
  365. /**
  366. * Determine if the batch has been cancelled.
  367. *
  368. * @return bool
  369. */
  370. public function cancelled()
  371. {
  372. return ! is_null($this->cancelledAt);
  373. }
  374. /**
  375. * Delete the batch from storage.
  376. *
  377. * @return void
  378. */
  379. public function delete()
  380. {
  381. $this->repository->delete($this->id);
  382. }
  383. /**
  384. * Invoke a batch callback handler.
  385. *
  386. * @param callable $handler
  387. * @return void
  388. */
  389. protected function invokeHandlerCallback($handler, Batch $batch, ?Throwable $e = null)
  390. {
  391. try {
  392. $handler($batch, $e);
  393. } catch (Throwable $e) {
  394. if (function_exists('report')) {
  395. report($e);
  396. }
  397. }
  398. }
  399. /**
  400. * Convert the batch to an array.
  401. *
  402. * @return array
  403. */
  404. public function toArray()
  405. {
  406. return [
  407. 'id' => $this->id,
  408. 'name' => $this->name,
  409. 'totalJobs' => $this->totalJobs,
  410. 'pendingJobs' => $this->pendingJobs,
  411. 'processedJobs' => $this->processedJobs(),
  412. 'progress' => $this->progress(),
  413. 'failedJobs' => $this->failedJobs,
  414. 'options' => $this->options,
  415. 'createdAt' => $this->createdAt,
  416. 'cancelledAt' => $this->cancelledAt,
  417. 'finishedAt' => $this->finishedAt,
  418. ];
  419. }
  420. /**
  421. * Get the JSON serializable representation of the object.
  422. */
  423. public function jsonSerialize(): array
  424. {
  425. return $this->toArray();
  426. }
  427. /**
  428. * Dynamically access the batch's "options" via properties.
  429. *
  430. * @param string $key
  431. * @return mixed
  432. */
  433. public function __get($key)
  434. {
  435. return $this->options[$key] ?? null;
  436. }
  437. }