src/Products/NotificationsBundle/Subscriber/BroadcastSubscriber.php line 50

Open in your IDE?
  1. <?php
  3. namespace Products\NotificationsBundle\Subscriber;
  5. use Cms\CoreBundle\Util\Doctrine\EntityManager;
  6. use Platform\QueueBundle\Event\AsyncEvent;
  7. use Products\NotificationsBundle\Entity\AbstractNotification;
  8. use Products\NotificationsBundle\Entity\Job;
  9. use Products\NotificationsBundle\Entity\Notifications\Channels\ChannelsInterface;
  10. use Products\NotificationsBundle\Service\Switchboard;
  11. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  13. final class BroadcastSubscriber implements EventSubscriberInterface
  14. {
  15.     /**
  16.      * @var EntityManager
  17.      */
  18.     protected EntityManager $em;
  20.     /**
  21.      * @var Switchboard
  22.      */
  23.     protected Switchboard $switchboard;
  25.     /**
  26.      * @param EntityManager $em
  27.      * @param Switchboard $switchboard
  28.      */
  29.     public function __construct(EntityManager $em, Switchboard $switchboard)
  30.     {
  31.         $this->em = $em;
  32.         $this->switchboard = $switchboard;
  33.     }
  35.     /**
  36.      * {@inheritDoc}
  37.      */
  38.     public static function getSubscribedEvents(): array
  39.     {
  40.         return [
  41.             Switchboard::EVENTS__BROADCAST_JOB => ['onBroadcastJob', 0],
  42.             Switchboard::EVENTS__BROADCAST_CHANNEL => ['onBroadcastChannel', 0],
  43.             Switchboard::EVENTS__BROADCAST_ITEM => ['onBroadcastItem', 0],
  44.         ];
  45.     }
  47.     /**
  48.      * @param AsyncEvent $event
  49.      */
  50.     public function onBroadcastJob(AsyncEvent $event)
  51.     {
  52.         $job = $this->em->getRepository(Job::class)->find(
  53.             $event->getBody()->get('job')
  54.         );
  55.         if ( ! $job instanceof Job) {
  56.             throw new \Exception();
  57.         }
  58.         $result = $this->switchboard->broadcastJob($job);
  59.         $event->getOutput()->writeln(sprintf(
  60.             'Broadcasting to %s channels.',
  61.             $result,
  62.         ));
  63.     }
  65.     /**
  66.      * @param AsyncEvent $event
  67.      */
  68.     public function onBroadcastChannel(AsyncEvent $event)
  69.     {
  70.         $job = $this->em->getRepository(Job::class)->find(
  71.             $event->getBody()->get('job')
  72.         );
  73.         if ( ! $job instanceof Job) {
  74.             throw new \Exception();
  75.         }
  76.         $result = $this->switchboard->broadcastChannel(
  77.             $job,
  78.             $channel = $event->getBody()->get('channel')
  79.         );
  80.         $event->getOutput()->writeln(sprintf(
  81.             'Broadcasting %s items to channel %s.',
  82.             $result,
  83.             array_search($channel, ChannelsInterface::CHANNELS),
  84.         ));
  85.     }
  87.     /**
  88.      * @param AsyncEvent $event
  89.      */
  90.     public function onBroadcastItem(AsyncEvent $event)
  91.     {
  92.         $job = $this->em->getRepository(Job::class)->find(
  93.             $event->getBody()->get('job')
  94.         );
  95.         if ( ! $job instanceof Job) {
  96.             throw new \Exception();
  97.         }
  98.         $this->switchboard->broadcastItem(
  99.             $job,
  100.             $event->getBody()->get('channel'),
  101.             $event->getBody()->get('item')// this will still be a scalar value, the handler for the channel should know how to load it
  102.         );
  103.     }
  104. }