src/App/RabbitMq/TasksPool.php line 57

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace App\RabbitMq;
  5. use App\RabbitMq\Processing\PushProcessingEvent;
  6. use \DomainException;
  7. use Sindrive\RabbitMqTaskBundle\Service\TaskHandler;
  8. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  9. use const \SORT_NUMERIC;
  11. class TasksPool implements EventSubscriberInterface
  12. {
  14.     /**
  15.      * @var mixed[][][][][]
  16.      */
  17.     private array $tasks = [];
  19.     private bool $sent = false;
  21.     private TaskHandler $rabbitmqProducer;
  23.     public function __construct(TaskHandler $producer)
  24.     {
  25.         $this->rabbitmqProducer = $producer;
  26.     }
  28.     /**
  29.      * @param mixed[] $arguments
  30.      */
  31.     public function addTask(
  32.         string $service,
  33.         string $method,
  34.         string $entityClass,
  35.         int $entityId,
  36.         array $arguments = [],
  37.         bool $isImageHandler = false,
  38.         int $priority = 255
  39.     ): void
  40.     {
  41.         if ($entityId < 1) {
  42.             throw new DomainException('Entity without an id is not supported');
  44.             // how do you want to handle null/string/negative IDs in RMQ?
  45.         }
  47.         $this->sent = false;
  49.         $this->tasks[$priority][$entityClass][$entityId][] = [
  50.             'service' => $service,
  51.             'method' => $method,
  52.             'arguments' => $arguments,
  53.             'imageHandler' => $isImageHandler,
  54.         ];
  55.     }
  57.     public function onProcess(PushProcessingEvent $event, string $eventName): void
  58.     {
  59.         $this->addTask(
  60.             $event->getService(),
  61.             $event->getMethod(),
  62.             $event->getEntityClass(),
  63.             $event->getEntityId(),
  64.             $event->getArguments(),
  65.             $event->isImageHandler(),
  66.             $event->getPriority()
  67.         );
  68.     }
  70.     public function send(): void
  71.     {
  72.         if (count($this->tasks) > 0) {
  73.             $tasks = $this->tasks;
  74.             krsort($tasks, SORT_NUMERIC);
  75.             // need to clean up tasks before sending out a RabbitMQ task, because it can call this method and we will
  76.             // pile up a queue of duplicate tasks
  77.             $this->tasks = [];
  78.             $this->rabbitmqProducer->sendTask(TasksProcessor::QUEUE, serialize($tasks));
  79.         }
  81.         $this->sent = true;
  82.     }
  84.     public function isSent(): bool
  85.     {
  86.         return $this->sent;
  87.     }
  89.     /**
  90.      * @return array<string, array<int|string, array<int|string, int|string>|int|string>|string>
  91.      */
  92.     public static function getSubscribedEvents(): array
  93.     {
  94.         return [
  95.             PushProcessingEvent::class => 'onProcess',
  96.         ];
  97.     }
  99. }