<?php
// $Id$
/**
* @file
* FeedsScheduler class and related.
*/
/**
* Describe a scheduler.
*/
interface FeedsSchedulerInterface {
/**
* Run Drupal cron.
*/
public function cron();
/**
* Add a feed to the schedule.
*
* @param $importer_id
* Id of a FeedsImporter object.
* @param $callback
* The callback to invoke on importer. Either 'import' or 'expire'.
* @param $feed_nid
* Feed nid that identifies the source for this configuration.
*/
public function add($importer_id, $callback, $feed_nid = 0);
/**
* Remove a feed from the schedule.
*
* @param $feed_nid
* Feed nid that identifies the source for this configuration.
*/
public function remove($importer_id, $callback, $feed_nid = 0);
/**
* Work off a given feed identified by $job.
*
* @param $job
* Array where 'id' key is the id of a FeedsImporter object,
* and 'feed_nid' is the feed node id that identifies the
* source of a FeedsSource object.
*/
public function work($job);
}
/**
* Implementation of FeedsSchedulerInterface.
*
* This scheduler uses the last_scheduled_time paradigm: By storing the time
* when a particular feed was scheduled to be refreshed last rather than
* storing when a feed should be _refreshed_ next, we gain two advantages:
*
* 1) If a feed's import_period setting changes, it has immediate effects -
* without batch updating an existing schedule.
* 2) The time between refreshes will always be scheduled based on when it
* has been scheduled last. Less drift occurs.
*/
class FeedsScheduler implements FeedsSchedulerInterface {
/**
* Create a single instance of FeedsScheduler.
*/
public static function instance() {
static $instance;
if (!isset($instance)) {
$class = variable_get('feeds_scheduler_class', 'FeedsScheduler');
$instance = new $class();
}
return $instance;
}
/**
* Protect constructor.
*/
protected function __construct() {}
/**
* Implementation of FeedsSchedulerInterface::cron().
*
* Refreshes scheduled feeds.
*
* If drupal_queue is present, only pushes refresh tasks to queue and
* returns. If drupal_queue is not available, works off tasks.
*/
public function cron() {
// Check and set scheduler semaphore, take time.
if (variable_get('feeds_scheduler_cron', FALSE)) {
watchdog('FeedsScheduler', 'Last cron process did not finish.', array(), WATCHDOG_ERROR);
}
variable_set('feeds_scheduler_cron', TRUE);
$start = time();
// Release schedule lock where the lock is older than 1 hour.
db_query('UPDATE {feeds_schedule} SET scheduled = 0 WHERE scheduled < %d', FEEDS_REQUEST_TIME - 3600);
// Iterate over feed importers, pick $num jobs for each of them and
// schedule them.
if ($importers = feeds_importer_load_all()) {
$num = $this->queue() ? variable_get('feeds_schedule_queue_num', 200) : variable_get('feeds_schedule_num', 5);
foreach ($importers as $importer) {
foreach ($importer->getScheduleCallbacks() as $callback) {
$period = $importer->getSchedulePeriod($callback);
if ($period != FEEDS_SCHEDULE_NEVER) {
$result = db_query_range('SELECT feed_nid, id, callback, last_executed_time FROM {feeds_schedule} WHERE id = "%s" AND callback = "%s" AND scheduled = 0 AND (last_executed_time < %d OR last_executed_time = 0) ORDER BY last_executed_time ASC', $importer->id, $callback, FEEDS_REQUEST_TIME - $period, 0, $num);
while ($job = db_fetch_array($result)) {
$this->schedule($job);
// @todo Add time limit.
}
}
}
}
}
// Unflag and post a message that we're done.
variable_set('feeds_scheduler_cron', FALSE);
watchdog('FeedsScheduler', 'Finished processing schedule after !time.', array('!time' => format_interval(time() - $start)));
}
/**
* Implementation of FeedsSchedulerInterface::add().
*
* Add a feed to the scheduler.
*
* @todo Create optional parameter $last_executed_time to pass in. Set this
* value if a feed is refreshed on creation.
*/
public function add($importer_id, $callback, $feed_nid = 0) {
$save = array(
'id' => $importer_id,
'callback' => $callback,
'feed_nid' => $feed_nid,
'last_executed_time' => 0,
'scheduled' => 0, // Means NOT scheduled at the moment.
);
drupal_write_record('feeds_schedule', $save, array('id', 'callback', 'feed_nid'));
if (!db_affected_rows()) {
drupal_write_record('feeds_schedule', $save);
}
}
/**
* Implementation of FeedsSchedulerInterface::remove().
*/
public function remove($importer_id, $callback, $feed_nid = 0) {
db_query('DELETE FROM {feeds_schedule} WHERE id = "%s" AND callback = "%s" AND feed_nid = %d', $importer_id, $callback, $feed_nid);
}
/**
* Implementation of FeedsSchedulerInterface::work().
*
* Refresh a feed.
*
* Used as worker callback invoked from feeds_scheduler_refresh() or
* if drupal_queue is not enabled, directly from $this->cron().
*/
public function work($job) {
$importer = feeds_importer($job['id']);
try {
if (FEEDS_BATCH_COMPLETE == $importer->work($job)) {
$this->finished($job);
}
}
catch (Exception $e) {
watchdog('FeedsScheduler', $e->getMessage(), array(), WATCHDOG_ERROR);
}
// Make sure that job is not scheduled after this method has executed.
$this->unschedule($job);
}
/**
* @return
* Drupal Queue if available, NULL otherwise.
*/
protected function queue() {
if (module_exists('drupal_queue')) {
drupal_queue_include();
return drupal_queue_get(FEEDS_SCHEDULER_QUEUE);
}
}
/**
* Attempt to reserve a job. If successful work it off or - if Drupal Queue is
* available - queue it.
*
* The lock/release mechanism makes sure that an item does not get queued
* twice. It has a different purpose than the FeedsSource level locking
* which is in place to avoid concurrent import/clear operations on a source.
*
* @param $job
* A job array.
*/
protected function schedule($job) {
db_query("UPDATE {feeds_schedule} SET scheduled = %d WHERE id = '%s' AND feed_nid = %d AND callback = '%s'", FEEDS_REQUEST_TIME, $job['id'], $job['feed_nid'], $job['callback']);
if (db_affected_rows()) {
if ($this->queue()) {
if (!$queue->createItem($job)) {
$this->unschedule($job);
watchdog('FeedsScheduler', 'Error adding item to queue.', WATCHDOG_CRITICAL);
return;
}
}
else {
$this->work($job);
}
}
}
/**
* Remove a job from schedule.
*
* This function sets the source's scheduled bit to 0 and thus makes
* it eligible for being added to the queue again.
*
* @param $job
* A job array.
*/
protected function unschedule($job) {
unset($job['last_executed_time']);
$job = array(
'scheduled' => 0,
) + $job;
drupal_write_record('feeds_schedule', $job, array('id', 'callback', 'feed_nid'));
}
/**
* Release a job and set its last_executed_time flag.
*
* @param $job
* A job array.
*/
protected function finished($job) {
$job = array(
'scheduled' => 0,
'last_executed_time' => FEEDS_REQUEST_TIME,
) + $job;
drupal_write_record('feeds_schedule', $job, array('id', 'callback', 'feed_nid'));
}
}