modules/datasync/datasync.module

<?php
// $Id$

/*
 * TODO: There should be some way for administrators to see the statuses of all
 * the running/completed jobs
 * TODO: Statistics on jobs
 */
include_once(drupal_get_path('module','datasync') . '/datasync_constants.inc');


/**************************
 * DRUPAL HOOKS
 ***************************/
function datasync_menu($may_cache) {
  $items = array();
  if ($may_cache) {
    $items[] = array(
      'path' => 'datasync/queue',
      'title' => '',
      'access' => TRUE,
      'callback' => 'datasync_queue_callback',
      'type' => MENU_CALLBACK,
    );
    $items[] = array(
      'path' => 'datasync/jobs',
      'title' => '',
      'access' => TRUE,
      'callback' => 'datasync_jobs_callback',
      'type' => MENU_CALLBACK,
    );
    $items[] = array(
      'path' => 'admin/settings/datasync',
      'title' => t('DataSync'),
      'description' => t('Manage DataSync Settings'),
      'access' => user_access('administer site configuration'),
      'callback' => 'drupal_get_form',
      'callback arguments' => array('datasync_admin_settings'),
      'type' => MENU_NORMAL_ITEM,
    );
    $items[] = array(
    	'path' => 'admin/settings/datasync/general',
    	'title' => t('DataSync'),
      	'type' => MENU_DEFAULT_LOCAL_TASK,
      	'weight' => -10
    );
  }
  else {

  }
  return $items;
}

/**********************************
 * DATASYNC CORE FUNCTIONS
 * -All these functions except take_jobs should probably only be called from within transactions (D6)
 * -They all return FALSE on error
 * -All except take_jobs return TRUE on success
 * -take_jobs returns an array of job_ids on success and runs inside of its own transactions (D6)
 **********************************/

/**
 * Reserve $num_jobs jobs as running and return job IDs to be worked on.
 *
 * @param  $num_jobs
 *   Number of jobs to reserve and return.
 * @param  $valid_status
 *   Only get jobs with these statuses.
 * @param  $consumers
 *   Only get jobs of this consumer type.
 * @return
 *   An array of job IDs
 */
function datasync_take_jobs($num_jobs, $valid_status, $consumers) {
  $consumers = datasync_enabled_consumers($consumers);
  $status_in_string = implode(', ', $valid_status);
  $consumer_in_string = "'" . implode("', '", $consumers) . "'";
  //db_query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE");
  //db_query('BEGIN;');
  $result = db_query_range("SELECT job_id FROM {datasync_jobs} WHERE consumer IN ($consumer_in_string) AND status IN ($status_in_string) AND completed=%d AND started=%d ORDER BY job_id ASC", DATASYNC_INCOMPLETE, DATASYNC_UNSTARTED, 0, $num_jobs);
  $job_ids = array();
  while ($row = db_fetch_object($result)) {
    $job_ids[] = $row->job_id;
  }
  if (empty($job_ids)) {
    //db_query('ROLLBACK;');
    return array();
  }
  $in_string = implode(', ', $job_ids);
  db_query("UPDATE {datasync_jobs} SET started=%d WHERE job_id IN ($in_string)", time());

  if (db_affected_rows() == count($job_ids)) {
    //db_query('COMMIT;');
    return $job_ids;
  }
  //db_query('ROLLBACK;');
  return FALSE;
}

/**
 * Change a job's status to $current_status and set its next status to $next_status
 *
 * @param $job_id
 *   Job ID of the job that we are modifying
 * @param $current_status
 *   status that the job will have after this switch.
 * @param $next_status
 *   next_status the job will have after this switch. NULL means just increment status by one.
 * @param $started
 *   unix time value of when the job was started (usually DATASYNC_UNSTARTED).
 * @param $completed
 *   unix time value of when the job was completed (usually DATASYNC_INCOMPLETE)
 * @return
 *   TRUE on success, FALSE on failure
 */
function datasync_switch_status($job_id, $current_status, $next_status=NULL, $started=DATASYNC_UNSTARTED, $completed=DATASYNC_INCOMPLETE) {
  $c = datasync_enabled_consumers(array(datasync_consumer_by_job_id($job_id)));
  if (empty($c)) {
    return FALSE;
  }
  if ($next_status !== NULL) {
    db_query("UPDATE {datasync_jobs} SET started=%d, completed=%d, status=%d, next_status=%d WHERE job_id=%d", $started, $completed, $current_status, $next_status, $job_id);
  }
  else {
    db_query("UPDATE {datasync_jobs} SET started=%d, completed=%d, status=%d, next_status=NULL WHERE job_id=%d", $started, $completed, $current_status, $job_id);
  }
  return TRUE;
}

/**
 * Add a job of type $consumer to the job queue
 *
 * @param $consumer
 *   Type of job we are adding
 * @return
 *   Not FALSE on success, FALSE on failure
 */
function datasync_add_job($consumer) {
  $c = datasync_enabled_consumers(array($consumer));
  if (empty($c)) {
    return FALSE;
  }
  $job_id = db_next_id('{datasync_jobs}_job_id');
  db_query("INSERT INTO {datasync_jobs} (job_id, consumer, status, next_status, started, completed) VALUES (%d, '%s', %d, %d, %d, %d)",
        $job_id, $consumer, DATASYNC_ADDED, DATASYNC_INITIALIZE, time(), time());
  return db_error();
}

/**
 * Remove a job from the jobs queue
 *
 * @param $job_id
 *   Job ID of the job to delete
 * @return
 *   TRUE on success, FALSE on failure
 */
function datasync_remove_job($job_id) {
  $c = datasync_enabled_consumers(array(datasync_consumer_by_job_id($job_id)));
  if (empty($c)) {
    return FALSE;
  }
  db_query("DELETE FROM {datasync_queue} WHERE job_id=%d", $job_id);
  if (db_affected_rows() == 1) {
    return TRUE;
  }
  return FALSE;
}

/**
 * Return information from the datasync_jobs table on a job
 *
 * @param $job_id
 *   Job ID of the job we are returning info on.
 * @return
 *   An object of the row of the datasync_jobs table.
 */
function datasync_get_job_info($job_id) {
  return db_fetch_object(db_query("SELECT * FROM {datasync_jobs} WHERE job_id=%d", $job_id));
}

/**
 * Enqueue's an item to be versioned and inserted at a later date.
 *
 * @param $guid
 *   GUID of the item we are inserting.
 * @param $job_id
 *   Job ID we are enqueueing an item for.
 * @param $action
 *   Action we are performing (normally DATASYNC_INSERT, DATASYING_UPDATE or DATASYNC_DELETE)
 * @param $data,
 *   Data to be inserted in the datasync_queue table.
 * @param $version
 *   Version of the data we are inserting (must be integer)
 * @return
 *   TRUE on success, FALSE on failure.
 */
function datasync_enqueue($guid, $job_id, $action, $data="", $version=NULL) {
  if ($consumer = datasync_consumer_by_job_id($job_id)) {
    $c = datasync_enabled_consumers(array($consumer));
    if (empty($c)) {
      return FALSE;
    }
    $qid = db_next_id("{datasync_queue}_qid");
    if ($version === NULL) {
      $version = time();
    }
    //print_r(array($qid, $action, $c[0], $guid, $job_id, $version, $data));
    db_query("INSERT INTO {datasync_queue} (qid, action, consumer, guid, job_id, version, data) VALUES (%d, '%s', '%s', '%s', %d, %d, '%s')",
      $qid, $action, $c[0], $guid, $job_id, $version, $data);
    if (db_affected_rows() == 1) {
      return TRUE;
    }
  }
  return FALSE;
}

/**
 * Dequeue an item that has already had its action take place.
 *
 * @param $qid
 *   The QID of the item we are deuqueuing
 * @return
 *   TRUE on success, FALSE on failure
 */
function datasync_dequeue($qid) {
  $c = datasync_enabled_consumers(array(datasync_consumer_by_qid($qid)));
  if (empty($c)) {
    return FALSE;
  }
  //TODO we can make a queue_deleted table to take care of myISAM's
  //speed when using just INSERT and SELECT
  db_query("DELETE FROM {datasync_queue} WHERE qid=%d", $qid);
  if (db_affected_rows() == 1) {
    return TRUE;
  }
  return FALSE;
}

/************************************
 * DATASYNC CALLBACKS
 * //WARNING: AS OF YET, THESE CALLBACKS ARE UNTESTED
 * //TODO add data encryption if necessary
 **************************************/
function datasync_jobs_callback() {
  $input = $_POST['datasync'];
  //string returned means error, array means success
  if (!is_array(($result = datasync_callback_data($input)))) {
    print 'error|' . $result;
    exit;
  }

  switch($result['api_function']) {
    case 'switch_status':
      datasync_switch_status_callback_helper($result);
    break;
    case 'add_job':
      datasync_add_job_callback_helper($result);
    break;
    case 'remove_job':
      datasync_remove_job_callback_helper($result);
    break;
    case 'take_jobs':
      datasync_take_jobs_callback_helper($result);
    break;
    default:
      print 'error|' . t('No such API function @f', array('@f', $result['api_function']));
    break;
  }
  exit;
}

function datasync_switch_status_callback_helper($result) {
  if(!isset($result['job_id']) || !ctype_digit($result['job_id'])) {
    print 'error|' . t('Invalid job_id @jid given', array('@jid' => $result['job_id']));
    return;
  }
  if (isset($result['current_status']) && !ctype_digit($result['current_status'])) {
    print 'error|' . t('Invalid current_status: @str', array('@str' => $result['current_status']));
    return;
  }
  if (isset($result['next_status']) && !ctype_digit($result['next_status'])) {
    print 'error|' . t('Invalid next_status: @str', array('@str' => $result['next_status']));
    return;
  }
  //db_query('BEGIN;');
  $success = datasync_switch_status($result['job_id'], $result['current_status'], $result['next_status']);
  if ($success) {
    //db_query('COMMIT;');
    print 'success';
  }
  else {
    //db_query('ROLLBACK;');
    print 'error|' . t('status switch of "@jid" failed.', array('@jid' => $result['job_id']));
  }
}

function datasync_add_job_callback_helper($result) {
  if (!isset($result['consumer']) || strlen($result['consumer'] > 50)) {
    print 'error|' . t('Invalid consumer: @str', array('@str' => $result['consumer']));
    return;
  }
  //db_query('BEGIN;');
  $success = datasync_add_job($result['consumer']);
  if ($success) {
    //db_query('COMMIT;');
    print 'success';
  }
  else {
    //db_query('ROLLBACK;');
    print 'error|' . t('addition of job "@jid" failed.', array('@jid' => $result['job_id']));
  }
}

function datasync_remove_job_callback_helper($result) {
  if(!isset($result['job_id']) || !ctype_digit($result['job_id'])) {
    print 'error|' . t('Invalid job_id @jid given', array('@jid' => $result['job_id']));
    return;
  }
 //db_query('BEGIN;');
  $success = datasync_remove_job($result['job_id']);
  if ($success) {
    //db_query('COMMIT;');
    print 'success';
  }
  else {
    //db_query('ROLLBACK;');
    print 'error|' . t('addition of job "@jid" failed.', array('@jid' => $result['job_id']));
  }
}

function datasync_take_jobs_callback_helper($result) {
  if (!isset($result['num_jobs']) || !ctype_digit($result['num_jobs'])) {
    print 'error|' . t('Invalid num_jobs: @str', array('@str' => $result['num_jobs']));
    return;
  }
  if(!isset($result['valid_status']) || !($vs = base64_decode($result['valid_status']))) {
    print 'error|' . t('Invalid valid_status: @str', array('@str' => $result['valid_status']));
    return;
  }
  if(!isset($result['consumers']) || !($c = base64_decode($result['consumers']))) {
    print 'error|' . t('Invalid consumers: @str', array('@str' => $result['consumers']));
    return;
  }

  $job_ids = datasync_take_jobs($result['num_jobs'], $vs, $c);
  if ($job_ids !== FALSE) {
    print 'success|' . base64_encode(implode('|', $job_ids));
  }
  else {
    print 'error|' . t('Taking of jobs failed.');
  }
}

function datasync_queue_callback() {
  $input = $_POST['datasync'];
  //string returned means error, array means success
  if (!is_array(($result = datasync_callback_data($input)))) {
    print 'error|' . $result;
    exit;
  }

  switch($result['api_function']) {
    case 'enqueue':
      datasync_enqueue_callback_helper($result);
    break;
    case 'dequeue':
      datasync_dequeue_callback_helper($result);
    break;
    default:
      print 'error|' . t('No such API function @f', array('@f', $result['api_function']));
    break;
  }
  exit;
}

function datasync_enqueue_callback_helper($result) {
  if (!isset($result['guid']) || strlen($result['guid'] > 150)) {
    print 'error|' . t('Invalid guid: @str', array('@str' => $result['guid']));
    return;
  }
  else if (!isset($result['action']) || !ctype_digit($result['action'])) {
    print 'error|' . t('Invalid action: @str', array('@str' => $result['action']));
    return;
  }
  else if (isset($result['version']) && !ctype_digit($result['version'])) {
    print 'error|' . t('Invalid version: @str', array('@str' => $result['version']));
    return;
  }
  else if (!isset($result['version'])) {
    $result['version'] = NULL;
  }
  if (!isset($result['data'])) {
    $result['data'] = '';
  }

  //db_query('BEGIN;');
  $success = datasync_enqueue($result['guid'], $result['job_id'], $result['action'], $result['data'], $result['version']);
  if ($success) {
    //db_query('COMMIT;');
    print 'success';
  }
  else {
    //db_query('ROLLBACK;');
    print 'error|' . t('enqueue of "@jid" failed.', array('@jid' => $result['job_id']));
  }
}

function datasync_dequeue_callback_helper($result) {
  if (!isset($result['qid']) || !ctype_digit($result['qid'])) {
    print 'error|' . t('Invalid qid: @str', array('@str' => $result['guid']));
    return;
  }
  //db_query('BEGIN;');
  $success = datasync_dequeue($result['qid']);
  if ($success) {
    //db_query('COMMIT;');
    print 'success';
  }
  else {
    //db_query('ROLLBACK;');
    print 'error|' . t('dequeue of "@qid" failed.', array('@qid' => $result['qid']));
  }
}
/*
 * Returns a string with the error message on error.
 * Returns an array of data on success.
 */
function datasync_callback_data($input) {
  $input = explode('|', $input);
  if (count($input) != 2) {
    return t('Bad input format.');
  }
  $key = sha1(ds_variable_get('datasync_hmac', ''));
  if ($input[1] != datasync_hmacsha1($key, $input[0], 128, 'sha1')) {
    return t('Invalid hmacsha1.');
  }
  $data = base64_decode($input[0]);
  $data = explode('|', $data);
  if ((count($data) % 2) != 0) {
    return t('Invalid number of items in data array.');
  }
  $items = array();
  for ($i=0; $i<count($data);$i+=2) {
    $items[$data[$i]] = $data[$i+1];
  }
  if(!isset($items['time']) || !ctype_digit($items['time']) || abs(time() - $items['time']) > 15) {
    return t('Invalid time given');
  }
  return $items;
}

function datasync_hmacsha1($key, $data, $blocksize, $hashfunc) {
  if (strlen($key)>$blocksize)
    $key=pack('H*', $hashfunc($key));
  $key=str_pad($key,$blocksize,chr(0x00));
  $ipad=str_repeat(chr(0x36),$blocksize);
  $opad=str_repeat(chr(0x5c),$blocksize);
  $hmac = pack(
    'H*',$hashfunc(
      ($key^$opad).pack(
        'H*',$hashfunc(
          ($key^$ipad).$data
         )
      )
    )
   );
  return bin2hex($hmac);
}

/**************************************
 * INTERFACE STUFF
 **************************************/
function datasync_admin_settings() {
  $form = array();

  $form['hmac'] = array(
    '#type' => 'textfield',
    '#title' => t('HMAC Key'),
    '#description' => t('Secret key that is required to use callbacks.'),
    '#default_value' => ds_variable_get('datasync_hmac', ''),
  );

  $producers = datasync_get_producers(FALSE, FALSE);
  $options = array();
  foreach ($producers as $key => $producer) {
    $options[$key] = $producer['name'];
  }
  $form['producers'] = array(
    '#type' => 'checkboxes',
    '#title' => t('Enable these Producers'),
    '#default_value' => ds_variable_get('datasync_producers', array()),
    '#options' => $options,
  );
  $form['consumers'] = array(
    '#type' => 'checkboxes',
    '#title' => t('Enable these Consumers'),
    '#default_value' => ds_variable_get('datasync_consumers', array()),
    '#options' => $options,
  );

  $consumers = datasync_get_consumers(FALSE, FALSE);
  $options = array();
  foreach ($consumers as $key => $consumer) {
    $options[$key] = $consumer['name'];
  }

  $form['submit'] = array(
    '#type' => 'submit',
    '#value' => t('Save'),
  );

  return $form;
}

function datasync_admin_settings_submit($form_id, $form_values) {
  ds_variable_set('datasync_producers', $form_values['producers']);
  ds_variable_set('datasync_consumers', $form_values['consumers']);
  ds_variable_set('datasync_hmac', $form_values['hmac']);
}

/**************************************
 * MISC HELPER FUNCTIONS
 **************************************/
function datasync_get_producers($enabled_only, $generic_only) {
  static $producers;
  static $producers_enabled;
  static $producers_generic;
  static $result;
  $key = (int)$enabled_only . (int)$generic_only;
  if(!is_array($result) || !isset($result[$key])) {
    if (!is_array($producers)) {
      $producers = module_invoke_all('ds_producers');
      //sort producers by weight
      uasort($producers, 'datasync_sort_by_weight');
    }
    //always fill with producers
    $result[$key] = $producers;
    if ($enabled_only) {
      if (!is_array($producers_enabled)) {
        $enabled = ds_variable_get('datasync_producers', array());
        $producers_enabled = $producers;
        foreach ($producers as $name => $producer) {
          if (!$enabled[$name]) {
            unset($producers_enabled[$name]);
          }
        }
      }
      $result[$key] = array_intersect_key($result[$key], $producers_enabled);
    }
    if ($generic_only) {
      if (!is_array($producers_generic)) {
        $enabled = ds_variable_get('datasync_producers', array());
        $producers_generic = $producers;
        foreach ($producers as $name => $producer) {
          if (!isset($producer['generic']['consumers'])) {
            unset($producers_generic[$name]);
          }
        }
      }
      $result[$key] = array_intersect_key($result[$key], $producers_generic);
    }
  }
  return $result[$key];
}

function datasync_enabled_producers($producers) {
  $enabled = ds_variable_get('datasync_producers', array());
  foreach ($producers as $k => $name) {
    if (!$enabled[$name]) {
      unset($producers[$k]);
    }
  }
  return $producers;
}

function datasync_enabled_consumers($consumers) {
  $enabled = ds_variable_get('datasync_consumers', array());
  foreach ($consumers as $k => $name) {
    if (!$enabled[$name]) {
      unset($consumers[$k]);
    }
  }
  return $consumers;
}

function datasync_get_consumers($enabled_only, $generic_only) {
  static $consumers;
  static $consumers_enabled;
  static $consumers_generic;
  static $result;
  $key = (int)$enabled_only . (int)$generic_only;
  if(!is_array($result) || !isset($result[$key])) {
    if (!is_array($consumers)) {
      $consumers = module_invoke_all('ds_consumers');
      //sort consumers by weight
      uasort($consumers, 'datasync_sort_by_weight');
    }
    //always fill with consumers
    $result[$key] = $consumers;
    if ($enabled_only) {
      if (!is_array($consumers_enabled)) {
        $enabled = ds_variable_get('datasync_consumers', array());
        $consumers_enabled = $consumers;
        foreach ($consumers as $name => $consumer) {
          if (!$enabled[$name]) {
            unset($consumers_enabled[$name]);
          }
        }
      }
      $result[$key] = array_intersect_key($result[$key], $consumers_enabled);
    }
    if ($generic_only) {
      if (!is_array($consumers_generic)) {
        $enabled = ds_variable_get('datasync_consumers', array());
        $consumers_generic = $consumers;
        foreach ($consumers as $name => $consumer) {
          if (!isset($consumer['generic'])) {
            unset($consumers_generic[$name]);
          }
        }
      }
      $result[$key] = array_intersect_key($result[$key], $consumers_generic);
    }
  }
  return $result[$key];
}

function datasync_sort_by_weight($a, $b) {
  $aw = array_key_exists('weight', $a) ? $a['weight'] : 0;
  $bw = array_key_exists('weight', $b) ? $b['weight'] : 0;
  if ($aw == $bw) {
    return 0;
  }
  return ($aw < $bw) ? -1 : 1;
}

function datasync_consumer_by_job_id($job_id) {
  //TODO static here
  return db_result(db_query("SELECT consumer FROM {datasync_jobs} WHERE job_id=%d", $job_id));
}

function datasync_consumer_by_qid($qid) {
  //TODO static here
  return db_result(db_query("SELECT consumer FROM {datasync_queue} WHERE qid=%d", $qid));
}

//Verifies if a process is running in linux
function datasync_process_running($pid) {
  exec("ps $pid", $state);
  return(count($state) >= 2);
}

//Checks if the current host is in the given list
function datasync_host_in_list($hosts) {
  exec("hostname", $current);
  trim($current[0]);
  return in_array($current[0], $hosts) ? $current[0] : FALSE;
}

/**
 * Copied from bootstrap.inc and modified to make sure
 * transactions aren't autocommitted and that we always get
 * the latest values (D6)
 */
function ds_variable_get($name, $default) {
  $result = db_result(db_query("SELECT value FROM {datasync_variable} WHERE name='%s'", $name));
  return $result === FALSE ? $default : unserialize($result);
}

function ds_variable_set($name, $value) {
  db_query("DELETE FROM {datasync_variable} WHERE name = '%s'", $name);
  db_query("INSERT INTO {datasync_variable} (name, value) VALUES ('%s', '%s')", $name, serialize($value));
  if (db_affected_rows() != 1) {
    return FALSE;
  }
  return TRUE;
}

function ds_variable_del($name) {
  db_query("DELETE FROM {datasync_variable} WHERE name = '%s'", $name);
  if (db_error != 0) {
    return FALSE;
  }
  return TRUE;
}

function datasync_log_failure($message) {
  db_query("INSERT INTO {datasync_failures} (value, timestamp) VALUES ('%s', %d)", $message, time());
}
