<?php
// $Id: job_queue.module,v 1.1.2.7 2008-03-11 19:04:20 drumm Exp $

define('JOB_QUEUE_DO_NOT_RUN', 127); // tinyint max

function job_queue_menu($may_cache) {
  $items = array();

  if ($may_cache) {
    $items[] = array(
      'path' => 'admin/logs/job_queue',
      'title' => t('Queued jobs'),
      'description' => t('View a list of the queued jobs.'),
      'callback' => 'job_queue_list',
      'access' => user_access('view queued jobs'),
    );
    $items[] = array(
      'path' => 'admin/settings/job_queue',
      'title' => t('Job queue'),
      'description' => t('Set queue priorities.'),
      'callback' => 'drupal_get_form',
      'callback arguments' => array('job_queue_settings_form'),
      'access' => user_access('administer job queue'),
    );
  }

  return $items;
}

function job_queue_perm() {
  return array('view queued jobs', 'administer job queue');
}

function job_queue_requirements($phase) {
  $requirements = array();

  if ($phase == 'runtime') {
    $count = db_result(db_query('SELECT count(*) FROM {job_queue}'));
    $requirements['job_queue'] = array(
      'title' => t('Job queue'),
      'severity' => ($count > 0) ? REQUIREMENT_WARNING : REQUIREMENT_OK,
      'value' => ($count > 0) ? format_plural($count, 'There is <strong>@count</strong> queued job.', 'There are <strong>@count</strong> queued jobs.') : t('There are no queued jobs.'),
    );
  }

  return $requirements;
}

/**
 * Add a job to the queue. The function added will be called in the order it
 * was added during cron.
 *
 * @param $function
 *   The function name to call.
 * @param $description
 *   A human-readable description of the queued job.
 * @param $arguments
 *   Optional array of arguments to pass to the function.
 * @param $file
 *   Optional file path which needs to be included for $fucntion.
 * @param $no_duplicate
 *   If TRUE, do not add the job to the queue if one with the same function and
 *   arguments already exists.
 */
function job_queue_add($function, $description, $arguments = array(), $file = '', $no_duplicate = FALSE) {
  static $priorities;

  if (!isset($priorities[$function])) {
    $priorities[$function] = db_result(db_query("SELECT priority FROM {job_queue_functions} WHERE function = '%s'", $function));
  }

  if ($no_duplicate) {
    $result = db_result(db_query("SELECT jqid FROM {job_queue} WHERE function = '%s' AND arguments = '%s'", $function, serialize($arguments)));
    if (!empty($result)) {
      return;
    }
  }

  db_query("INSERT INTO {job_queue} (jqid, created, description, function, arguments, file, priority) VALUES (%d, %d, '%s', '%s', '%s', '%s', %d)", db_next_id('{job_queue}_jqid'), time(), $description, $function, serialize($arguments), $file, $priorities[$function]);
}

/**
 * Dequeue and execute a queued job. Actions are logged via watchdog().
 *
 * @return TRUE if a job was dequeued, reguardless of success or failure
 *   executing. FALSE if the queue was empty.
 */
function job_queue_dequeue() {
  $result = db_query_range('SELECT jqid, description, function, arguments, file FROM {job_queue} WHERE priority <> %d ORDER BY priority, jqid', JOB_QUEUE_DO_NOT_RUN, 0, 1);
  if (db_num_rows($result) > 0) {
    $job = db_fetch_object($result);
    if (!empty($job->file)) {
      include_once $job->file;
    }
    if (function_exists($job->function)) {
      call_user_func_array($job->function, unserialize($job->arguments));
      watchdog('job_queue', t('Ran queued job "!description"', array('!description' => $job->description)));
    }
    else {
      watchdog('job_queue', t('Failed to run queued job "!description" because the function %function is not defined.', array('!description' => $job->description, '%function' => $job->function)), WATCHDOG_ERROR);
    }
    db_query('DELETE FROM {job_queue} WHERE jqid = %d', $job->jqid);
    return TRUE;
  }
  return FALSE;
}

function job_queue_cron() {
  $job_count = db_result(db_query('SELECT count(*) FROM {job_queue}'));
  while ($job_count > 0 && job_queue_dequeue()) {
    if ((timer_read('page') / 1000) > (ini_get('max_execution_time') / 2)) {
      break; // Stop once we have used over half of the maximum execution time or exceeds the original number of jobs.
    }
    $job_count -= 1;
  }
}

function job_queue_list() {
  $count = db_result(db_query('SELECT count(*) FROM {job_queue}'));

  if ($count == 0) {
    $output = '<p>'. t('There are <strong>no queued jobs</strong>. Modules may programmatically add queued jobs.') .'</p>';
  }
  else {
    $output = '<p>'. format_plural($count, 'There is <strong>@count</strong> queued job.', 'There are <strong>@count</strong> queued jobs.') .'</p>';

    $header = array(
      t('Created'),
      t('Description'),
    );
    $result = pager_query('SELECT created, description FROM {job_queue} ORDER BY priority, jqid', 20);
    $rows = array();
    while ($job = db_fetch_object($result)) {
      $rows[] = array(
        format_date($job->created),
        $job->description,
      );
    }
    $output .= theme('table', $header, $rows);
    $output .= theme('pager', array(), 20);
  }

  return $output;
}

/**
 * Job queue function prioritization.
 *
 * hook_job_queue_functions() should return
 *   array(
 *     'function_name' => array(
 *       'title' => t('Human-readable explanation'),
 *     ),
 *     ...
 *   );
 *
 * Functions which do not have entries in a implementation of this hook will
 * always have the default priority.
 */
function job_queue_settings_form() {
  $result = db_query('SELECT function, priority FROM {job_queue_functions}');
  $priorities = array();
  while ($function = db_fetch_object($result)) {
    $priorities[$function->function] = $function->priority;
  }

  $form = array();

  $form['priorities'] = array(
    '#theme' => 'job_queue_priorities_table',
    '#tree' => TRUE,
  );
  foreach (module_invoke_all('job_queue_functions') as $function => $metadata) {
    $form['priorities'][$function] = array();
    $form['priorities'][$function]['title'] = array(
      '#value' => $metadata['title'],
    );
    $form['priorities'][$function]['priority'] = array(
      '#type' => 'select',
      '#options' => (drupal_map_assoc(range(-10, 10)) + array(JOB_QUEUE_DO_NOT_RUN => t('Do not run'))),
      '#default_value' => array_key_exists($function, $priorities) ? $priorities[$function] : 0,
    );
  }
  $form['submit'] = array(
    '#type' => 'submit',
    '#value' => t('Save settings'),
  );

  return $form;
}

function theme_job_queue_priorities_table($element) {
  $headers = array(
    t('Job type'),
    t('Priority'),
  );

  $rows = array();
  foreach (element_children($element) as $function) {
    $rows[] = array(
      drupal_render($element[$function]['title']),
      drupal_render($element[$function]['priority']),
    );
  }

  return theme('table', $headers, $rows);
}

function job_queue_settings_form_submit($form_id, $form_values) {
  foreach ($form_values['priorities'] as $function => $data) {
    db_query("DELETE FROM {job_queue_functions} WHERE function = '%s'", $function);
    db_query("INSERT INTO {job_queue_functions} (function, priority) VALUES ('%s', %d)", $function, $data['priority']);
    db_query("UPDATE {job_queue} SET priority = %d WHERE function = '%s'", $data['priority'], $function);
  }
  drupal_set_message(t('Saved job queue settings.'));
}