. * * The interactive user interfaces in modified source and object code versions * of this program must display Appropriate Legal Notices, as required under * Section 5 of the GNU Affero General Public License version 3. * * In accordance with Section 7(b) of the GNU Affero General Public License version 3, * these Appropriate Legal Notices must retain the display of the "EspoCRM" word. ************************************************************************/ namespace Espo\Core\Utils\Event; use Closure; use RuntimeException; use Throwable; class CrossInstanceEventDispatcher { /** * @var array, (Closure(CrossInstanceEvent, Context): void)[]> */ private array $callbacks = []; private bool $isTransportSubscribed = false; public function __construct( private EventTransport $transport, private OriginProvider $originProvider, ) {} /** * @param class-string $className * @param Closure(CrossInstanceEvent, Context): void $callback */ public function subscribe(string $className, Closure $callback): void { $this->ensureSubscribeTransport(); $this->callbacks[$className] ??= []; $this->callbacks[$className][] = $callback; } /** * @param class-string $className * @param Closure(CrossInstanceEvent, Context): void $callback */ public function unsubscribe(string $className, Closure $callback): void { if (!array_key_exists($className, $this->callbacks)) { return; } $list = &$this->callbacks[$className]; $index = array_search($callback, $list); if ($index !== false) { unset($list[$index]); $list = array_values($list); } } public function dispatch(CrossInstanceEvent $event): void { $envelope = new Envelope( eventClassName: $event::class, payload: $event->toPayload(), origin: $this->originProvider->get(), ); $this->transport->publish($envelope); } private function ensureSubscribeTransport(): void { if ($this->isTransportSubscribed) { return; } $this->transport->subscribe(fn (Envelope $envelope) => $this->transportCallback($envelope)); $this->isTransportSubscribed = true; } private function transportCallback(Envelope $envelope): void { if ($envelope->origin === $this->originProvider->get()) { return; } $className = $envelope->eventClassName; if (!is_subclass_of($className, CrossInstanceEvent::class)) { throw new RuntimeException("Non-valid event class name."); } $callbacks = $this->callbacks[$className] ?? []; if ($callbacks === []) { return; } try { $event = $className::fromPayload($envelope->payload); } catch (Throwable $e) { throw new RuntimeException("Could not hydrate event '$className'.", previous: $e); } $context = new Context( isLocal: false, origin: $envelope->origin, ); foreach ($callbacks as $callback) { $callback($event, $context); } } }