QueuedClosure.php 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. <?php
  2. namespace Illuminate\Events;
  3. use Closure;
  4. use Illuminate\Support\Collection;
  5. use Laravel\SerializableClosure\SerializableClosure;
  6. use function Illuminate\Support\enum_value;
  7. class QueuedClosure
  8. {
  9. /**
  10. * The underlying Closure.
  11. *
  12. * @var \Closure
  13. */
  14. public $closure;
  15. /**
  16. * The name of the connection the job should be sent to.
  17. *
  18. * @var string|null
  19. */
  20. public $connection;
  21. /**
  22. * The name of the queue the job should be sent to.
  23. *
  24. * @var string|null
  25. */
  26. public $queue;
  27. /**
  28. * The job "group" the job should be sent to.
  29. *
  30. * @var string|null
  31. */
  32. public $messageGroup;
  33. /**
  34. * The job deduplicator callback the job should use to generate the deduplication ID.
  35. *
  36. * @var \Laravel\SerializableClosure\SerializableClosure|null
  37. */
  38. public $deduplicator;
  39. /**
  40. * The number of seconds before the job should be made available.
  41. *
  42. * @var \DateTimeInterface|\DateInterval|int|null
  43. */
  44. public $delay;
  45. /**
  46. * All of the "catch" callbacks for the queued closure.
  47. *
  48. * @var array
  49. */
  50. public $catchCallbacks = [];
  51. /**
  52. * Create a new queued closure event listener resolver.
  53. *
  54. * @param \Closure $closure
  55. */
  56. public function __construct(Closure $closure)
  57. {
  58. $this->closure = $closure;
  59. }
  60. /**
  61. * Set the desired connection for the job.
  62. *
  63. * @param \UnitEnum|string|null $connection
  64. * @return $this
  65. */
  66. public function onConnection($connection)
  67. {
  68. $this->connection = enum_value($connection);
  69. return $this;
  70. }
  71. /**
  72. * Set the desired queue for the job.
  73. *
  74. * @param \UnitEnum|string|null $queue
  75. * @return $this
  76. */
  77. public function onQueue($queue)
  78. {
  79. $this->queue = enum_value($queue);
  80. return $this;
  81. }
  82. /**
  83. * Set the desired job "group".
  84. *
  85. * This feature is only supported by some queues, such as Amazon SQS.
  86. *
  87. * @param \UnitEnum|string $group
  88. * @return $this
  89. */
  90. public function onGroup($group)
  91. {
  92. $this->messageGroup = enum_value($group);
  93. return $this;
  94. }
  95. /**
  96. * Set the desired job deduplicator callback.
  97. *
  98. * This feature is only supported by some queues, such as Amazon SQS FIFO.
  99. *
  100. * @param callable|null $deduplicator
  101. * @return $this
  102. */
  103. public function withDeduplicator($deduplicator)
  104. {
  105. $this->deduplicator = $deduplicator instanceof Closure
  106. ? new SerializableClosure($deduplicator)
  107. : $deduplicator;
  108. return $this;
  109. }
  110. /**
  111. * Set the desired delay in seconds for the job.
  112. *
  113. * @param \DateTimeInterface|\DateInterval|int|null $delay
  114. * @return $this
  115. */
  116. public function delay($delay)
  117. {
  118. $this->delay = $delay;
  119. return $this;
  120. }
  121. /**
  122. * Specify a callback that should be invoked if the queued listener job fails.
  123. *
  124. * @param \Closure $closure
  125. * @return $this
  126. */
  127. public function catch(Closure $closure)
  128. {
  129. $this->catchCallbacks[] = $closure;
  130. return $this;
  131. }
  132. /**
  133. * Resolve the actual event listener callback.
  134. *
  135. * @return \Closure
  136. */
  137. public function resolve()
  138. {
  139. return function (...$arguments) {
  140. dispatch(new CallQueuedListener(InvokeQueuedClosure::class, 'handle', [
  141. 'closure' => new SerializableClosure($this->closure),
  142. 'arguments' => $arguments,
  143. 'catch' => (new Collection($this->catchCallbacks))
  144. ->map(fn ($callback) => new SerializableClosure($callback))
  145. ->all(),
  146. ]))
  147. ->onConnection($this->connection)
  148. ->onQueue($this->queue)
  149. ->delay($this->delay)
  150. ->onGroup($this->messageGroup)
  151. ->withDeduplicator($this->deduplicator);
  152. };
  153. }
  154. }