Starting to integrate the EventDispatcher

Adding a few events to the main Worker class. Each event will be able to cancel a Job / Worker from running.
justinrainbow · Jul 22, 2012 · 33988d6 · 33988d6
1 parent e2a0f10
commit 33988d6
Show file tree

Hide file tree

Showing 13 changed files with 301 additions and 35 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1 +1,2 @@
 vendor
+phpunit.xml
diff --git a/src/Presque/AbstractJob.php b/src/Presque/AbstractJob.php
@@ -27,28 +27,27 @@ public function getStatus()
         return $this->status;
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function isSuccessful()
     {
         return StatusInterface::SUCCESS === $this->getStatus();
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function isError()
     {
         return StatusInterface::FAILED === $this->getStatus();
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function isActive()
     {
         return StatusInterface::RUNNING === $this->getStatus();
     }
-
-    public function isTrackable()
-    {
-
-    }
-
-    public function getMaxAttempts()
-    {
-
-    }
 }
diff --git a/src/Presque/Event/JobEvent.php b/src/Presque/Event/JobEvent.php
@@ -0,0 +1,62 @@
+<?php
+
+/*
+ * This file is part of the Presque package.
+ *
+ * (c) Justin Rainbow <justin.rainbow@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Presque\Event;
+
+use Presque\JobInterface;
+use Presque\QueueInterface;
+use Presque\WorkerInterface;
+
+class JobEvent extends Event
+{
+    private $worker;
+    private $canceled;
+
+    public function __construct(JobInterface $job, QueueInterface $queue, WorkerInterface $worker)
+    {
+        $this->job      = $job;
+        $this->queue    = $queue;
+        $this->worker   = $worker;
+        $this->canceled = false;
+    }
+
+    public function getJob()
+    {
+        return $this->job;
+    }
+
+    public function setJob(JobInterface $job)
+    {
+        $this->job = $job;
+
+        return $this;
+    }
+
+    public function getQueue()
+    {
+        return $this->queue;
+    }
+
+    public function getWorker()
+    {
+        return $this->worker;
+    }
+
+    public function cancel()
+    {
+        $this->canceled = true;
+    }
+
+    public function isCanceled()
+    {
+        return $this->canceled;
+    }
+}
diff --git a/src/Presque/Event/WorkerEvent.php b/src/Presque/Event/WorkerEvent.php
@@ -11,12 +11,31 @@
 
 namespace Presque\Event;
 
+use Presque\WorkerInterface;
+
 class WorkerEvent extends Event
 {
     private $worker;
+    private $canceled;
 
     public function __construct(WorkerInterface $worker)
     {
         $this->worker = $worker;
+        $this->canceled = false;
+    }
+
+    public function getWorker()
+    {
+        return $this->worker;
+    }
+
+    public function cancel()
+    {
+        $this->canceled = true;
+    }
+
+    public function isCanceled()
+    {
+        return $this->canceled;
     }
 }
diff --git a/src/Presque/Events.php b/src/Presque/Events.php
@@ -18,4 +18,6 @@ final class Events
 	const WORK_PAUSED = 'presque.work.paused';
 
 	const WORK_STOPPED = 'presque.work.stopped';
+
+	const JOB_STARTED = 'presque.job.started';
 }
diff --git a/src/Presque/Job.php b/src/Presque/Job.php
@@ -25,11 +25,20 @@ class Job extends AbstractJob
     private $reflMethod;
     private $instance;
 
+    /**
+     * {@inheritDoc}
+     */
     public static function create($class, array $args = array())
     {
         return new static($class, $args);
     }
 
+    /**
+     * @param string $class Class that will be performing
+     * @param array  $args  List of arguments to pass to the `$class->perform()` method
+     *
+     * @throws InvalidArgumentException If `$class` does not implement the required methods
+     */
     public function __construct($class, array $args = array())
     {
         try {
@@ -50,8 +59,15 @@ public function __construct($class, array $args = array())
         $method = $refl->getMethod('perform');
 
         if (!$method->isPublic()) {
+            $visibility = 'unknown';
+            if ($method->isPrivate()) {
+                $visibility = 'private';
+            } elseif ($method->isProtected()) {
+                $visibility = 'protected';
+            }
+
             throw new InvalidArgumentException(
-                'The "perform" method for class "' . $class . '" must be public, not ' . $method->getVisiblity()
+                'The "perform" method for class "' . $class . '" must be public, not ' . $visibility
             );
         }
 
@@ -69,6 +85,37 @@ public function __construct($class, array $args = array())
         $this->reflMethod = $method;
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    public function getClass()
+    {
+        return $this->class;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getArguments()
+    {
+        return $this->args;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getInstance()
+    {
+        if (!$this->instance) {
+            $this->instance = $this->reflClass->newInstance();
+        }
+
+        return $this->instance;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
     public function perform()
     {
         $this->lastResult = $this->lastError = null;
@@ -89,28 +136,4 @@ public function perform()
 
         return $this;
     }
-
-    public function getMaxAttempts()
-    {
-
-    }
-
-    public function getClass()
-    {
-        return $this->class;
-    }
-
-    public function getArguments()
-    {
-        return $this->args;
-    }
-
-    public function getInstance()
-    {
-        if (!$this->instance) {
-            $this->instance = $this->reflClass->newInstance();
-        }
-
-        return $this->instance;
-    }
 }
diff --git a/src/Presque/JobInterface.php b/src/Presque/JobInterface.php
@@ -13,17 +13,52 @@
 
 interface JobInterface
 {
+    /**
+     * @param string $class Class that will be performing
+     * @param array  $args  List of arguments to pass to the `$class->perform()` method
+     *
+     * @throws InvalidArgumentException If `$class` does not implement the required methods
+     */
     static function create($class, array $args = array());
 
+    /**
+     * @return Boolean
+     */
     function isSuccessful();
 
+    /**
+     * @return Boolean
+     */
     function isError();
 
+    /**
+     * @return Boolean
+     */
     function isActive();
 
+    /**
+     * Does the work
+     */
     function perform();
 
+    /**
+     * Returns an instance of the `class`
+     *
+     * @return mixed
+     */
+    function getInstance();
+
+    /**
+     * Returns the class name associated with this Job
+     *
+     * @return string
+     */
     function getClass();
 
+    /**
+     * Returns a list of arguments to pass to the `perform` method
+     *
+     * @return array
+     */
     function getArguments();
 }
diff --git a/src/Presque/Worker.php b/src/Presque/Worker.php
@@ -12,12 +12,15 @@
 namespace Presque;
 
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Presque\Event\JobEvent;
+use Presque\Event\WorkerEvent;
 
 class Worker implements WorkerInterface
 {
     private $id;
     private $queues;
     private $status;
+    private $eventDispatcher;
 
     public function __construct($id = null)
     {
@@ -111,6 +114,14 @@ public function isDying()
      */
     public function start()
     {
+        if ($this->hasEventDispatcher()) {
+            $event = $this->eventDispatcher(Events::WORK_STARTED, new WorkerEvent($this));
+
+            if ($event->isCanceled()) {
+                return;
+            }
+        }
+
         $this->setStatus(StatusInterface::RUNNING);
 
         $this->run();
@@ -145,6 +156,16 @@ protected function process(QueueInterface $queue)
             return;
         }
 
+        if ($this->hasEventDispatcher()) {
+            $event = $this->eventDispatcher->dispatch(Events::JOB_STARTED, new JobEvent($job, $queue, $this));
+
+            if ($event->isCanceled()) {
+                return;
+            }
+
+            $job = $event->getJob();
+        }
+
         $job->perform();
 
         if ($job->isSuccessful()) {

diff --git a/tests/Presque/Tests/JobTest.php b/tests/Presque/Tests/JobTest.php
@@ -39,6 +39,30 @@ public function testCreatingASimpleJobWithTooFewArguments()
         Job::create('Presque\Tests\Jobs\SimpleJob', array(1));
     }
 
+    /**
+     * @expectedException Presque\Exception\InvalidArgumentException
+     */
+    public function testCreatingAJobWithInvalidPerformMethod()
+    {
+        Job::create('Presque\Tests\Jobs\InvalidJob');
+    }
+
+    /**
+     * @expectedException Presque\Exception\InvalidArgumentException
+     */
+    public function testCreatingAJobWithPrivatePerformMethod()
+    {
+        Job::create('Presque\Tests\Jobs\PrivateJob');
+    }
+
+    /**
+     * @expectedException Presque\Exception\InvalidArgumentException
+     */
+    public function testCreatingAnIncompleteJob()
+    {
+        Job::create('Presque\Tests\Jobs\IncompleteJob');
+    }
+
     public function testCreatingASimpleJob()
     {
         $job = Job::create('Presque\Tests\Jobs\SimpleJob', array('simple', 'jobs'));