src/AppBundle/Worker/WorkerScheduler.php line 109

Open in your IDE?
  1. <?php
  3. namespace AppBundle\Worker;
  5. use AppBundle\Integration\Aws\Sqs;
  6. use Doctrine\ORM\EntityManagerInterface;
  7. use Symfony\Component\Console\ConsoleEvents;
  8. use Symfony\Component\DependencyInjection\Container;
  9. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  10. use Symfony\Component\HttpKernel\KernelEvents;
  12. /**
  13.  * Provide ability to schedule workers to run at a later time. In development with no
  14.  *   background worker environment, execute the worker immediately.
  15.  *
  16.  */
  17. class WorkerScheduler implements EventSubscriberInterface
  18. {
  19.     public static function getSubscribedEvents()
  20.     {
  21.         return array(
  22.             ConsoleEvents::TERMINATE => 'onTermination',
  23.             KernelEvents::TERMINATE => 'onTermination'
  24.         );
  25.     }
  27.     /** @var Sqs */
  28.     protected $sqs;
  30.     /** @var WorkerManager */
  31.     protected $manager;
  33.     /** @var EntityManagerInterface */
  34.     protected $em;
  36.     /** @var Container */
  37.     protected $container;
  39.     /** @var bool */
  40.     protected $immediateExecution = true;
  42.     /** @var array */
  43.     protected $pending = array();
  45.     /**
  46.      * @param Sqs $sqs
  47.      * @param WorkerManager $manager
  48.      * @param EntityManagerInterface $em
  49.      */
  50.     public function __construct(Sqs $sqs, WorkerManager $manager, EntityManagerInterface $em, Container $container)
  51.     {
  52.         $this->sqs = $sqs;
  53.         $this->manager = $manager;
  54.         $this->em = $em;
  55.         $this->container = $container;
  56.     }
  58.     /**
  59.      * @param bool $value
  60.      */
  61.     public function allowImmediateExecution($value)
  62.     {
  63.         $this->immediateExecution = (bool) $value;
  64.     }
  66.     /**
  67.      * @param string $jobCode
  68.      * @param array $data
  69.      */
  70.     public function schedule($jobCode, $data = array())
  71.     {
  72.         
  73.         if ($this->sqs->hasQueueUrl()) {
  74.             
  75.             $this->sqs->send(array_merge(
  76.                 $data,
  77.                 array('worker' => $jobCode)
  78.             ));
  79.         }
  80.         else {
  81.             
  82.             
  83.             $this->immediate($jobCode, $data);
  84.         }
  85.     }
  87.     /**
  88.      * @param string $jobCode
  89.      * @param array $data
  90.      */
  91.     public function scheduleBatch($jobCode, $data = array())
  92.     {
  93.         if ($this->sqs->hasQueueUrl()) {
  94.             $this->sqs->sendBatch(array_merge(
  95.                 $data,
  96.                 array('worker' => $jobCode)
  97.             ));
  98.         }
  99.         else {
  100.             $this->immediate($jobCode, $data);
  101.         }
  102.     }
  104.     public function flush()
  105.     {
  106.         $this->sqs->flushBatch();
  107.     }
  109.     public function onTermination()
  110.     {
  111.         // Execution post-termination might cause new pending tasks to be added, so keep going until there are no pending tasks
  112.         while (!empty($this->pending)) {
  113.             $jobs = $this->pending;
  114.             $this->pending = array();
  115.             foreach ($jobs as $info) {
  116.                 $this->run($info['code'], $info['data']);
  117.             }
  118.         }
  119.     }
  121.     /**
  122.      * It might not be possible to execute immediately in dev (triggered by
  123.      *   Doctrine event and unable to flush, or already in a job), so save for kernel.terminate
  124.      *
  125.      * @param string $jobCode
  126.      * @param array $data
  127.      */
  128.     protected function immediate($jobCode, $data)
  129.     {
  130.         
  131.         if ($this->immediateExecution && !$this->manager->isExecuting()) {
  132.             return $this->run($jobCode, $data);
  133.         }
  134.         
  135.         $this->pending[] = array('code' => $jobCode, 'data' => $data);
  136.     }
  138.     /**
  139.      * @param string $jobCode
  140.      * @param array $data
  141.      */
  142.     protected function run($jobCode, $data)
  143.     {
  144.         $worker = $this->container->get("workers.$jobCode");
  146.         // Some calling code still specifically designates entities
  147.         if (!empty($data['entities'])) {
  148.             $entities = $data['entities'];
  149.             unset($data['entities']);
  150.             $data = array_merge($data, $entities);
  151.         }
  153.         foreach ($data as $key => $val) {
  154.             if (is_array($val) && !empty($val['class']) && !empty($val['id']) && count(array_keys($val)) === 2) {
  155.                 // If called from legacy code, load entities
  156.                 $val = $this->em->find($val['class'], $val['id']);
  157.             }
  159.             $setter = 'set' . ucfirst($key);
  160.             if (!method_exists($worker, $setter)) {
  161.                 throw new \RuntimeException("Unable to set '$key' on worker '$jobCode'");
  162.             }
  163.             
  164.             $worker->{$setter}($val);
  165.         }
  167.         try {
  168.             
  169.             $this->manager->execute($worker);
  171.         } catch (\Exception $e) {
  172.             // The exception is already logged on the job, and immediate execution only happens in the dev environment
  173.         }
  174.     }
  175. }