From 8314c99b56e66c2b86dd32432bcf5ed94a3ece02 Mon Sep 17 00:00:00 2001 From: Frederic Guillot Date: Mon, 23 May 2016 20:43:51 -0400 Subject: Added QueueManager to process background jobs --- app/Core/Queue/JobHandler.php | 50 +++++++++++++++++++++++++++++ app/Core/Queue/QueueManager.php | 71 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) create mode 100644 app/Core/Queue/JobHandler.php create mode 100644 app/Core/Queue/QueueManager.php (limited to 'app/Core/Queue') diff --git a/app/Core/Queue/JobHandler.php b/app/Core/Queue/JobHandler.php new file mode 100644 index 00000000..a2c4a2c7 --- /dev/null +++ b/app/Core/Queue/JobHandler.php @@ -0,0 +1,50 @@ + 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']); + } +} diff --git a/app/Core/Queue/QueueManager.php b/app/Core/Queue/QueueManager.php new file mode 100644 index 00000000..f34cb220 --- /dev/null +++ b/app/Core/Queue/QueueManager.php @@ -0,0 +1,71 @@ +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); + } + } +} -- cgit v1.2.3