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',
    '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.
  $importers = feeds_reschedule();
  if ($importers) {
    // @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);
  }

  // Sync the files in the cache directory with entries in the cache every now
  // and then. By default: every six hours.
  $last_check = variable_get('feeds_sync_cache_feeds_http_last_check');
  $interval = variable_get('feeds_sync_cache_feeds_http_interval', 21600);
  if ($last_check < (REQUEST_TIME - $interval)) {
    // Check first if the task isn't already queued.
    $queue = DrupalQueue::get('feeds_sync_cache_feeds_http');
    if ($queue->numberOfItems() < 1) {
      // Queue sync task.
      FeedsHTTPCache::getInstance('cache_feeds_http')->startSync();
    }
    variable_set('feeds_sync_cache_feeds_http_last_check', REQUEST_TIME);
  }
}

/**
 * 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',
  );
  $queues['feeds_sync_cache_feeds_http'] = array(
    'worker callback' => 'feeds_sync_cache_feeds_http',
  );

  return $queues;
}

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

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

/**
 * Scheduler callback for expiring content.
 */
function feeds_source_expire(array $job) {
  $source = _feeds_queue_worker_helper($job, 'expire');
  if ($source->doesExist()) {
    $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) {
    // Eventually remove the job from job scheduler.
    JobScheduler::get('feeds_source_import')->remove($job);
  }
  catch (Exception $e) {
    $source->log($method, $e->getMessage(), array(), WATCHDOG_ERROR);
  }

  return $source;
}

/**
 * Scheduler callback for syncing the feeds cache directory with the entries in
 * the cache.
 */
function feeds_sync_cache_feeds_http(array $job) {
  FeedsHTTPCache::getInstance('cache_feeds_http')->sync($job['files']);
}

/**
 * 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);
}