Added QueueManager to process background jobs
This commit is contained in:
@@ -27,6 +27,7 @@ use Pimple\Container;
|
||||
* @property \Kanboard\Core\Http\Response $response
|
||||
* @property \Kanboard\Core\Http\Router $router
|
||||
* @property \Kanboard\Core\Http\Route $route
|
||||
* @property \Kanboard\Core\Queue\QueueManager $queueManager
|
||||
* @property \Kanboard\Core\Mail\Client $emailClient
|
||||
* @property \Kanboard\Core\ObjectStorage\ObjectStorageInterface $objectStorage
|
||||
* @property \Kanboard\Core\Plugin\Hook $hook
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
namespace Kanboard\Core\Mail;
|
||||
|
||||
use Kanboard\Job\EmailJob;
|
||||
use Pimple\Container;
|
||||
use Kanboard\Core\Base;
|
||||
|
||||
@@ -46,25 +47,31 @@ class Client extends Base
|
||||
public function send($email, $name, $subject, $html)
|
||||
{
|
||||
if (! empty($email)) {
|
||||
$this->logger->debug('Sending email to '.$email.' ('.MAIL_TRANSPORT.')');
|
||||
|
||||
$start_time = microtime(true);
|
||||
$author = 'Kanboard';
|
||||
|
||||
if ($this->userSession->isLogged()) {
|
||||
$author = e('%s via Kanboard', $this->helper->user->getFullname());
|
||||
}
|
||||
|
||||
$this->getTransport(MAIL_TRANSPORT)->sendEmail($email, $name, $subject, $html, $author);
|
||||
|
||||
if (DEBUG) {
|
||||
$this->logger->debug('Email sent in '.round(microtime(true) - $start_time, 6).' seconds');
|
||||
}
|
||||
$this->queueManager->push(EmailJob::getInstance($this->container)
|
||||
->withParams($email, $name, $subject, $html, $this->getAuthor())
|
||||
);
|
||||
}
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get email author
|
||||
*
|
||||
* @access public
|
||||
* @return string
|
||||
*/
|
||||
public function getAuthor()
|
||||
{
|
||||
$author = 'Kanboard';
|
||||
|
||||
if ($this->userSession->isLogged()) {
|
||||
$author = e('%s via Kanboard', $this->helper->user->getFullname());
|
||||
}
|
||||
|
||||
return $author;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get mail transport instance
|
||||
*
|
||||
|
||||
50
app/Core/Queue/JobHandler.php
Normal file
50
app/Core/Queue/JobHandler.php
Normal file
@@ -0,0 +1,50 @@
|
||||
<?php
|
||||
|
||||
namespace Kanboard\Core\Queue;
|
||||
|
||||
use Kanboard\Core\Base;
|
||||
use Kanboard\Job\BaseJob;
|
||||
use SimpleQueue\Job;
|
||||
|
||||
/**
|
||||
* Class JobHandler
|
||||
*
|
||||
* @package Kanboard\Core\Queue
|
||||
* @author Frederic Guillot
|
||||
*/
|
||||
class JobHandler extends Base
|
||||
{
|
||||
/**
|
||||
* Serialize a job
|
||||
*
|
||||
* @access public
|
||||
* @param BaseJob $job
|
||||
* @return Job
|
||||
*/
|
||||
public function serializeJob(BaseJob $job)
|
||||
{
|
||||
return new Job(array(
|
||||
'class' => get_class($job),
|
||||
'params' => $job->getJobParams(),
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a job
|
||||
*
|
||||
* @access public
|
||||
* @param Job $job
|
||||
*/
|
||||
public function executeJob(Job $job)
|
||||
{
|
||||
$payload = $job->getBody();
|
||||
$className = $payload['class'];
|
||||
|
||||
if (DEBUG) {
|
||||
$this->logger->debug(__METHOD__.' Received job => '.$className);
|
||||
}
|
||||
|
||||
$worker = new $className($this->container);
|
||||
call_user_func_array(array($worker, 'execute'), $payload['params']);
|
||||
}
|
||||
}
|
||||
71
app/Core/Queue/QueueManager.php
Normal file
71
app/Core/Queue/QueueManager.php
Normal file
@@ -0,0 +1,71 @@
|
||||
<?php
|
||||
|
||||
namespace Kanboard\Core\Queue;
|
||||
|
||||
use Kanboard\Core\Base;
|
||||
use Kanboard\Job\BaseJob;
|
||||
use LogicException;
|
||||
use SimpleQueue\Queue;
|
||||
|
||||
/**
|
||||
* Class QueueManager
|
||||
*
|
||||
* @package Kanboard\Core\Queue
|
||||
* @author Frederic Guillot
|
||||
*/
|
||||
class QueueManager extends Base
|
||||
{
|
||||
/**
|
||||
* @var Queue
|
||||
*/
|
||||
protected $queue = null;
|
||||
|
||||
/**
|
||||
* Set queue driver
|
||||
*
|
||||
* @access public
|
||||
* @param Queue $queue
|
||||
* @return $this
|
||||
*/
|
||||
public function setQueue(Queue $queue)
|
||||
{
|
||||
$this->queue = $queue;
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a new job to the queue
|
||||
*
|
||||
* @access public
|
||||
* @param BaseJob $job
|
||||
* @return $this
|
||||
*/
|
||||
public function push(BaseJob $job)
|
||||
{
|
||||
if ($this->queue !== null) {
|
||||
$this->queue->push(JobHandler::getInstance($this->container)->serializeJob($job));
|
||||
} else {
|
||||
call_user_func_array(array($job, 'execute'), $job->getJobParams());
|
||||
}
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait for new jobs
|
||||
*
|
||||
* @access public
|
||||
* @throws LogicException
|
||||
*/
|
||||
public function listen()
|
||||
{
|
||||
if ($this->queue === null) {
|
||||
throw new LogicException('No Queue Driver defined!');
|
||||
}
|
||||
|
||||
while ($job = $this->queue->pull()) {
|
||||
JobHandler::getInstance($this->container)->executeJob($job);
|
||||
$this->queue->completed($job);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user