diff --git a/src/CronJobDiscovery.php b/src/CronJobDiscovery.php index 3e85453..c656c43 100644 --- a/src/CronJobDiscovery.php +++ b/src/CronJobDiscovery.php @@ -68,7 +68,7 @@ class CronJobDiscovery { 'module' => $definition['provider'], // Process queue jobs later by default. 'weight' => 10, - 'callback' => 'ultimate_cron_queue_callback', + 'callback' => 'ultimate_cron.queue_manager.createController', 'scheduler' => [ 'id' => 'simple', 'configuration' => [ diff --git a/src/Entity/CronJob.php b/src/Entity/CronJob.php index 35f113c..fcd8aba 100644 --- a/src/Entity/CronJob.php +++ b/src/Entity/CronJob.php @@ -295,7 +295,41 @@ class CronJob extends ConfigEntityBase implements CronJobInterface { */ protected function invokeCallback() { $callback = $this->getCallback(); - return $callback($this); + return call_user_func($callback, $this); + } + + /** + * Returns a callable for the given controller. + * + * @param string $callback + * A Controller string. + * + * @return mixed + * A PHP callable. + * + * @throws \LogicException + * If the controller cannot be parsed. + * + * @throws \InvalidArgumentException + * If the controller class does not exist. + */ + protected function resolveCallback($callback) { + // Controller in the service:method notation. + $count = substr_count($callback, ':'); + if ($count == 1) { + list($class_or_service, $method) = explode(':', $callback, 2); + } + // Controller in the class::method notation. + elseif (strpos($callback, '::') !== FALSE) { + list($class_or_service, $method) = explode('::', $callback, 2); + } + else { + throw new \LogicException(sprintf('Unable to parse the callback name "%s".', $callback)); + } + + $callback = $this->classResolver->getInstanceFromDefinition($class_or_service); + + return array($callback, $method); } /** @@ -710,7 +744,12 @@ class CronJob extends ConfigEntityBase implements CronJobInterface { * {@inheritdoc} */ public function getCallback() { - return $this->callback; + if(is_callable($this->callback)) { + return $this->callback; + } + else { + return $this->resolveCallback($this->callback); + } } /** diff --git a/src/QueueManager.php b/src/QueueManager.php new file mode 100644 index 0000000..c1233c7 --- /dev/null +++ b/src/QueueManager.php @@ -0,0 +1,131 @@ +plugin_manager_queue_worker = $plugin_manager_queue_worker; + $this->queue = $queue; + $this->config_factory = $config_factory; + } + + public function queueCallback(CronJobInterface $job) { + $queue_name = str_replace(CronJobInterface::QUEUE_ID_PREFIX, '', $job->id()); + + $queue_manager = $this->plugin_manager_queue_worker; + $queue_factory = $this->queue; + + $config = $this->config_factory->get('ultimate_cron.settings'); + + $info = $queue_manager->getDefinition($queue_name); + + // Make sure every queue exists. There is no harm in trying to recreate + // an existing queue. + $queue_factory->get($queue_name)->createQueue(); + + /** @var \Drupal\Core\Queue\QueueWorkerInterface $queue_worker */ + $queue_worker = $queue_manager->createInstance($queue_name); + $end = microtime(TRUE) + (isset($info['cron']['time']) ? $info['cron']['time'] : $config->get('queue.timeouts.time')); + + /** @var \Drupal\Core\Queue\QueueInterface $queue */ + $queue = $queue_factory->get($queue_name); + $items = 0; + while (microtime(TRUE) < $end) { + // Check kill signal. + if ($job->getSignal('kill')) { + \Drupal::logger('ultimate_cron')->warning('Kill signal received for job @job_id', ['@job_id' => $job->id()]); + break; + } + + $item = $queue->claimItem($config->get('queue.timeouts.lease_time')); + + // If there is no item, check the empty delay setting and wait if + // configured. + if (!$item) { + if ($config->get('queue.delays.empty_delay')) { + usleep($config->get('queue.delays.empty_delay') * 1000000); + continue; + } + else { + break; + } + } + + try { + // We have an item, check if we need to wait. + if ($config->get('queue.delays.item_delay')) { + if ($items == 0) { + // Move the boundary if using a throttle, + // to avoid waiting for nothing. + $end -= $config->get('queue.delays.item_delay'); + } + else { + // Sleep before retrieving. + usleep($config->get('queue.delays.item_delay') * 1000000); + } + } + + $queue_worker->processItem($item->data); + $queue->deleteItem($item); + $items++; + } + catch (RequeueException $e) { + // The worker requested the task be immediately requeued. + $queue->releaseItem($item); + } + catch (SuspendQueueException $e) { + // If the worker indicates there is a problem with the whole queue, + // release the item and skip to the next queue. + $queue->releaseItem($item); + + watchdog_exception('cron', $e); + + } + catch (\Exception $e) { + // In case of any other kind of exception, log it and leave the item + // in the queue to be processed again later. + watchdog_exception('ultimate_cron_queue', $e); + } + } + } +} diff --git a/src/QueueManagerInterface.php b/src/QueueManagerInterface.php new file mode 100644 index 0000000..29cdf46 --- /dev/null +++ b/src/QueueManagerInterface.php @@ -0,0 +1,19 @@ +id()); - - $queue_manager = \Drupal::service('plugin.manager.queue_worker'); - $queue_factory = \Drupal::service('queue'); - - $config = \Drupal::config('ultimate_cron.settings'); - - $info = $queue_manager->getDefinition($queue_name); - - // Make sure every queue exists. There is no harm in trying to recreate - // an existing queue. - $queue_factory->get($queue_name)->createQueue(); - - /** @var \Drupal\Core\Queue\QueueWorkerInterface $queue_worker */ - $queue_worker = $queue_manager->createInstance($queue_name); - $end = microtime(TRUE) + (isset($info['cron']['time']) ? $info['cron']['time'] : $config->get('queue.timeouts.time')); - - /** @var \Drupal\Core\Queue\QueueInterface $queue */ - $queue = $queue_factory->get($queue_name); - $items = 0; - while (microtime(TRUE) < $end) { - // Check kill signal. - if ($job->getSignal('kill')) { - \Drupal::logger('ultimate_cron')->warning('Kill signal received for job @job_id', ['@job_id' => $job->id()]); - break; - } - - $item = $queue->claimItem($config->get('queue.timeouts.lease_time')); - - // If there is no item, check the empty delay setting and wait if - // configured. - if (!$item) { - if ($config->get('queue.delays.empty_delay')) { - usleep($config->get('queue.delays.empty_delay') * 1000000); - continue; - } - else { - break; - } - } - - try { - // We have an item, check if we need to wait. - if ($config->get('queue.delays.item_delay')) { - if ($items == 0) { - // Move the boundary if using a throttle, - // to avoid waiting for nothing. - $end -= $config->get('queue.delays.item_delay'); - } - else { - // Sleep before retrieving. - usleep($config->get('queue.delays.item_delay') * 1000000); - } - } - - $queue_worker->processItem($item->data); - $queue->deleteItem($item); - $items++; - } - catch (RequeueException $e) { - // The worker requested the task be immediately requeued. - $queue->releaseItem($item); - } - catch (SuspendQueueException $e) { - // If the worker indicates there is a problem with the whole queue, - // release the item and skip to the next queue. - $queue->releaseItem($item); - - watchdog_exception('cron', $e); - - } - catch (\Exception $e) { - // In case of any other kind of exception, log it and leave the item - // in the queue to be processed again later. - watchdog_exception('ultimate_cron_queue', $e); - } - } -} - -/** * * Implements hook_modules_installed(). */ function ultimate_cron_modules_installed($modules) { diff --git a/ultimate_cron.services.yml b/ultimate_cron.services.yml index 61d369b..b93316b 100644 --- a/ultimate_cron.services.yml +++ b/ultimate_cron.services.yml @@ -31,3 +31,8 @@ services: arguments: ['@logger.log_message_parser'] tags: - { name: logger } + + ultimate_cron.queue_manager: + class: Drupal\ultimate_cron\QueueManager + arguments: ["@plugin.manager.queue_worker", "@queue", "@config.factory"] +