Add new event when job is created

Doctrine/BaseDoctrineJobManager.php

@@ -4,6 +4,7 @@
 
 use Doctrine\Common\Persistence\ObjectManager;
 use Doctrine\Common\Persistence\ObjectRepository;
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Manager\ArchivableJobManager;
 use Dtc\QueueBundle\Manager\JobTimingManager;
 use Dtc\QueueBundle\Manager\RunManager;
@@ -28,17 +29,19 @@ abstract class BaseDoctrineJobManager extends ArchivableJobManager
      * @param JobTimingManager $jobTimingManager
      * @param ObjectManager    $objectManager
      * @param $jobClass
+     * @param EventDispatcher $eventDispatcher
      * @param $jobArchiveClass
      */
     public function __construct(
         RunManager $runManager,
         JobTimingManager $jobTimingManager,
         ObjectManager $objectManager,
         $jobClass,
+        EventDispatcher $eventDispatcher,
         $jobArchiveClass
     ) {
         $this->objectManager = $objectManager;
-        parent::__construct($runManager, $jobTimingManager, $jobClass, $jobArchiveClass);
+        parent::__construct($runManager, $jobTimingManager, $jobClass, $eventDispatcher, $jobArchiveClass);
     }
 
     protected function getFetchCount($totalCount)

EventDispatcher/Event.php

@@ -6,6 +6,7 @@
 
 class Event
 {
+    const POST_CREATE_JOB = 'queue.post_create_job';
     const PRE_JOB = 'queue.pre_job';
     const POST_JOB = 'queue.post_job';
 

Manager/AbstractJobManager.php

@@ -2,6 +2,7 @@
 
 namespace Dtc\QueueBundle\Manager;
 
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Exception\UnsupportedException;
 use Dtc\QueueBundle\Model\BaseJob;
 use Dtc\QueueBundle\Model\Job;
@@ -11,12 +12,16 @@ abstract class AbstractJobManager implements JobManagerInterface
     protected $jobTiminigManager;
     protected $jobClass;
     protected $runManager;
-
-    public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass)
+
+    /** @var EventDispatcher $eventDispatcher */
+    protected $eventDispatcher;
+
+    public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass, EventDispatcher $eventDispatcher)
     {
         $this->runManager = $runManager;
         $this->jobTiminigManager = $jobTimingManager;
         $this->jobClass = $jobClass;
+        $this->eventDispatcher = $eventDispatcher;
     }
 
     /**

Manager/ArchivableJobManager.php

@@ -2,6 +2,8 @@
 
 namespace Dtc\QueueBundle\Manager;
 
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
+
 abstract class ArchivableJobManager extends StallableJobManager
 {
     /**
@@ -15,16 +17,18 @@ abstract class ArchivableJobManager extends StallableJobManager
      * @param RunManager       $runManager
      * @param JobTimingManager $jobTimingManager
      * @param $jobClass
+     * @param EventDispatcher $eventDispatcher
      * @param $jobArchiveClass
      */
     public function __construct(
         RunManager $runManager,
         JobTimingManager $jobTimingManager,
         $jobClass,
+        EventDispatcher $eventDispatcher,
         $jobArchiveClass
     ) {
         $this->jobArchiveClass = $jobArchiveClass;
-        parent::__construct($runManager, $jobTimingManager, $jobClass);
+        parent::__construct($runManager, $jobTimingManager, $jobClass, $eventDispatcher);
     }
 
     /**

Manager/RetryableJobManager.php

@@ -2,6 +2,7 @@
 
 namespace Dtc\QueueBundle\Manager;
 
+use Dtc\QueueBundle\EventDispatcher\Event;
 use Dtc\QueueBundle\Model\BaseJob;
 use Dtc\QueueBundle\Model\RetryableJob;
 use Dtc\QueueBundle\Model\Job;
@@ -48,7 +49,12 @@ public function save(Job $job)
         $this->recordTiming($job);
         $job->setUpdatedAt(new \DateTime());
 
-        return $this->retryableSave($job);
+        $result = $this->retryableSave($job);
+
+        $event = new Event($job);
+        $this->eventDispatcher->dispatch(Event::POST_CREATE_JOB, $event);
+
+        return $result;
     }
 
     /**

Model/Job.php

@@ -2,6 +2,8 @@
 
 namespace Dtc\QueueBundle\Model;
 
+use Dtc\QueueBundle\Manager\JobManagerInterface;
+
 class Job extends BaseJob
 {
     const STATUS_EXPIRED = 'expired';
@@ -28,7 +30,7 @@ public function __call($method, $args)
         }
 
         $job = $this->jobManager->save($this);
-
+       
         return $job;
     }
 

RabbitMQ/JobManager.php

@@ -2,6 +2,7 @@
 
 namespace Dtc\QueueBundle\RabbitMQ;
 
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Manager\SaveableTrait;
 use Dtc\QueueBundle\Manager\VerifyTrait;
 use Dtc\QueueBundle\Model\BaseJob;
@@ -39,11 +40,11 @@ class JobManager extends PriorityJobManager
     protected $hostname;
     protected $pid;
 
-    public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass)
+    public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass, EventDispatcher $eventDispatcher)
     {
         $this->hostname = gethostname() ?: '';
         $this->pid = getmypid();
-        parent::__construct($runManager, $jobTimingManager, $jobClass);
+        parent::__construct($runManager, $jobTimingManager, $jobClass, $eventDispatcher);
     }
 
     /**

Redis/BaseJobManager.php

@@ -2,6 +2,7 @@
 
 namespace Dtc\QueueBundle\Redis;
 
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Manager\JobIdTrait;
 use Dtc\QueueBundle\Manager\JobTimingManager;
 use Dtc\QueueBundle\Manager\PriorityJobManager;
@@ -31,13 +32,13 @@ abstract class BaseJobManager extends PriorityJobManager
     /**
      * @param string $cacheKeyPrefix
      */
-    public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass, $cacheKeyPrefix)
+    public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass, EventDispatcher $eventDispatcher, $cacheKeyPrefix)
     {
         $this->cacheKeyPrefix = $cacheKeyPrefix;
         $this->hostname = gethostname() ?: '';
         $this->pid = getmypid();
 
-        parent::__construct($runManager, $jobTimingManager, $jobClass);
+        parent::__construct($runManager, $jobTimingManager, $jobClass, $eventDispatcher);
     }
 
     public function setRedis(RedisInterface $redis)

Resources/config/queue.yml

@@ -76,6 +76,7 @@ services:
             - "@dtc_queue.manager.job_timing"
             - '@dtc_queue.document_manager'
             - '%dtc_queue.class.job%'
+            - '@dtc_queue.event_dispatcher'
             - '%dtc_queue.class.job_archive%'
         calls:
             - ["setMaxPriority", ['%dtc_queue.priority.max%']]
@@ -94,6 +95,7 @@ services:
             - "@dtc_queue.manager.run"
             - "@dtc_queue.manager.job_timing"
             - '%dtc_queue.class.job%'
+            - '@dtc_queue.event_dispatcher'
         calls:
             - ["setDefaultMaxRetries", ['%dtc_queue.retry.max.retries%']]
             - ["setDefaultMaxFailures", ['%dtc_queue.retry.max.failures%']]
@@ -107,6 +109,7 @@ services:
             - "@dtc_queue.manager.run"
             - "@dtc_queue.manager.job_timing"
             - "%dtc_queue.class.job%"
+            - '@dtc_queue.event_dispatcher'
         calls:
             - ["setMaxPriority", ['%dtc_queue.priority.max%']]
             - ["setPriorityDirection", ['%dtc_queue.priority.direction%']]
@@ -123,6 +126,7 @@ services:
             - "@dtc_queue.manager.run"
             - "@dtc_queue.manager.job_timing"
             - "%dtc_queue.class.job%"
+            - '@dtc_queue.event_dispatcher'
             - "%dtc_queue.redis.prefix%"
         calls:
             - ["setMaxPriority", ['%dtc_queue.priority.max%']]
@@ -199,6 +203,7 @@ services:
             - "@dtc_queue.manager.job_timing"
             - '@dtc_queue.entity_manager'
             - '%dtc_queue.class.job%'
+            - '@dtc_queue.event_dispatcher'
             - '%dtc_queue.class.job_archive%'
         calls:
             - ["setMaxPriority", ['%dtc_queue.priority.max%']]

Tests/Beanstalkd/JobManagerTest.php

@@ -3,6 +3,7 @@
 namespace Dtc\QueueBundle\Tests\Beanstalkd;
 
 use Dtc\QueueBundle\Beanstalkd\JobManager;
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Manager\JobTimingManager;
 use Dtc\QueueBundle\Manager\RunManager;
 use Dtc\QueueBundle\Tests\FibonacciWorker;
@@ -32,7 +33,8 @@ public static function setUpBeforeClass()
         self::$beanstalkd = new Pheanstalk($host);
         self::$jobTimingManager = new JobTimingManager($jobTimingClass, false);
         self::$runManager = new RunManager($runClass);
-        self::$jobManager = new JobManager(self::$runManager, self::$jobTimingManager, $className);
+        self::$eventDispatcher = new EventDispatcher();
+        self::$jobManager = new JobManager(self::$runManager, self::$jobTimingManager, $className, self::$eventDispatcher);
         self::$jobManager->setBeanstalkd(self::$beanstalkd);
         self::$worker = new FibonacciWorker();
 

Tests/Command/CommandTrait.php

@@ -2,6 +2,7 @@
 
 namespace Dtc\QueueBundle\Tests\Command;
 
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Model\Job;
 use Dtc\QueueBundle\Model\JobTiming;
 use Dtc\QueueBundle\Manager\JobTimingManager;
@@ -86,7 +87,8 @@ protected function runStubCommand($className, $params, $call, $expectedResult =
 
         $jobTimingManager = new JobTimingManager(JobTiming::class, false);
         $runManager = new StubRunManager($jobTimingManager, \Dtc\QueueBundle\Model\Run::class);
-        $jobManager = new StubJobManager($runManager, $jobTimingManager, Job::class);
+        $eventDispatcher = new EventDispatcher();
+        $jobManager = new StubJobManager($runManager, $jobTimingManager, Job::class, $eventDispatcher);
         $container = new Container();
         $container->set('dtc_queue.manager.job', $jobManager);
         $container->set('dtc_queue.manager.run', $runManager);

Tests/Command/CountCommandTest.php

@@ -3,6 +3,7 @@
 namespace Dtc\QueueBundle\Tests\Command;
 
 use Dtc\QueueBundle\Command\CountCommand;
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Model\Job;
 use Dtc\QueueBundle\Model\JobTiming;
 use Dtc\QueueBundle\Manager\JobTimingManager;
@@ -22,7 +23,8 @@ public function testCountCommand()
         $container = new Container();
         $jobTimingManager = new JobTimingManager(JobTiming::class, false);
         $runManager = new RunManager($jobTimingManager, Run::class);
-        $container->set('dtc_queue.manager.job', new StubJobManager($runManager, $jobTimingManager, Job::class));
+        $eventDispatcher = new EventDispatcher();
+        $container->set('dtc_queue.manager.job', new StubJobManager($runManager, $jobTimingManager, Job::class, $eventDispatcher));
         $this->runCommandException(CountCommand::class, $container, []);
     }
 

Tests/Command/CreateJobCommandTest.php

@@ -22,7 +22,8 @@ public function testCreateJobCommand()
     {
         $jobTimingManager = new JobTimingManager(JobTimingManager::class, false);
         $runManager = new RunManager($jobTimingManager, Run::class);
-        $jobManager = new StubJobManager($runManager, $jobTimingManager, Job::class);
+        $eventDispatcher = new EventDispatcher();
+        $jobManager = new StubJobManager($runManager, $jobTimingManager, Job::class, $eventDispatcher);
         $container = new Container();
         $container->set('dtc_queue.manager.job', $jobManager);
         $this->runCommandException(CreateJobCommand::class, $container, ['worker_name' => 'fibonacci', 'method' => 'fibonacci', 'args' => [1]]);

Tests/Doctrine/DoctrineJobManagerTest.php

@@ -6,6 +6,7 @@
 use Doctrine\ORM\EntityManager;
 use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
 use Dtc\QueueBundle\Doctrine\DtcQueueListener;
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Manager\StallableJobManager;
 use Dtc\QueueBundle\Model\BaseJob;
 use Dtc\QueueBundle\Model\Job;
@@ -52,8 +53,9 @@ public static function setUpBeforeClass()
     {
         self::$jobTimingManager = new self::$jobTimingManagerClass(self::$objectManager, self::$jobTimingClass, true);
         self::$runManager = new self::$runManagerClass(self::$objectManager, self::$runClass, self::$runArchiveClass);
+        self::$eventDispatcher = new EventDispatcher();
         /** @var JobManager|\Dtc\QueueBundle\ORM\JobManager $jobManager */
-        $jobManager = new self::$jobManagerClass(self::$runManager, self::$jobTimingManager, self::$objectManager, self::$objectName, self::$archiveObjectName);
+        $jobManager = new self::$jobManagerClass(self::$runManager, self::$jobTimingManager, self::$objectManager, self::$objectName,  self::$eventDispatcher, self::$archiveObjectName);
         self::$jobManager = $jobManager;
         $jobManager->setMaxPriority(255);
 
@@ -341,7 +343,7 @@ public function testResetExceptionJobs()
         self::assertEquals(BaseJob::STATUS_EXCEPTION, $result->getStatus());
         if ($objectManager instanceof EntityManager) {
             JobManagerTest::createObjectManager();
-            $jobManager = new self::$jobManagerClass(self::$runManager, self::$jobTimingManager, self::$objectManager, self::$objectName, self::$archiveObjectName);
+            $jobManager = new self::$jobManagerClass(self::$runManager, self::$jobTimingManager, self::$objectManager, self::$objectName, self::$eventDispatcher, self::$archiveObjectName);
             $jobManager->getObjectManager()->clear();
             $objectManager = $jobManager->getObjectManager();
         }

Tests/EventDispatcher/EventDispatcherTest.php

@@ -16,16 +16,28 @@ public function testEventDispatcher()
 
         self::assertFalse($eventDispatcher->hasListeners(Event::POST_JOB));
         self::assertFalse($eventDispatcher->hasListeners(Event::PRE_JOB));
+        self::assertFalse($eventDispatcher->hasListeners(Event::POST_CREATE_JOB));
 
         $eventDispatcher->addSubscriber($eventSubscriber);
         self::assertTrue($eventDispatcher->hasListeners(Event::POST_JOB));
         self::assertTrue($eventDispatcher->hasListeners(Event::PRE_JOB));
+        self::assertTrue($eventDispatcher->hasListeners(Event::POST_CREATE_JOB));
 
         $job = new Job();
         $event = new Event($job);
 
         self::assertEmpty($eventSubscriber->getPostJobCalled());
         self::assertEmpty($eventSubscriber->getPreJobCalled());
+        self::assertEmpty($eventSubscriber->getPostCreateJobCalled());
+
+        $eventDispatcher->dispatch(Event::POST_CREATE_JOB, $event);
+
+        self::assertEmpty($eventSubscriber->getPostJobCalled());
+        self::assertEmpty($eventSubscriber->getPreJobCalled());
+        $postCreateJobCalled = $eventSubscriber->getPostCreateJobCalled();
+        self::assertNotEmpty($postCreateJobCalled);
+        $dispatchedEvent = $postCreateJobCalled[0];
+        self::assertEquals($event, $dispatchedEvent);
 
         $eventDispatcher->dispatch(Event::PRE_JOB, $event);
 

Tests/EventDispatcher/StubEventSubscriber.php

@@ -7,6 +7,7 @@
 
 class StubEventSubscriber implements EventSubscriberInterface
 {
+    protected $postCreateJobCalled;
     protected $preJobCalled;
     protected $postJobCalled;
 
@@ -20,6 +21,11 @@ public function postJob(Event $event)
         $this->postJobCalled[] = $event;
     }
 
+    public function postCreateJob(Event $event)
+    {
+        $this->postCreateJobCalled[] = $event;
+    }
+
     public function getPreJobCalled()
     {
         return $this->preJobCalled;
@@ -30,9 +36,15 @@ public function getPostJobCalled()
         return $this->postJobCalled;
     }
 
+    public function getPostCreateJobCalled()
+    {
+        return $this->postCreateJobCalled;
+    }
+
     public static function getSubscribedEvents()
     {
         return [
+            Event::POST_CREATE_JOB => 'postCreateJob',
             Event::PRE_JOB => 'preJob',
             Event::POST_JOB => 'postJob',
         ];

Tests/Manager/BaseJobManagerTest.php

@@ -2,6 +2,7 @@
 
 namespace Dtc\QueueBundle\Tests\Manager;
 
+use Dtc\QueueBundle\EventDispatcher\EventDispatcher;
 use Dtc\QueueBundle\Manager\JobManagerInterface;
 use Dtc\QueueBundle\Manager\PriorityJobManager;
 use Dtc\QueueBundle\Manager\RunManager;
@@ -30,6 +31,9 @@ abstract class BaseJobManagerTest extends TestCase
     /** @var JobTimingManager */
     public static $jobTimingManager;
 
+    /** @var EventDispatcher */
+    public static $eventDispatcher;
+
     public static function setUpBeforeClass()
     {
         self::$jobClass = self::$jobManager->getJobClass();