Queueable.php 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386
  1. <?php
  2. namespace Illuminate\Bus;
  3. use Closure;
  4. use Illuminate\Queue\CallQueuedClosure;
  5. use Illuminate\Support\Arr;
  6. use Illuminate\Support\Collection;
  7. use Laravel\SerializableClosure\SerializableClosure;
  8. use PHPUnit\Framework\Assert as PHPUnit;
  9. use RuntimeException;
  10. use function Illuminate\Support\enum_value;
  11. trait Queueable
  12. {
  13. /**
  14. * The name of the connection the job should be sent to.
  15. *
  16. * @var string|null
  17. */
  18. public $connection;
  19. /**
  20. * The name of the queue the job should be sent to.
  21. *
  22. * @var string|null
  23. */
  24. public $queue;
  25. /**
  26. * The job "group" the job should be sent to.
  27. *
  28. * @var string|null
  29. */
  30. public $messageGroup;
  31. /**
  32. * The job deduplicator callback the job should use to generate the deduplication ID.
  33. *
  34. * @var \Laravel\SerializableClosure\SerializableClosure|null
  35. */
  36. public $deduplicator;
  37. /**
  38. * The number of seconds before the job should be made available.
  39. *
  40. * @var \DateTimeInterface|\DateInterval|array|int|null
  41. */
  42. public $delay;
  43. /**
  44. * Indicates whether the job should be dispatched after all database transactions have committed.
  45. *
  46. * @var bool|null
  47. */
  48. public $afterCommit;
  49. /**
  50. * The middleware the job should be dispatched through.
  51. *
  52. * @var array
  53. */
  54. public $middleware = [];
  55. /**
  56. * The jobs that should run if this job is successful.
  57. *
  58. * @var array
  59. */
  60. public $chained = [];
  61. /**
  62. * The name of the connection the chain should be sent to.
  63. *
  64. * @var string|null
  65. */
  66. public $chainConnection;
  67. /**
  68. * The name of the queue the chain should be sent to.
  69. *
  70. * @var string|null
  71. */
  72. public $chainQueue;
  73. /**
  74. * The callbacks to be executed on chain failure.
  75. *
  76. * @var array|null
  77. */
  78. public $chainCatchCallbacks;
  79. /**
  80. * Set the desired connection for the job.
  81. *
  82. * @param \UnitEnum|string|null $connection
  83. * @return $this
  84. */
  85. public function onConnection($connection)
  86. {
  87. $this->connection = enum_value($connection);
  88. return $this;
  89. }
  90. /**
  91. * Set the desired queue for the job.
  92. *
  93. * @param \UnitEnum|string|null $queue
  94. * @return $this
  95. */
  96. public function onQueue($queue)
  97. {
  98. $this->queue = enum_value($queue);
  99. return $this;
  100. }
  101. /**
  102. * Set the desired job "group".
  103. *
  104. * This feature is only supported by some queues, such as Amazon SQS.
  105. *
  106. * @param \UnitEnum|string $group
  107. * @return $this
  108. */
  109. public function onGroup($group)
  110. {
  111. $this->messageGroup = enum_value($group);
  112. return $this;
  113. }
  114. /**
  115. * Set the desired job deduplicator callback.
  116. *
  117. * This feature is only supported by some queues, such as Amazon SQS FIFO.
  118. *
  119. * @param callable|null $deduplicator
  120. * @return $this
  121. */
  122. public function withDeduplicator($deduplicator)
  123. {
  124. $this->deduplicator = $deduplicator instanceof Closure
  125. ? new SerializableClosure($deduplicator)
  126. : $deduplicator;
  127. return $this;
  128. }
  129. /**
  130. * Set the desired connection for the chain.
  131. *
  132. * @param \UnitEnum|string|null $connection
  133. * @return $this
  134. */
  135. public function allOnConnection($connection)
  136. {
  137. $resolvedConnection = enum_value($connection);
  138. $this->chainConnection = $resolvedConnection;
  139. $this->connection = $resolvedConnection;
  140. return $this;
  141. }
  142. /**
  143. * Set the desired queue for the chain.
  144. *
  145. * @param \UnitEnum|string|null $queue
  146. * @return $this
  147. */
  148. public function allOnQueue($queue)
  149. {
  150. $resolvedQueue = enum_value($queue);
  151. $this->chainQueue = $resolvedQueue;
  152. $this->queue = $resolvedQueue;
  153. return $this;
  154. }
  155. /**
  156. * Set the desired delay in seconds for the job.
  157. *
  158. * @param \DateTimeInterface|\DateInterval|array|int|null $delay
  159. * @return $this
  160. */
  161. public function delay($delay)
  162. {
  163. $this->delay = $delay;
  164. return $this;
  165. }
  166. /**
  167. * Set the delay for the job to zero seconds.
  168. *
  169. * @return $this
  170. */
  171. public function withoutDelay()
  172. {
  173. $this->delay = 0;
  174. return $this;
  175. }
  176. /**
  177. * Indicate that the job should be dispatched after all database transactions have committed.
  178. *
  179. * @return $this
  180. */
  181. public function afterCommit()
  182. {
  183. $this->afterCommit = true;
  184. return $this;
  185. }
  186. /**
  187. * Indicate that the job should not wait until database transactions have been committed before dispatching.
  188. *
  189. * @return $this
  190. */
  191. public function beforeCommit()
  192. {
  193. $this->afterCommit = false;
  194. return $this;
  195. }
  196. /**
  197. * Specify the middleware the job should be dispatched through.
  198. *
  199. * @param array|object $middleware
  200. * @return $this
  201. */
  202. public function through($middleware)
  203. {
  204. $this->middleware = Arr::wrap($middleware);
  205. return $this;
  206. }
  207. /**
  208. * Set the jobs that should run if this job is successful.
  209. *
  210. * @param array $chain
  211. * @return $this
  212. */
  213. public function chain($chain)
  214. {
  215. $this->chained = ChainedBatch::prepareNestedBatches(new Collection($chain))
  216. ->map(fn ($job) => $this->serializeJob($job))
  217. ->all();
  218. return $this;
  219. }
  220. /**
  221. * Prepend a job to the current chain so that it is run after the currently running job.
  222. *
  223. * @param mixed $job
  224. * @return $this
  225. */
  226. public function prependToChain($job)
  227. {
  228. $jobs = ChainedBatch::prepareNestedBatches(Collection::wrap($job));
  229. foreach ($jobs->reverse() as $job) {
  230. $this->chained = Arr::prepend($this->chained, $this->serializeJob($job));
  231. }
  232. return $this;
  233. }
  234. /**
  235. * Append a job to the end of the current chain.
  236. *
  237. * @param mixed $job
  238. * @return $this
  239. */
  240. public function appendToChain($job)
  241. {
  242. $jobs = ChainedBatch::prepareNestedBatches(Collection::wrap($job));
  243. foreach ($jobs as $job) {
  244. $this->chained = array_merge($this->chained, [$this->serializeJob($job)]);
  245. }
  246. return $this;
  247. }
  248. /**
  249. * Serialize a job for queuing.
  250. *
  251. * @param mixed $job
  252. * @return string
  253. *
  254. * @throws \RuntimeException
  255. */
  256. protected function serializeJob($job)
  257. {
  258. if ($job instanceof Closure) {
  259. if (! class_exists(CallQueuedClosure::class)) {
  260. throw new RuntimeException(
  261. 'To enable support for closure jobs, please install the illuminate/queue package.'
  262. );
  263. }
  264. $job = CallQueuedClosure::create($job);
  265. }
  266. return serialize($job);
  267. }
  268. /**
  269. * Dispatch the next job on the chain.
  270. *
  271. * @return void
  272. */
  273. public function dispatchNextJobInChain()
  274. {
  275. if (! empty($this->chained)) {
  276. dispatch(tap(unserialize(array_shift($this->chained)), function ($next) {
  277. $next->chained = $this->chained;
  278. $next->onConnection($next->connection ?: $this->chainConnection);
  279. $next->onQueue($next->queue ?: $this->chainQueue);
  280. $next->chainConnection = $this->chainConnection;
  281. $next->chainQueue = $this->chainQueue;
  282. $next->chainCatchCallbacks = $this->chainCatchCallbacks;
  283. }));
  284. }
  285. }
  286. /**
  287. * Invoke all of the chain's failed job callbacks.
  288. *
  289. * @param \Throwable $e
  290. * @return void
  291. */
  292. public function invokeChainCatchCallbacks($e)
  293. {
  294. (new Collection($this->chainCatchCallbacks))->each(function ($callback) use ($e) {
  295. $callback($e);
  296. });
  297. }
  298. /**
  299. * Assert that the job has the given chain of jobs attached to it.
  300. *
  301. * @param array $expectedChain
  302. * @return void
  303. */
  304. public function assertHasChain($expectedChain)
  305. {
  306. PHPUnit::assertTrue(
  307. (new Collection($expectedChain))->isNotEmpty(),
  308. 'The expected chain can not be empty.'
  309. );
  310. if ((new Collection($expectedChain))->contains(fn ($job) => is_object($job))) {
  311. $expectedChain = (new Collection($expectedChain))->map(fn ($job) => serialize($job))->all();
  312. } else {
  313. $chain = (new Collection($this->chained))->map(fn ($job) => get_class(unserialize($job)))->all();
  314. }
  315. PHPUnit::assertTrue(
  316. $expectedChain === ($chain ?? $this->chained),
  317. 'The job does not have the expected chain.'
  318. );
  319. }
  320. /**
  321. * Assert that the job has no remaining chained jobs.
  322. *
  323. * @return void
  324. */
  325. public function assertDoesntHaveChain()
  326. {
  327. PHPUnit::assertEmpty($this->chained, 'The job has chained jobs.');
  328. }
  329. }