Dispatcher.php 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777
  1. <?php
  2. namespace Illuminate\Events;
  3. use Closure;
  4. use Exception;
  5. use Illuminate\Container\Container;
  6. use Illuminate\Contracts\Broadcasting\Factory as BroadcastFactory;
  7. use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
  8. use Illuminate\Contracts\Container\Container as ContainerContract;
  9. use Illuminate\Contracts\Events\Dispatcher as DispatcherContract;
  10. use Illuminate\Contracts\Events\ShouldDispatchAfterCommit;
  11. use Illuminate\Contracts\Events\ShouldHandleEventsAfterCommit;
  12. use Illuminate\Contracts\Queue\ShouldBeEncrypted;
  13. use Illuminate\Contracts\Queue\ShouldQueue;
  14. use Illuminate\Contracts\Queue\ShouldQueueAfterCommit;
  15. use Illuminate\Support\Arr;
  16. use Illuminate\Support\Str;
  17. use Illuminate\Support\Traits\Macroable;
  18. use Illuminate\Support\Traits\ReflectsClosures;
  19. use ReflectionClass;
  20. class Dispatcher implements DispatcherContract
  21. {
  22. use Macroable, ReflectsClosures;
  23. /**
  24. * The IoC container instance.
  25. *
  26. * @var \Illuminate\Contracts\Container\Container
  27. */
  28. protected $container;
  29. /**
  30. * The registered event listeners.
  31. *
  32. * @var array
  33. */
  34. protected $listeners = [];
  35. /**
  36. * The wildcard listeners.
  37. *
  38. * @var array
  39. */
  40. protected $wildcards = [];
  41. /**
  42. * The cached wildcard listeners.
  43. *
  44. * @var array
  45. */
  46. protected $wildcardsCache = [];
  47. /**
  48. * The queue resolver instance.
  49. *
  50. * @var callable
  51. */
  52. protected $queueResolver;
  53. /**
  54. * The database transaction manager resolver instance.
  55. *
  56. * @var callable
  57. */
  58. protected $transactionManagerResolver;
  59. /**
  60. * Create a new event dispatcher instance.
  61. *
  62. * @param \Illuminate\Contracts\Container\Container|null $container
  63. * @return void
  64. */
  65. public function __construct(?ContainerContract $container = null)
  66. {
  67. $this->container = $container ?: new Container;
  68. }
  69. /**
  70. * Register an event listener with the dispatcher.
  71. *
  72. * @param \Closure|string|array $events
  73. * @param \Closure|string|array|null $listener
  74. * @return void
  75. */
  76. public function listen($events, $listener = null)
  77. {
  78. if ($events instanceof Closure) {
  79. return collect($this->firstClosureParameterTypes($events))
  80. ->each(function ($event) use ($events) {
  81. $this->listen($event, $events);
  82. });
  83. } elseif ($events instanceof QueuedClosure) {
  84. return collect($this->firstClosureParameterTypes($events->closure))
  85. ->each(function ($event) use ($events) {
  86. $this->listen($event, $events->resolve());
  87. });
  88. } elseif ($listener instanceof QueuedClosure) {
  89. $listener = $listener->resolve();
  90. }
  91. foreach ((array) $events as $event) {
  92. if (str_contains($event, '*')) {
  93. $this->setupWildcardListen($event, $listener);
  94. } else {
  95. $this->listeners[$event][] = $listener;
  96. }
  97. }
  98. }
  99. /**
  100. * Setup a wildcard listener callback.
  101. *
  102. * @param string $event
  103. * @param \Closure|string $listener
  104. * @return void
  105. */
  106. protected function setupWildcardListen($event, $listener)
  107. {
  108. $this->wildcards[$event][] = $listener;
  109. $this->wildcardsCache = [];
  110. }
  111. /**
  112. * Determine if a given event has listeners.
  113. *
  114. * @param string $eventName
  115. * @return bool
  116. */
  117. public function hasListeners($eventName)
  118. {
  119. return isset($this->listeners[$eventName]) ||
  120. isset($this->wildcards[$eventName]) ||
  121. $this->hasWildcardListeners($eventName);
  122. }
  123. /**
  124. * Determine if the given event has any wildcard listeners.
  125. *
  126. * @param string $eventName
  127. * @return bool
  128. */
  129. public function hasWildcardListeners($eventName)
  130. {
  131. foreach ($this->wildcards as $key => $listeners) {
  132. if (Str::is($key, $eventName)) {
  133. return true;
  134. }
  135. }
  136. return false;
  137. }
  138. /**
  139. * Register an event and payload to be fired later.
  140. *
  141. * @param string $event
  142. * @param object|array $payload
  143. * @return void
  144. */
  145. public function push($event, $payload = [])
  146. {
  147. $this->listen($event.'_pushed', function () use ($event, $payload) {
  148. $this->dispatch($event, $payload);
  149. });
  150. }
  151. /**
  152. * Flush a set of pushed events.
  153. *
  154. * @param string $event
  155. * @return void
  156. */
  157. public function flush($event)
  158. {
  159. $this->dispatch($event.'_pushed');
  160. }
  161. /**
  162. * Register an event subscriber with the dispatcher.
  163. *
  164. * @param object|string $subscriber
  165. * @return void
  166. */
  167. public function subscribe($subscriber)
  168. {
  169. $subscriber = $this->resolveSubscriber($subscriber);
  170. $events = $subscriber->subscribe($this);
  171. if (is_array($events)) {
  172. foreach ($events as $event => $listeners) {
  173. foreach (Arr::wrap($listeners) as $listener) {
  174. if (is_string($listener) && method_exists($subscriber, $listener)) {
  175. $this->listen($event, [get_class($subscriber), $listener]);
  176. continue;
  177. }
  178. $this->listen($event, $listener);
  179. }
  180. }
  181. }
  182. }
  183. /**
  184. * Resolve the subscriber instance.
  185. *
  186. * @param object|string $subscriber
  187. * @return mixed
  188. */
  189. protected function resolveSubscriber($subscriber)
  190. {
  191. if (is_string($subscriber)) {
  192. return $this->container->make($subscriber);
  193. }
  194. return $subscriber;
  195. }
  196. /**
  197. * Fire an event until the first non-null response is returned.
  198. *
  199. * @param string|object $event
  200. * @param mixed $payload
  201. * @return mixed
  202. */
  203. public function until($event, $payload = [])
  204. {
  205. return $this->dispatch($event, $payload, true);
  206. }
  207. /**
  208. * Fire an event and call the listeners.
  209. *
  210. * @param string|object $event
  211. * @param mixed $payload
  212. * @param bool $halt
  213. * @return array|null
  214. */
  215. public function dispatch($event, $payload = [], $halt = false)
  216. {
  217. // When the given "event" is actually an object we will assume it is an event
  218. // object and use the class as the event name and this event itself as the
  219. // payload to the handler, which makes object based events quite simple.
  220. [$isEventObject, $event, $payload] = [
  221. is_object($event),
  222. ...$this->parseEventAndPayload($event, $payload),
  223. ];
  224. // If the event is not intended to be dispatched unless the current database
  225. // transaction is successful, we'll register a callback which will handle
  226. // dispatching this event on the next successful DB transaction commit.
  227. if ($isEventObject &&
  228. $payload[0] instanceof ShouldDispatchAfterCommit &&
  229. ! is_null($transactions = $this->resolveTransactionManager())) {
  230. $transactions->addCallback(
  231. fn () => $this->invokeListeners($event, $payload, $halt)
  232. );
  233. return null;
  234. }
  235. return $this->invokeListeners($event, $payload, $halt);
  236. }
  237. /**
  238. * Broadcast an event and call its listeners.
  239. *
  240. * @param string|object $event
  241. * @param mixed $payload
  242. * @param bool $halt
  243. * @return array|null
  244. */
  245. protected function invokeListeners($event, $payload, $halt = false)
  246. {
  247. if ($this->shouldBroadcast($payload)) {
  248. $this->broadcastEvent($payload[0]);
  249. }
  250. $responses = [];
  251. foreach ($this->getListeners($event) as $listener) {
  252. $response = $listener($event, $payload);
  253. // If a response is returned from the listener and event halting is enabled
  254. // we will just return this response, and not call the rest of the event
  255. // listeners. Otherwise we will add the response on the response list.
  256. if ($halt && ! is_null($response)) {
  257. return $response;
  258. }
  259. // If a boolean false is returned from a listener, we will stop propagating
  260. // the event to any further listeners down in the chain, else we keep on
  261. // looping through the listeners and firing every one in our sequence.
  262. if ($response === false) {
  263. break;
  264. }
  265. $responses[] = $response;
  266. }
  267. return $halt ? null : $responses;
  268. }
  269. /**
  270. * Parse the given event and payload and prepare them for dispatching.
  271. *
  272. * @param mixed $event
  273. * @param mixed $payload
  274. * @return array
  275. */
  276. protected function parseEventAndPayload($event, $payload)
  277. {
  278. if (is_object($event)) {
  279. [$payload, $event] = [[$event], get_class($event)];
  280. }
  281. return [$event, Arr::wrap($payload)];
  282. }
  283. /**
  284. * Determine if the payload has a broadcastable event.
  285. *
  286. * @param array $payload
  287. * @return bool
  288. */
  289. protected function shouldBroadcast(array $payload)
  290. {
  291. return isset($payload[0]) &&
  292. $payload[0] instanceof ShouldBroadcast &&
  293. $this->broadcastWhen($payload[0]);
  294. }
  295. /**
  296. * Check if the event should be broadcasted by the condition.
  297. *
  298. * @param mixed $event
  299. * @return bool
  300. */
  301. protected function broadcastWhen($event)
  302. {
  303. return method_exists($event, 'broadcastWhen')
  304. ? $event->broadcastWhen() : true;
  305. }
  306. /**
  307. * Broadcast the given event class.
  308. *
  309. * @param \Illuminate\Contracts\Broadcasting\ShouldBroadcast $event
  310. * @return void
  311. */
  312. protected function broadcastEvent($event)
  313. {
  314. $this->container->make(BroadcastFactory::class)->queue($event);
  315. }
  316. /**
  317. * Get all of the listeners for a given event name.
  318. *
  319. * @param string $eventName
  320. * @return array
  321. */
  322. public function getListeners($eventName)
  323. {
  324. $listeners = array_merge(
  325. $this->prepareListeners($eventName),
  326. $this->wildcardsCache[$eventName] ?? $this->getWildcardListeners($eventName)
  327. );
  328. return class_exists($eventName, false)
  329. ? $this->addInterfaceListeners($eventName, $listeners)
  330. : $listeners;
  331. }
  332. /**
  333. * Get the wildcard listeners for the event.
  334. *
  335. * @param string $eventName
  336. * @return array
  337. */
  338. protected function getWildcardListeners($eventName)
  339. {
  340. $wildcards = [];
  341. foreach ($this->wildcards as $key => $listeners) {
  342. if (Str::is($key, $eventName)) {
  343. foreach ($listeners as $listener) {
  344. $wildcards[] = $this->makeListener($listener, true);
  345. }
  346. }
  347. }
  348. return $this->wildcardsCache[$eventName] = $wildcards;
  349. }
  350. /**
  351. * Add the listeners for the event's interfaces to the given array.
  352. *
  353. * @param string $eventName
  354. * @param array $listeners
  355. * @return array
  356. */
  357. protected function addInterfaceListeners($eventName, array $listeners = [])
  358. {
  359. foreach (class_implements($eventName) as $interface) {
  360. if (isset($this->listeners[$interface])) {
  361. foreach ($this->prepareListeners($interface) as $names) {
  362. $listeners = array_merge($listeners, (array) $names);
  363. }
  364. }
  365. }
  366. return $listeners;
  367. }
  368. /**
  369. * Prepare the listeners for a given event.
  370. *
  371. * @param string $eventName
  372. * @return \Closure[]
  373. */
  374. protected function prepareListeners(string $eventName)
  375. {
  376. $listeners = [];
  377. foreach ($this->listeners[$eventName] ?? [] as $listener) {
  378. $listeners[] = $this->makeListener($listener);
  379. }
  380. return $listeners;
  381. }
  382. /**
  383. * Register an event listener with the dispatcher.
  384. *
  385. * @param \Closure|string|array $listener
  386. * @param bool $wildcard
  387. * @return \Closure
  388. */
  389. public function makeListener($listener, $wildcard = false)
  390. {
  391. if (is_string($listener)) {
  392. return $this->createClassListener($listener, $wildcard);
  393. }
  394. if (is_array($listener) && isset($listener[0]) && is_string($listener[0])) {
  395. return $this->createClassListener($listener, $wildcard);
  396. }
  397. return function ($event, $payload) use ($listener, $wildcard) {
  398. if ($wildcard) {
  399. return $listener($event, $payload);
  400. }
  401. return $listener(...array_values($payload));
  402. };
  403. }
  404. /**
  405. * Create a class based listener using the IoC container.
  406. *
  407. * @param string $listener
  408. * @param bool $wildcard
  409. * @return \Closure
  410. */
  411. public function createClassListener($listener, $wildcard = false)
  412. {
  413. return function ($event, $payload) use ($listener, $wildcard) {
  414. if ($wildcard) {
  415. return call_user_func($this->createClassCallable($listener), $event, $payload);
  416. }
  417. $callable = $this->createClassCallable($listener);
  418. return $callable(...array_values($payload));
  419. };
  420. }
  421. /**
  422. * Create the class based event callable.
  423. *
  424. * @param array|string $listener
  425. * @return callable
  426. */
  427. protected function createClassCallable($listener)
  428. {
  429. [$class, $method] = is_array($listener)
  430. ? $listener
  431. : $this->parseClassCallable($listener);
  432. if (! method_exists($class, $method)) {
  433. $method = '__invoke';
  434. }
  435. if ($this->handlerShouldBeQueued($class)) {
  436. return $this->createQueuedHandlerCallable($class, $method);
  437. }
  438. $listener = $this->container->make($class);
  439. return $this->handlerShouldBeDispatchedAfterDatabaseTransactions($listener)
  440. ? $this->createCallbackForListenerRunningAfterCommits($listener, $method)
  441. : [$listener, $method];
  442. }
  443. /**
  444. * Parse the class listener into class and method.
  445. *
  446. * @param string $listener
  447. * @return array
  448. */
  449. protected function parseClassCallable($listener)
  450. {
  451. return Str::parseCallback($listener, 'handle');
  452. }
  453. /**
  454. * Determine if the event handler class should be queued.
  455. *
  456. * @param string $class
  457. * @return bool
  458. */
  459. protected function handlerShouldBeQueued($class)
  460. {
  461. try {
  462. return (new ReflectionClass($class))->implementsInterface(
  463. ShouldQueue::class
  464. );
  465. } catch (Exception) {
  466. return false;
  467. }
  468. }
  469. /**
  470. * Create a callable for putting an event handler on the queue.
  471. *
  472. * @param string $class
  473. * @param string $method
  474. * @return \Closure
  475. */
  476. protected function createQueuedHandlerCallable($class, $method)
  477. {
  478. return function () use ($class, $method) {
  479. $arguments = array_map(function ($a) {
  480. return is_object($a) ? clone $a : $a;
  481. }, func_get_args());
  482. if ($this->handlerWantsToBeQueued($class, $arguments)) {
  483. $this->queueHandler($class, $method, $arguments);
  484. }
  485. };
  486. }
  487. /**
  488. * Determine if the given event handler should be dispatched after all database transactions have committed.
  489. *
  490. * @param object|mixed $listener
  491. * @return bool
  492. */
  493. protected function handlerShouldBeDispatchedAfterDatabaseTransactions($listener)
  494. {
  495. return (($listener->afterCommit ?? null) ||
  496. $listener instanceof ShouldHandleEventsAfterCommit) &&
  497. $this->resolveTransactionManager();
  498. }
  499. /**
  500. * Create a callable for dispatching a listener after database transactions.
  501. *
  502. * @param mixed $listener
  503. * @param string $method
  504. * @return \Closure
  505. */
  506. protected function createCallbackForListenerRunningAfterCommits($listener, $method)
  507. {
  508. return function () use ($method, $listener) {
  509. $payload = func_get_args();
  510. $this->resolveTransactionManager()->addCallback(
  511. function () use ($listener, $method, $payload) {
  512. $listener->$method(...$payload);
  513. }
  514. );
  515. };
  516. }
  517. /**
  518. * Determine if the event handler wants to be queued.
  519. *
  520. * @param string $class
  521. * @param array $arguments
  522. * @return bool
  523. */
  524. protected function handlerWantsToBeQueued($class, $arguments)
  525. {
  526. $instance = $this->container->make($class);
  527. if (method_exists($instance, 'shouldQueue')) {
  528. return $instance->shouldQueue($arguments[0]);
  529. }
  530. return true;
  531. }
  532. /**
  533. * Queue the handler class.
  534. *
  535. * @param string $class
  536. * @param string $method
  537. * @param array $arguments
  538. * @return void
  539. */
  540. protected function queueHandler($class, $method, $arguments)
  541. {
  542. [$listener, $job] = $this->createListenerAndJob($class, $method, $arguments);
  543. $connection = $this->resolveQueue()->connection(method_exists($listener, 'viaConnection')
  544. ? (isset($arguments[0]) ? $listener->viaConnection($arguments[0]) : $listener->viaConnection())
  545. : $listener->connection ?? null);
  546. $queue = method_exists($listener, 'viaQueue')
  547. ? (isset($arguments[0]) ? $listener->viaQueue($arguments[0]) : $listener->viaQueue())
  548. : $listener->queue ?? null;
  549. $delay = method_exists($listener, 'withDelay')
  550. ? (isset($arguments[0]) ? $listener->withDelay($arguments[0]) : $listener->withDelay())
  551. : $listener->delay ?? null;
  552. is_null($delay)
  553. ? $connection->pushOn($queue, $job)
  554. : $connection->laterOn($queue, $delay, $job);
  555. }
  556. /**
  557. * Create the listener and job for a queued listener.
  558. *
  559. * @param string $class
  560. * @param string $method
  561. * @param array $arguments
  562. * @return array
  563. */
  564. protected function createListenerAndJob($class, $method, $arguments)
  565. {
  566. $listener = (new ReflectionClass($class))->newInstanceWithoutConstructor();
  567. return [$listener, $this->propagateListenerOptions(
  568. $listener, new CallQueuedListener($class, $method, $arguments)
  569. )];
  570. }
  571. /**
  572. * Propagate listener options to the job.
  573. *
  574. * @param mixed $listener
  575. * @param \Illuminate\Events\CallQueuedListener $job
  576. * @return mixed
  577. */
  578. protected function propagateListenerOptions($listener, $job)
  579. {
  580. return tap($job, function ($job) use ($listener) {
  581. $data = array_values($job->data);
  582. if ($listener instanceof ShouldQueueAfterCommit) {
  583. $job->afterCommit = true;
  584. } else {
  585. $job->afterCommit = property_exists($listener, 'afterCommit') ? $listener->afterCommit : null;
  586. }
  587. $job->backoff = method_exists($listener, 'backoff') ? $listener->backoff(...$data) : ($listener->backoff ?? null);
  588. $job->maxExceptions = $listener->maxExceptions ?? null;
  589. $job->retryUntil = method_exists($listener, 'retryUntil') ? $listener->retryUntil(...$data) : null;
  590. $job->shouldBeEncrypted = $listener instanceof ShouldBeEncrypted;
  591. $job->timeout = $listener->timeout ?? null;
  592. $job->failOnTimeout = $listener->failOnTimeout ?? false;
  593. $job->tries = $listener->tries ?? null;
  594. $job->through(array_merge(
  595. method_exists($listener, 'middleware') ? $listener->middleware(...$data) : [],
  596. $listener->middleware ?? []
  597. ));
  598. });
  599. }
  600. /**
  601. * Remove a set of listeners from the dispatcher.
  602. *
  603. * @param string $event
  604. * @return void
  605. */
  606. public function forget($event)
  607. {
  608. if (str_contains($event, '*')) {
  609. unset($this->wildcards[$event]);
  610. } else {
  611. unset($this->listeners[$event]);
  612. }
  613. foreach ($this->wildcardsCache as $key => $listeners) {
  614. if (Str::is($event, $key)) {
  615. unset($this->wildcardsCache[$key]);
  616. }
  617. }
  618. }
  619. /**
  620. * Forget all of the pushed listeners.
  621. *
  622. * @return void
  623. */
  624. public function forgetPushed()
  625. {
  626. foreach ($this->listeners as $key => $value) {
  627. if (str_ends_with($key, '_pushed')) {
  628. $this->forget($key);
  629. }
  630. }
  631. }
  632. /**
  633. * Get the queue implementation from the resolver.
  634. *
  635. * @return \Illuminate\Contracts\Queue\Queue
  636. */
  637. protected function resolveQueue()
  638. {
  639. return call_user_func($this->queueResolver);
  640. }
  641. /**
  642. * Set the queue resolver implementation.
  643. *
  644. * @param callable $resolver
  645. * @return $this
  646. */
  647. public function setQueueResolver(callable $resolver)
  648. {
  649. $this->queueResolver = $resolver;
  650. return $this;
  651. }
  652. /**
  653. * Get the database transaction manager implementation from the resolver.
  654. *
  655. * @return \Illuminate\Database\DatabaseTransactionsManager|null
  656. */
  657. protected function resolveTransactionManager()
  658. {
  659. return call_user_func($this->transactionManagerResolver);
  660. }
  661. /**
  662. * Set the database transaction manager resolver implementation.
  663. *
  664. * @param callable $resolver
  665. * @return $this
  666. */
  667. public function setTransactionManagerResolver(callable $resolver)
  668. {
  669. $this->transactionManagerResolver = $resolver;
  670. return $this;
  671. }
  672. /**
  673. * Gets the raw, unprepared listeners.
  674. *
  675. * @return array
  676. */
  677. public function getRawListeners()
  678. {
  679. return $this->listeners;
  680. }
  681. }