src/App/RabbitMq/Processing/ProcessingSubscriber.php line 31

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace App\RabbitMq\Processing;
  5. use App\Entity\AbstractDelayed;
  6. use DateTime;
  7. use Doctrine\ORM\EntityManagerInterface;
  8. use Doctrine\Persistence\ManagerRegistry;
  9. use InvalidArgumentException;
  10. use Psr\Log\LoggerInterface;
  11. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  13. class ProcessingSubscriber implements EventSubscriberInterface
  14. {
  16.     private ManagerRegistry $doctrine;
  18.     private LoggerInterface $logger;
  20.     public function __construct(ManagerRegistry $doctrine, LoggerInterface $logger)
  21.     {
  22.         $this->doctrine = $doctrine;
  23.         $this->logger = $logger;
  24.     }
  26.     public function onPushProcess(PushProcessingEvent $event): void
  27.     {
  28.         $this->onProcess($event, true);
  29.     }
  31.     public function onPopProcess(PopProcessingEvent $event): void
  32.     {
  33.         $this->onProcess($event, false);
  34.     }
  36.     private function onProcess(ProcessingEvent $event, bool $isPush): void
  37.     {
  38.         $entity = $event->getEntity();
  40.         if (!($entity instanceof AbstractDelayed)) {
  41.             return;
  42.         }
  44.         $this->logger->debug(($isPush ? 'Pushing ' : 'Popping ') . ' task ' . $event->getService() . ':' . $event->getMethod());
  46.         $this->delayedTasks($entity, $event->getEntityClass(), $isPush);
  47.     }
  49.     private function delayedTasks(AbstractDelayed $entity, string $entityClass, bool $increment): void
  50.     {
  51.         if (null !== $entity->getId()) {
  52.             $delayedTasksInc = ($increment ? 1 : 0) * 2 - 1;
  54.             /** @var EntityManagerInterface $manager */
  55.             $manager = $this->doctrine->getManager();
  56.             $qb = $manager->createQueryBuilder();
  58.             $qb->update($entityClass, 'e')
  59.                 ->set('e.delayedTasks', 'e.delayedTasks + :delayedTasksProcess')
  60.                 ->where('e.id = :entity_id')
  61.                 ->setParameter('delayedTasksProcess', $delayedTasksInc)
  62.                 ->setParameter('entity_id', $entity->getId())
  63.                 ->getQuery()->execute();
  65.             // this value will not be saved to database, but will allow us to see if tasks were scheduled
  66.             $entity->incrementTasks($delayedTasksInc);
  68.             if (0 === $entity->getDelayedTasks() + $entity->getTasksIncrement()) {
  69.                 $entity->setLastModified(new DateTime());
  70.             }
  71.         } else {
  72.             $increment
  73.                 ? $entity->addDelayedTask()
  74.                 : $entity->deductDelayedTask();
  75.         }
  76.     }
  78.     public function forceDelayedTasks(AbstractDelayed $entity, string $entityName, bool $increment, string $iKnow): void
  79.     {
  80.         // this thing was designed to track how many RabbitMQ tasks are left to process in
  81.         // \App\RabbitMq\TasksProcessor so overriding delayedTasks need to be done with
  82.         // EXTREME CAUTION
  83.         if ($iKnow !== 'Yes, i am sure i want to do this') {
  84.             throw new InvalidArgumentException('You should not do this then');
  85.         }
  87.         $this->delayedTasks($entity, $entityName, $increment);
  88.     }
  90.     /**
  91.      * @return array<string, array<int|string, array<int|string, int|string>|int|string>|string>
  92.      */
  93.     public static function getSubscribedEvents(): array
  94.     {
  95.         return [
  96.             PushProcessingEvent::class => 'onPushProcess',
  97.             PopProcessingEvent::class => 'onPopProcess',
  98.         ];
  99.     }
  101. }