job queue

application/Espo/Core/CronManager.php

@@ -189,9 +189,25 @@ class CronManager
         $this->getCronJobUtil()->updateFailedJobAttempts();
         $this->createJobsFromScheduledJobs();
         $this->getCronJobUtil()->removePendingJobDuplicates();
-        $pendingJobList = $this->getCronJobUtil()->getPendingJobList();
 
-        if ($this->useProcessPool()) {
+        $this->processPendingJobs();
+    }
+
+    public function processPendingJobs($queue = null, $limit = null, $poolDisabled = false, $noLock = false)
+    {
+        if (is_null($limit)) {
+            $limit = intval($this->getConfig()->get('jobMaxPortion', 0));
+        }
+
+        $pendingJobList = $this->getCronJobUtil()->getPendingJobList($queue, $limit);
+
+        $useProcessPool = $this->useProcessPool();
+
+        if ($poolDisabled) {
+            $useProcessPool = false;
+        }
+
+        if ($useProcessPool) {
             $pool = \Spatie\Async\Pool::create()
                 ->autoload(getcwd() . '/vendor/autoload.php')
                 ->concurrency($this->getConfig()->get('jobPoolConcurrencyNumber'))
@@ -200,8 +216,8 @@ class CronManager
 
         foreach ($pendingJobList as $job) {
             $skip = false;
-            $this->getEntityManager()->getPdo()->query('LOCK TABLES `job` WRITE');
-            if ($this->getCronJobUtil()->isJobPending($job->id)) {
+            if (!$noLock) $this->lockJobTable();
+            if ($noLock || $this->getCronJobUtil()->isJobPending($job->id)) {
                 if ($job->get('scheduledJobId')) {
                     if ($this->getCronJobUtil()->isScheduledJobRunning($job->get('scheduledJobId'), $job->get('targetId'), $job->get('targetType'))) {
                         $skip = true;
@@ -212,10 +228,10 @@ class CronManager
             }
 
             if ($skip) {
-                $this->getEntityManager()->getPdo()->query('UNLOCK TABLES');
+                if (!$noLock) $this->unlockTables();
                 continue;
             }
-            if ($this->useProcessPool()) {
+            if ($useProcessPool) {
                 $job->set('status', self::READY);
             } else {
                 $job->set('status', self::RUNNING);
@@ -223,9 +239,9 @@ class CronManager
             }
 
             $this->getEntityManager()->saveEntity($job);
-            $this->getEntityManager()->getPdo()->query('UNLOCK TABLES');
+            if (!$noLock) $this->unlockTables();
 
-            if ($this->useProcessPool()) {
+            if ($useProcessPool) {
                 $task = new \Espo\Core\Utils\Cron\JobTask($job->id);
                 $pool->add($task);
             } else {
@@ -233,11 +249,21 @@ class CronManager
             }
         }
 
-        if ($this->useProcessPool()) {
+        if ($useProcessPool) {
             $pool->wait();
         }
     }
 
+    protected function lockJobTable()
+    {
+        $this->getEntityManager()->getPdo()->query('LOCK TABLES `job` WRITE');
+    }
+
+    protected function unlockTables()
+    {
+        $this->getEntityManager()->getPdo()->query('UNLOCK TABLES');
+    }
+
     public function runJobById($id)
     {
         if (empty($id)) throw new Error();
@@ -281,6 +307,11 @@ class CronManager
         $status = $isSuccess ? self::SUCCESS : self::FAILED;
 
         $job->set('status', $status);
+
+        if ($isSuccess) {
+            $job->set('executedAt', date('Y-m-d H:i:s'));
+        }
+
         $this->getEntityManager()->saveEntity($job);
 
         if ($job->get('scheduledJobId') && !$skipLog) {

application/Espo/Core/Mail/Importer.php

@@ -520,7 +520,8 @@ class Importer
                     'targetType' => 'Email',
                     'targetId' => $duplicate->id
                 ],
-                'executeAt' => $executeAt
+                'executeAt' => $executeAt,
+                'queue' => 'q1'
             ]);
             $this->getEntityManager()->saveEntity($job);
         }

application/Espo/Core/Utils/Cron/Job.php

@@ -74,10 +74,8 @@ class Job
         ])->findOne();
     }
 
-    public function getPendingJobList()
+    public function getPendingJobList($queue = null, $limit = 0)
     {
-        $limit = intval($this->getConfig()->get('jobMaxPortion', 0));
-
         $selectParams = [
             'select' => [
                 'id',
@@ -89,11 +87,13 @@ class Job
                 'methodName',
                 'method', // TODO remove deprecated
                 'serviceName',
-                'data'
+                'data',
+                'queue'
             ],
             'whereClause' => [
                 'status' => CronManager::PENDING,
-                'executeTime<=' => date('Y-m-d H:i:s')
+                'executeTime<=' => date('Y-m-d H:i:s'),
+                'queue' => $queue
             ],
             'orderBy' => 'executeTime'
         ];

application/Espo/Hooks/Common/AssignmentEmailNotification.php

@@ -87,17 +87,18 @@ class AssignmentEmailNotification extends \Espo\Core\Hooks\Base
     protected function createJob(Entity $entity, $userId)
     {
         $job = $this->getEntityManager()->getEntity('Job');
-        $job->set(array(
+        $job->set([
             'serviceName' => 'EmailNotification',
             'methodName' => 'notifyAboutAssignmentJob',
-            'data' => json_encode(array(
+            'data' => [
                 'userId' => $userId,
                 'assignerUserId' => $this->getUser()->id,
                 'entityId' => $entity->id,
                 'entityType' => $entity->getEntityType()
-            )),
+            ],
             'executeTime' => date('Y-m-d H:i:s'),
-        ));
+            'queue' => 'q1'
+        ]);
         $this->getEntityManager()->saveEntity($job);
     }
 }

application/Espo/Hooks/Common/Stream.php

@@ -258,15 +258,16 @@ class Stream extends \Espo\Core\Hooks\Base
 
                 if (!empty($autofollowUserIdList)) {
                     $job = $this->getEntityManager()->getEntity('Job');
-                    $job->set(array(
+                    $job->set([
                         'serviceName' => 'Stream',
                         'methodName' => 'afterRecordCreatedJob',
-                        'data' => array(
+                        'data' => [
                             'userIdList' => $autofollowUserIdList,
                             'entityType' => $entity->getEntityType(),
                             'entityId' => $entity->id
-                        )
-                    ));
+                        ],
+                        'queue' => 'q1'
+                    ]);
                     $this->getEntityManager()->saveEntity($job);
                 }
             } else {
@@ -337,14 +338,15 @@ class Stream extends \Espo\Core\Hooks\Base
                     )
                 ) {
                     $job = $this->getEntityManager()->getEntity('Job');
-                    $job->set(array(
+                    $job->set([
                         'serviceName' => 'Stream',
                         'methodName' => 'controlFollowersJob',
-                        'data' => array(
+                        'data' => [
                             'entityType' => $entity->getEntityType(),
                             'entityId' => $entity->id
-                        )
-                    ));
+                        ],
+                        'queue' => 'q1'
+                    ]);
                     $this->getEntityManager()->saveEntity($job);
                 }
             }

application/Espo/Jobs/ProcessJobQueueQ1.php

@@ -0,0 +1,44 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM - Open Source CRM application.
+ * Copyright (C) 2014-2018 Yuri Kuznetsov, Taras Machyshyn, Oleksiy Avramenko
+ * Website: http://www.espocrm.com
+ *
+ * EspoCRM is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * EspoCRM is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with EspoCRM. If not, see http://www.gnu.org/licenses/.
+ *
+ * The interactive user interfaces in modified source and object code versions
+ * of this program must display Appropriate Legal Notices, as required under
+ * Section 5 of the GNU General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Jobs;
+
+use \Espo\Core\Exceptions;
+
+class ProcessJobQueueQ1 extends \Espo\Core\Jobs\Base
+{
+    public function run()
+    {
+        $limit = $this->getConfig()->get('jobQ1MaxPortion', 500);
+
+        $cronManager = new \Espo\Core\CronManager($this->getContainer());
+
+        $cronManager->processPendingJobs('q1', $limit, true, true);
+    }
+}

application/Espo/Resources/i18n/en_US/Job.json

@@ -2,6 +2,7 @@
     "fields": {
         "status": "Status",
         "executeTime": "Execute At",
+        "executedAt": "Executed At",
         "attempts": "Attempts Left",
         "failedAttempts": "Failed Attempts",
         "serviceName": "Service",
@@ -11,7 +12,8 @@
         "scheduledJobJob": "Scheduled Job Name",
         "data": "Data",
         "targetType": "Target Type",
-        "targetId": "Target ID"
+        "targetId": "Target ID",
+        "queue": "Queue"
     },
     "options": {
         "status": {

application/Espo/Resources/layouts/Job/detailSmall.json

@@ -3,9 +3,15 @@
         "label":"",
         "rows":[
             [{"name":"name"}, {"name": "status"}],
-            [{"name":"executeTime"}, {"name": "attempts"}],
-            [{"name":"createdAt"}, {"name": "failedAttempts"}],
-            [{"name":"modifiedAt"}, false]
+            [{"name":"queue"}, false]
+        ]
+    },
+    {
+        "label":"",
+        "rows":[
+            [{"name":"executeTime"}, {"name": "createdAt"}],
+            [{"name":"executedAt"}, {"name": "modifiedAt"}],
+            [{"name":"attempts"}, {"name": "failedAttempts"}]
         ]
     },
     {

application/Espo/Resources/layouts/Job/filters.json

@@ -1,5 +1,7 @@
 [
+    "status",
     "createdAt",
     "executeTime",
-    "status"
+    "executedAt",
+    "queue"
 ]

application/Espo/Resources/layouts/Job/list.json

@@ -1,7 +1,7 @@
 [
-    {"name":"name", "width": 30},
-    {"name":"status"},
-    {"name":"executeTime"},
-    {"name":"attempts", "width": 10},
-    {"name":"createdAt", "align": "right", "width": 15}
+    {"name":"name"},
+    {"name":"status", "width": 14},
+    {"name":"executeTime", "width": 14},
+    {"name":"queue", "width": 10},
+    {"name":"createdAt", "align": "right", "width": 14}
 ]

application/Espo/Resources/metadata/entityDefs/Job.json

@@ -8,7 +8,13 @@
         "status": {
             "type": "enum",
             "options": ["Pending", "Ready", "Running", "Success", "Failed"],
-            "default": "Pending"
+            "default": "Pending",
+            "style": {
+                "Success": "success",
+                "Failed": "danger",
+                "Running": "warning",
+                "Ready": "warning"
+            }
         },
         "executeTime": {
             "type": "datetime",
@@ -39,6 +45,14 @@
             "link": "scheduledJob",
             "field": "job"
         },
+        "queue": {
+            "type": "varchar",
+            "maxLength": 36,
+            "default": null
+        },
+        "executedAt": {
+            "type": "datetime"
+        },
         "pid": {
             "type": "int"
         },

application/Espo/Resources/metadata/entityDefs/ScheduledJob.json

@@ -89,6 +89,11 @@
             "name": "Check for New Versions of Installed Extensions",
             "isSystem": true,
             "scheduling": "25 5 * * *"
+        },
+        "ProcessJobQueueQ1": {
+            "name": "Process Job Queue Q1",
+            "isSystem": true,
+            "scheduling": "*/1 * * * *"
         }
     }
 }

application/Espo/Services/Pdf.php

@@ -227,7 +227,8 @@ class Pdf extends \Espo\Core\Services\Base
             'data' => [
                 'id' => $attachment->id
             ],
-            'executeTime' => (new \DateTime())->modify('+' . $this->removeMassFilePeriod)->format('Y-m-d H:i:s')
+            'executeTime' => (new \DateTime())->modify('+' . $this->removeMassFilePeriod)->format('Y-m-d H:i:s'),
+            'queue' => 'q1'
         ]);
         $this->getEntityManager()->saveEntity($job);
 

application/Espo/Services/User.php

@@ -225,14 +225,15 @@ class User extends Record
 
         $job = $this->getEntityManager()->getEntity('Job');
 
-        $job->set(array(
+        $job->set([
             'serviceName' => 'User',
             'methodName' => 'removeChangePasswordRequestJob',
             'data' => [
                 'id' => $passwordChangeRequest->id
             ],
-            'executeTime' => $dt->format('Y-m-d H:i:s')
-        ));
+            'executeTime' => $dt->format('Y-m-d H:i:s'),
+            'queue' => 'q1'
+        ]);
 
         $this->getEntityManager()->saveEntity($job);