feeds.module

<?php

/**
 * @file
 * Feeds - basic API functions and hook implementations.
 */

// Common request time, use as point of reference and to avoid calls to time().
define('FEEDS_REQUEST_TIME', time());
// Do not schedule a feed for refresh.
define('FEEDS_SCHEDULE_NEVER', -1);
// Never expire feed items.
define('FEEDS_EXPIRE_NEVER', -1);
// An object that is not persistent. Compare EXPORT_IN_DATABASE, EXPORT_IN_CODE.
define('FEEDS_EXPORT_NONE', 0x0);
// Status of batched operations.
define('FEEDS_BATCH_COMPLETE', 1.0);
define('FEEDS_BATCH_ACTIVE', 0.0);

/**
 * @defgroup hooks Hook and callback implementations
 * @{
 */

/**
 * Implements hook_hook_info().
 */
function feeds_hook_info() {
  $hooks = array(
    'feeds_plugins',
    'feeds_after_parse',
    'feeds_before_import',
    'feeds_before_update',
    'feeds_presave',
    'feeds_after_save',
    'feeds_after_import',
    'feeds_after_clear',
    'feeds_processor_targets_alter',
    'feeds_parser_sources_alter',
  );

  return array_fill_keys($hooks, array('group' => 'feeds'));
}

/**
 * Implements hook_cron().
 */
function feeds_cron() {
  // Expire old log entries.
  db_delete('feeds_log')
    ->condition('request_time', REQUEST_TIME - 604800, '<')
    ->execute();

  // Find importers that need to be rescheduled.
  if (!$importers = feeds_reschedule()) {
    return;
  }

  // @todo Maybe we should queue this somehow as well. This could be potentially
  // very long.
  $sources = db_query("SELECT feed_nid, id FROM {feeds_source} WHERE id IN (:ids)", array(':ids' => $importers));

  foreach ($sources as $source) {
    feeds_source($source->id, $source->feed_nid)->schedule();
  }

  feeds_reschedule(FALSE);
}

/**
 * Implements hook_cron_job_scheduler_info().
 *
 * Compare queue names with key names in feeds_cron_queue_info().
 */
function feeds_cron_job_scheduler_info() {
  $info = array();
  $info['feeds_source_import'] = array(
    'queue name' => 'feeds_source_import',
  );
  $info['feeds_source_clear'] = array(
    'queue name' => 'feeds_source_clear',
  );
  $info['feeds_source_expire'] = array(
    'queue name' => 'feeds_source_expire',
  );
  $info['feeds_push_unsubscribe'] = array(
    'queue name' => 'feeds_push_unsubscribe',
  );
  return $info;
}

/**
 * Implements hook_cron_queue_info().
 */
function feeds_cron_queue_info() {
  $queues = array();
  $queues['feeds_source_import'] = array(
    'worker callback' => 'feeds_source_import',
    'time' => 60,
  );
  $queues['feeds_source_clear'] = array(
    'worker callback' => 'feeds_source_clear',
  );
  $queues['feeds_source_expire'] = array(
    'worker callback' => 'feeds_source_expire',
  );
  $queues['feeds_push_unsubscribe'] = array(
    'worker callback' => 'feeds_push_unsubscribe',
  );

  return $queues;
}

/**
 * Scheduler callback for importing from a source.
 */
function feeds_source_import(array $job) {
  $source = _feeds_queue_worker_helper($job, 'import');
  $source->scheduleImport();
}

/**
 * Scheduler callback for deleting all items from a source.
 */
function feeds_source_clear(array $job) {
  $source = _feeds_queue_worker_helper($job, 'clear');
  $source->scheduleClear();
}

/**
 * Scheduler callback for expiring content.
 */
function feeds_source_expire(array $job) {
  $source = _feeds_queue_worker_helper($job, 'expire');
  $source->scheduleExpire();
}

/**
 * Executes a method on a feed source.
 *
 * @param array $job
 *   The job being run.
 * @param string $method
 *   The method to execute.
 */
function _feeds_queue_worker_helper(array $job, $method) {
  $source = feeds_source($job['type'], $job['id']);
  try {
    $source->existing()->$method();
  }
  catch (FeedsNotExistingException $e) {
    // Do nothing.
  }
  catch (Exception $e) {
    $source->log($method, $e->getMessage(), array(), WATCHDOG_ERROR);
  }

  return $source;
}

/**
 * Scheduler callback for unsubscribing from PuSH hubs.
 */
function feeds_push_unsubscribe($job) {
  $source = feeds_source($job['type'], $job['id']);
  $fetcher = feeds_plugin('FeedsHTTPFetcher', $source->importer->id);
  $fetcher->unsubscribe($source);
}

/**
 * Batch API worker callback. Used by FeedsSource::startBatchAPIJob().
 *
 * @see FeedsSource::startBatchAPIJob().
 *
 * @todo Harmonize Job Scheduler API callbacks with Batch API callbacks?
 *
 * @param $method
 *   Method to execute on importer; one of 'import' or 'clear'.
 * @param $importer_id
 *   Identifier of a FeedsImporter object.
 * @param $feed_nid
 *   If importer is attached to content type, feed node id identifying the
 *   source to be imported.
 * @param $context
 *   Batch context.
 */
function feeds_batch($method, $importer_id, $feed_nid = 0, &$context) {
  $context['finished'] = FEEDS_BATCH_COMPLETE;
  try {
    $context['finished'] = feeds_source($importer_id, $feed_nid)->$method();
  }
  catch (Exception $e) {
    drupal_set_message($e->getMessage(), 'error');
  }
}

/**
 * Reschedule one or all importers.
 *
 * @param string $importer_id