src/Messenger/EventListener/WorkersRegistryEventListener.php line 72

Open in your IDE?
  1. <?php
  3. /**
  4.  * Copyright (c) 2011-present Qualiteam software Ltd. All rights reserved.
  5.  * See https://www.x-cart.com/license-agreement.html for license details.
  6.  */
  8. namespace XCart\Messenger\EventListener;
  10. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  11. use Symfony\Component\Filesystem\Filesystem;
  12. use Symfony\Component\Messenger\Event\WorkerRunningEvent;
  13. use Symfony\Component\Messenger\Event\WorkerStartedEvent;
  14. use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
  15. use XCart\Messenger\Event\WorkerPingEvent;
  17. final class WorkersRegistryEventListener implements EventSubscriberInterface
  18. {
  19.     private Filesystem $filesystem;
  21.     private string $workersPoolPath;
  23.     /**
  24.      * @var false|int
  25.      */
  26.     private $pid;
  28.     public function __construct(Filesystem $filesystem, string $workersPoolPath)
  29.     {
  30.         $this->filesystem = $filesystem;
  31.         $this->workersPoolPath = $workersPoolPath;
  33.         $this->pid = getmypid();
  34.     }
  36.     /**
  37.      * @return array<string, string>
  38.      */
  39.     public static function getSubscribedEvents()
  40.     {
  41.         return [
  42.             WorkerStartedEvent::class => 'onWorkerStarted',
  43.             WorkerStoppedEvent::class => 'onWorkerStopped',
  44.             WorkerRunningEvent::class => 'onWorkerRunning',
  45.             WorkerPingEvent::class    => 'onWorkerPing',
  46.         ];
  47.     }
  49.     public function onWorkerStarted(): void
  50.     {
  51.         $this->dumpWorkerFile();
  53.         if (function_exists('pcntl_signal')) {
  54.             pcntl_signal(\SIGINT, function () {
  55.                 $this->onWorkerStopped();
  56.                 // We should only write remove our flag and exit, should not resume execution or wait for graceful stop
  57.                 exit();
  58.             });
  59.         }
  60.     }
  62.     public function onWorkerRunning(): void
  63.     {
  64.         $this->dumpWorkerFile();
  65.     }
  67.     public function onWorkerPing(): void
  68.     {
  69.         $this->dumpWorkerFile();
  70.     }
  72.     public function onWorkerStopped(): void
  73.     {
  74.         $this->filesystem->remove($this->workersPoolPath . $this->pid);
  75.     }
  77.     protected function dumpWorkerFile(): void
  78.     {
  79.         $this->filesystem->dumpFile($this->workersPoolPath . $this->pid, microtime(true));
  80.     }
  81. }