src/Cms/CoreBundle/EventSubscriber/OneRoster/OneRosterPrepareSubscriber.php line 48

Open in your IDE?
  1. <?php
  3. namespace Cms\CoreBundle\EventSubscriber\OneRoster;
  5. use Cms\CoreBundle\Entity\AbstractOneRosterEntity;
  6. use Cms\CoreBundle\Entity\OneRosterJob;
  7. use Cms\CoreBundle\Events\OneRosterAdhocEvent;
  8. use Cms\CoreBundle\Events\OneRosterEvents;
  9. use Platform\QueueBundle\Event\AsyncEvent;
  11. /**
  12.  * Class OneRosterPrepareSubscriber
  13.  * @package Cms\CoreBundle\EventSubscriber\OneRoster
  14.  */
  15. final class OneRosterPrepareSubscriber extends AbstractOneRosterSubscriber
  16. {
  17.     /**
  18.      * {@inheritDoc}
  19.      */
  20.     public static function getSubscribedEvents(): array
  21.     {
  22.         return [
  23.             OneRosterEvents::EVENT__PREPARE => [
  24.                 ['discardableDisable', self::PRIORITY__FIRST],
  25.                 ['phaseStart', self::PRIORITY__FIRST],
  26.                 ['markStashed', 1],
  27.                 ['cleanStashed', 1],
  28.                 ['phaseFin', self::PRIORITY__LAST],
  29.                 ['phaseTrigger', self::PRIORITY__LAST],
  30.                 ['discardableEnable', self::PRIORITY__LAST],
  31.             ],
  32.             OneRosterEvents::EVENT__PREPARE__ADHOC => [
  33.                 ['discardableDisable', self::PRIORITY__FIRST],
  34.                 ['adhoc', 0],
  35.                 ['phaseFin', self::PRIORITY__LAST],
  36.                 ['phaseTrigger', self::PRIORITY__LAST],
  37.                 ['discardableEnable', self::PRIORITY__LAST],
  38.             ],
  39.         ];
  40.     }
  42.     /**
  43.      * In case we skip the stash phase, will update all the stash entries to be tied to this current job.
  44.      *
  45.      * @param AsyncEvent $event
  46.      * @return void
  47.      */
  48.     public function markStashed(AsyncEvent $event): void
  49.     {
  50.         // data should be an array with an id of a sync
  51.         $job = $this->loadJob($event);
  53.         // we only want to mark entities if we skipped the stash phase
  54.         // TODO: do we need to worry about signed/unsigned here at some point?
  55.         if ($job->getStartPhase() <= OneRosterJob::PHASES__STASH) {
  56.             return;
  57.         }
  59.         // run bulk query
  60.         $marks = $this->em->createQueryBuilder()
  61.             ->update(AbstractOneRosterEntity::class, 'entities')
  62.             ->set('entities.job', ':job')
  63.             ->setParameter('job', $job->getId())
  64.             ->andWhere('entities.tenant = :tenant')
  65.             ->setParameter('tenant', $job->getTenant()->getId())
  66.             ->andWhere('entities.sync = :sync')
  67.             ->setParameter('sync', $job->getSync()->getId())
  68.             ->getQuery()
  69.             ->execute();
  71.         // DEBUGGING
  72.         $event->getOutput()->writeln(sprintf(
  73.             'Marked %s stashed objects (stash phase was skipped) for sync #%s',
  74.             $marks,
  75.             $job->getIdentifier()
  76.         ));
  77.     }
  79.     /**
  80.      * Removes "old" data from the stashed entities that are no longer valid based on this current sync.
  81.      *
  82.      * @param AsyncEvent $event
  83.      * @return void
  84.      */
  85.     public function cleanStashed(AsyncEvent $event): void
  86.     {
  87.         // data should be an array with an id of a sync
  88.         $job = $this->loadJob($event);
  90.         // run bulk query
  91.         $removals = $this->em->createQueryBuilder()
  92.             ->delete(AbstractOneRosterEntity::class, 'entities')
  93.             ->andWhere('entities.tenant = :tenant')
  94.             ->setParameter('tenant', $job->getTenant()->getId())
  95.             ->andWhere('entities.sync = :sync')
  96.             ->setParameter('sync', $job->getSync()->getId())
  97.             ->andWhere('entities.job != :job')
  98.             ->setParameter('job', $job->getId())
  99.             ->getQuery()
  100.             ->execute();
  102.         // DEBUGGING
  103.         $event->getOutput()->writeln(sprintf(
  104.             'Removed %s stashed objects for sync #%s',
  105.             $removals,
  106.             $job->getIdentifier()
  107.         ));
  108.     }
  110.     /**
  111.      * In case we skip the stash phase, will update all the stash entries to be tied to this current job.
  112.      *
  113.      * @param AsyncEvent $event
  114.      * @return void
  115.      */
  116.     public function adhoc(AsyncEvent $event): void
  117.     {
  118.         // data should be an array with an id of a sync
  119.         $job = $this->loadJob($event);
  121.         // trigger the event to do further processing
  122.         $this->dispatcher->dispatch(
  123.             new OneRosterAdhocEvent(
  124.                 $event->getParam('payload'),
  125.                 $job,
  126.                 $event->getOutput()
  127.             ),
  128.             $event->getParam('event'),
  129.         );
  131.         // DEBUGGING
  132.         $event->getOutput()->writeln(sprintf(
  133.             'Dispatched adhoc event "%s" for sync #%s',
  134.             $event->getParam('event'),
  135.             $job->getIdentifier()
  136.         ));
  137.     }
  139.     /**
  140.      * {@inheritDoc}
  141.      */
  142.     public function phase(): int
  143.     {
  144.         return OneRosterJob::PHASES__PREPARE;
  145.     }
  147.     /**
  148.      * {@inheritDoc}
  149.      */
  150.     public function next(): ?int
  151.     {
  152.         return OneRosterJob::PHASES__PROCESS;
  153.     }
  154. }