more code

bupy7 · Sep 7, 2017 · b671829 · b671829
1 parent 8d457bf
commit b671829
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 16 deletions.
diff --git a/src/Entity/TaskInterface.php b/src/Entity/TaskInterface.php
@@ -11,9 +11,10 @@
 interface TaskInterface
 {
     public const STATUS_WAIT = 10;
-    public const STATUS_ERROR = 20;
-    public const STATUS_OK = 30;
-    public const STATUS_IMPOSSIBLE = 40;
+    public const STATUS_IN_PROCESSING = 20;
+    public const STATUS_ERROR = 30;
+    public const STATUS_OK = 40;
+    public const STATUS_IMPOSSIBLE = 50;
 
     public function setId(int $id): TaskInterface;
 

diff --git a/src/Service/QueueService.php b/src/Service/QueueService.php
@@ -11,9 +11,15 @@
 use Bupy7\Queue\Task\TaskInterface;
 use Bupy7\Queue\Options\ModuleOptions;
 use Exception;
+use Zend\EventManager\EventManagerAwareInterface;
+use Zend\EventManager\EventManagerAwareTrait;
 
-class QueueService
+class QueueService implements EventManagerAwareInterface
 {
+    public const EVENT_ERROR_EXECUTE = 'errorExecute';
+
+    use EventManagerAwareTrait;
+
     /**
      * @var TaskRepositoryInterface
      */
@@ -46,17 +52,13 @@ public function __construct(
     public function run(): void
     {
         $entities = $this->taskRepository->findForRun($this->config->getOneTimeLimit() ?: null);
-        try {
-            foreach ($entities as $entity) {
-                if (in_array($entity->getStatusId(), [
-                    TaskEntityInterface::STATUS_WAIT,
-                    TaskEntityInterface::STATUS_ERROR,
-                ])) {
-                    $this->executeTask($entity);
-                }
+        foreach ($entities as $entity) {
+            if (in_array($entity->getStatusId(), [
+                TaskEntityInterface::STATUS_WAIT,
+                TaskEntityInterface::STATUS_ERROR,
+            ])) {
+                $this->executeTask($entity);
             }
-        } finally {
-            $this->entityManager->flush();
         }
     }
 
@@ -65,9 +67,13 @@ protected function executeTask(TaskEntityInterface $entity): void
         if (!$this->queueManager->has($entity->getName())) {
             throw new UnknownTaskException(sprintf('"%s task is unknown."', $entity->getName()));
         }
+
         /** @var TaskInterface $task */
         $task = $this->queueManager->get($entity->getName());
         $entity->setRunAt(new DateTime);
+        $entity->setStatusId(TaskEntityInterface::STATUS_IN_PROCESSING);
+        $this->entityManager->flush($entity);
+
         try {
             if ($task->execute($entity->getParams())) {
                 $entity->setStatusId(TaskEntityInterface::STATUS_OK);
@@ -76,8 +82,10 @@ protected function executeTask(TaskEntityInterface $entity): void
             }
         } catch (Exception $e) {
             $entity->setStatusId(TaskEntityInterface::STATUS_ERROR);
-
-            // @TODO: Logging
+            // event trigger
+            $this->getEventManager()->trigger(self::EVENT_ERROR_EXECUTE, $this, [
+                'exception' => $e,
+            ]);
         } finally {
             if ($entity->getStatusId() === TaskEntityInterface::STATUS_ERROR) {
                 $entity->incNumberErrors();
@@ -90,5 +98,7 @@ protected function executeTask(TaskEntityInterface $entity): void
             }
             $entity->setStopAt(new DateTime);
         }
+
+        $this->entityManager->flush($entity);
     }
 }
diff --git a/test/Assert/Entity/Task.php b/test/Assert/Entity/Task.php
@@ -78,6 +78,7 @@ public function setStatusId(int $statusId): TaskInterface
     {
         if (!in_array($statusId, [
             self::STATUS_WAIT,
+            self::STATUS_IN_PROCESSING,
             self::STATUS_ERROR,
             self::STATUS_OK,
             self::STATUS_IMPOSSIBLE,