Dispatcher.php 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. <?php
  2. namespace Illuminate\Bus;
  3. use Closure;
  4. use Illuminate\Contracts\Bus\QueueingDispatcher;
  5. use Illuminate\Contracts\Container\Container;
  6. use Illuminate\Contracts\Queue\Queue;
  7. use Illuminate\Contracts\Queue\ShouldQueue;
  8. use Illuminate\Foundation\Bus\PendingChain;
  9. use Illuminate\Pipeline\Pipeline;
  10. use Illuminate\Queue\InteractsWithQueue;
  11. use Illuminate\Queue\Jobs\SyncJob;
  12. use Illuminate\Support\Collection;
  13. use RuntimeException;
  14. class Dispatcher implements QueueingDispatcher
  15. {
  16. /**
  17. * The container implementation.
  18. *
  19. * @var \Illuminate\Contracts\Container\Container
  20. */
  21. protected $container;
  22. /**
  23. * The pipeline instance for the bus.
  24. *
  25. * @var \Illuminate\Pipeline\Pipeline
  26. */
  27. protected $pipeline;
  28. /**
  29. * The pipes to send commands through before dispatching.
  30. *
  31. * @var array
  32. */
  33. protected $pipes = [];
  34. /**
  35. * The command to handler mapping for non-self-handling events.
  36. *
  37. * @var array
  38. */
  39. protected $handlers = [];
  40. /**
  41. * The queue resolver callback.
  42. *
  43. * @var \Closure|null
  44. */
  45. protected $queueResolver;
  46. /**
  47. * Create a new command dispatcher instance.
  48. *
  49. * @param \Illuminate\Contracts\Container\Container $container
  50. * @param \Closure|null $queueResolver
  51. * @return void
  52. */
  53. public function __construct(Container $container, ?Closure $queueResolver = null)
  54. {
  55. $this->container = $container;
  56. $this->queueResolver = $queueResolver;
  57. $this->pipeline = new Pipeline($container);
  58. }
  59. /**
  60. * Dispatch a command to its appropriate handler.
  61. *
  62. * @param mixed $command
  63. * @return mixed
  64. */
  65. public function dispatch($command)
  66. {
  67. return $this->queueResolver && $this->commandShouldBeQueued($command)
  68. ? $this->dispatchToQueue($command)
  69. : $this->dispatchNow($command);
  70. }
  71. /**
  72. * Dispatch a command to its appropriate handler in the current process.
  73. *
  74. * Queueable jobs will be dispatched to the "sync" queue.
  75. *
  76. * @param mixed $command
  77. * @param mixed $handler
  78. * @return mixed
  79. */
  80. public function dispatchSync($command, $handler = null)
  81. {
  82. if ($this->queueResolver &&
  83. $this->commandShouldBeQueued($command) &&
  84. method_exists($command, 'onConnection')) {
  85. return $this->dispatchToQueue($command->onConnection('sync'));
  86. }
  87. return $this->dispatchNow($command, $handler);
  88. }
  89. /**
  90. * Dispatch a command to its appropriate handler in the current process without using the synchronous queue.
  91. *
  92. * @param mixed $command
  93. * @param mixed $handler
  94. * @return mixed
  95. */
  96. public function dispatchNow($command, $handler = null)
  97. {
  98. $uses = class_uses_recursive($command);
  99. if (in_array(InteractsWithQueue::class, $uses) &&
  100. in_array(Queueable::class, $uses) &&
  101. ! $command->job) {
  102. $command->setJob(new SyncJob($this->container, json_encode([]), 'sync', 'sync'));
  103. }
  104. if ($handler || $handler = $this->getCommandHandler($command)) {
  105. $callback = function ($command) use ($handler) {
  106. $method = method_exists($handler, 'handle') ? 'handle' : '__invoke';
  107. return $handler->{$method}($command);
  108. };
  109. } else {
  110. $callback = function ($command) {
  111. $method = method_exists($command, 'handle') ? 'handle' : '__invoke';
  112. return $this->container->call([$command, $method]);
  113. };
  114. }
  115. return $this->pipeline->send($command)->through($this->pipes)->then($callback);
  116. }
  117. /**
  118. * Attempt to find the batch with the given ID.
  119. *
  120. * @param string $batchId
  121. * @return \Illuminate\Bus\Batch|null
  122. */
  123. public function findBatch(string $batchId)
  124. {
  125. return $this->container->make(BatchRepository::class)->find($batchId);
  126. }
  127. /**
  128. * Create a new batch of queueable jobs.
  129. *
  130. * @param \Illuminate\Support\Collection|array|mixed $jobs
  131. * @return \Illuminate\Bus\PendingBatch
  132. */
  133. public function batch($jobs)
  134. {
  135. return new PendingBatch($this->container, Collection::wrap($jobs));
  136. }
  137. /**
  138. * Create a new chain of queueable jobs.
  139. *
  140. * @param \Illuminate\Support\Collection|array $jobs
  141. * @return \Illuminate\Foundation\Bus\PendingChain
  142. */
  143. public function chain($jobs)
  144. {
  145. $jobs = Collection::wrap($jobs);
  146. $jobs = ChainedBatch::prepareNestedBatches($jobs);
  147. return new PendingChain($jobs->shift(), $jobs->toArray());
  148. }
  149. /**
  150. * Determine if the given command has a handler.
  151. *
  152. * @param mixed $command
  153. * @return bool
  154. */
  155. public function hasCommandHandler($command)
  156. {
  157. return array_key_exists(get_class($command), $this->handlers);
  158. }
  159. /**
  160. * Retrieve the handler for a command.
  161. *
  162. * @param mixed $command
  163. * @return bool|mixed
  164. */
  165. public function getCommandHandler($command)
  166. {
  167. if ($this->hasCommandHandler($command)) {
  168. return $this->container->make($this->handlers[get_class($command)]);
  169. }
  170. return false;
  171. }
  172. /**
  173. * Determine if the given command should be queued.
  174. *
  175. * @param mixed $command
  176. * @return bool
  177. */
  178. protected function commandShouldBeQueued($command)
  179. {
  180. return $command instanceof ShouldQueue;
  181. }
  182. /**
  183. * Dispatch a command to its appropriate handler behind a queue.
  184. *
  185. * @param mixed $command
  186. * @return mixed
  187. *
  188. * @throws \RuntimeException
  189. */
  190. public function dispatchToQueue($command)
  191. {
  192. $connection = $command->connection ?? null;
  193. $queue = call_user_func($this->queueResolver, $connection);
  194. if (! $queue instanceof Queue) {
  195. throw new RuntimeException('Queue resolver did not return a Queue implementation.');
  196. }
  197. if (method_exists($command, 'queue')) {
  198. return $command->queue($queue, $command);
  199. }
  200. return $this->pushCommandToQueue($queue, $command);
  201. }
  202. /**
  203. * Push the command onto the given queue instance.
  204. *
  205. * @param \Illuminate\Contracts\Queue\Queue $queue
  206. * @param mixed $command
  207. * @return mixed
  208. */
  209. protected function pushCommandToQueue($queue, $command)
  210. {
  211. if (isset($command->queue, $command->delay)) {
  212. return $queue->laterOn($command->queue, $command->delay, $command);
  213. }
  214. if (isset($command->queue)) {
  215. return $queue->pushOn($command->queue, $command);
  216. }
  217. if (isset($command->delay)) {
  218. return $queue->later($command->delay, $command);
  219. }
  220. return $queue->push($command);
  221. }
  222. /**
  223. * Dispatch a command to its appropriate handler after the current process.
  224. *
  225. * @param mixed $command
  226. * @param mixed $handler
  227. * @return void
  228. */
  229. public function dispatchAfterResponse($command, $handler = null)
  230. {
  231. $this->container->terminating(function () use ($command, $handler) {
  232. $this->dispatchSync($command, $handler);
  233. });
  234. }
  235. /**
  236. * Set the pipes through which commands should be piped before dispatching.
  237. *
  238. * @param array $pipes
  239. * @return $this
  240. */
  241. public function pipeThrough(array $pipes)
  242. {
  243. $this->pipes = $pipes;
  244. return $this;
  245. }
  246. /**
  247. * Map a command to a handler.
  248. *
  249. * @param array $map
  250. * @return $this
  251. */
  252. public function map(array $map)
  253. {
  254. $this->handlers = array_merge($this->handlers, $map);
  255. return $this;
  256. }
  257. }