Dispatcher.php 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316
  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. * Indicates if dispatching after response is disabled.
  48. *
  49. * @var bool
  50. */
  51. protected $allowsDispatchingAfterResponses = true;
  52. /**
  53. * Create a new command dispatcher instance.
  54. */
  55. public function __construct(Container $container, ?Closure $queueResolver = null)
  56. {
  57. $this->container = $container;
  58. $this->queueResolver = $queueResolver;
  59. $this->pipeline = new Pipeline($container);
  60. }
  61. /**
  62. * Dispatch a command to its appropriate handler.
  63. *
  64. * @param mixed $command
  65. * @return mixed
  66. */
  67. public function dispatch($command)
  68. {
  69. return $this->queueResolver && $this->commandShouldBeQueued($command)
  70. ? $this->dispatchToQueue($command)
  71. : $this->dispatchNow($command);
  72. }
  73. /**
  74. * Dispatch a command to its appropriate handler in the current process.
  75. *
  76. * Queueable jobs will be dispatched to the "sync" queue.
  77. *
  78. * @param mixed $command
  79. * @param mixed $handler
  80. * @return mixed
  81. */
  82. public function dispatchSync($command, $handler = null)
  83. {
  84. if ($this->queueResolver &&
  85. $this->commandShouldBeQueued($command) &&
  86. method_exists($command, 'onConnection')) {
  87. return $this->dispatchToQueue($command->onConnection('sync'));
  88. }
  89. return $this->dispatchNow($command, $handler);
  90. }
  91. /**
  92. * Dispatch a command to its appropriate handler in the current process without using the synchronous queue.
  93. *
  94. * @param mixed $command
  95. * @param mixed $handler
  96. * @return mixed
  97. */
  98. public function dispatchNow($command, $handler = null)
  99. {
  100. $uses = class_uses_recursive($command);
  101. if (isset($uses[InteractsWithQueue::class], $uses[Queueable::class]) && ! $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. * @return \Illuminate\Bus\Batch|null
  121. */
  122. public function findBatch(string $batchId)
  123. {
  124. return $this->container->make(BatchRepository::class)->find($batchId);
  125. }
  126. /**
  127. * Create a new batch of queueable jobs.
  128. *
  129. * @param \Illuminate\Support\Collection|mixed $jobs
  130. * @return \Illuminate\Bus\PendingBatch
  131. */
  132. public function batch($jobs)
  133. {
  134. return new PendingBatch($this->container, Collection::wrap($jobs));
  135. }
  136. /**
  137. * Create a new chain of queueable jobs.
  138. *
  139. * @param \Illuminate\Support\Collection|array|null $jobs
  140. * @return \Illuminate\Foundation\Bus\PendingChain
  141. */
  142. public function chain($jobs = null)
  143. {
  144. $jobs = Collection::wrap($jobs);
  145. $jobs = ChainedBatch::prepareNestedBatches($jobs);
  146. return new PendingChain($jobs->shift(), $jobs->toArray());
  147. }
  148. /**
  149. * Determine if the given command has a handler.
  150. *
  151. * @param mixed $command
  152. * @return bool
  153. */
  154. public function hasCommandHandler($command)
  155. {
  156. return array_key_exists(get_class($command), $this->handlers);
  157. }
  158. /**
  159. * Retrieve the handler for a command.
  160. *
  161. * @param mixed $command
  162. * @return mixed
  163. */
  164. public function getCommandHandler($command)
  165. {
  166. if ($this->hasCommandHandler($command)) {
  167. return $this->container->make($this->handlers[get_class($command)]);
  168. }
  169. return false;
  170. }
  171. /**
  172. * Determine if the given command should be queued.
  173. *
  174. * @param mixed $command
  175. * @return bool
  176. */
  177. protected function commandShouldBeQueued($command)
  178. {
  179. return $command instanceof ShouldQueue;
  180. }
  181. /**
  182. * Dispatch a command to its appropriate handler behind a queue.
  183. *
  184. * @param mixed $command
  185. * @return mixed
  186. *
  187. * @throws \RuntimeException
  188. */
  189. public function dispatchToQueue($command)
  190. {
  191. $connection = $command->connection ?? null;
  192. $queue = ($this->queueResolver)($connection);
  193. if (! $queue instanceof Queue) {
  194. throw new RuntimeException('Queue resolver did not return a Queue implementation.');
  195. }
  196. if (method_exists($command, 'queue')) {
  197. return $command->queue($queue, $command);
  198. }
  199. return $this->pushCommandToQueue($queue, $command);
  200. }
  201. /**
  202. * Push the command onto the given queue instance.
  203. *
  204. * @param \Illuminate\Contracts\Queue\Queue $queue
  205. * @param mixed $command
  206. * @return mixed
  207. */
  208. protected function pushCommandToQueue($queue, $command)
  209. {
  210. if (isset($command->delay)) {
  211. return $queue->later($command->delay, $command, queue: $command->queue ?? null);
  212. }
  213. return $queue->push($command, queue: $command->queue ?? null);
  214. }
  215. /**
  216. * Dispatch a command to its appropriate handler after the current process.
  217. *
  218. * @param mixed $command
  219. * @param mixed $handler
  220. * @return void
  221. */
  222. public function dispatchAfterResponse($command, $handler = null)
  223. {
  224. if (! $this->allowsDispatchingAfterResponses) {
  225. $this->dispatchSync($command);
  226. return;
  227. }
  228. $this->container->terminating(function () use ($command, $handler) {
  229. $this->dispatchSync($command, $handler);
  230. });
  231. }
  232. /**
  233. * Set the pipes through which commands should be piped before dispatching.
  234. *
  235. * @return $this
  236. */
  237. public function pipeThrough(array $pipes)
  238. {
  239. $this->pipes = $pipes;
  240. return $this;
  241. }
  242. /**
  243. * Map a command to a handler.
  244. *
  245. * @return $this
  246. */
  247. public function map(array $map)
  248. {
  249. $this->handlers = array_merge($this->handlers, $map);
  250. return $this;
  251. }
  252. /**
  253. * Allow dispatching after responses.
  254. *
  255. * @return $this
  256. */
  257. public function withDispatchingAfterResponses()
  258. {
  259. $this->allowsDispatchingAfterResponses = true;
  260. return $this;
  261. }
  262. /**
  263. * Disable dispatching after responses.
  264. *
  265. * @return $this
  266. */
  267. public function withoutDispatchingAfterResponses()
  268. {
  269. $this->allowsDispatchingAfterResponses = false;
  270. return $this;
  271. }
  272. }