? .git
? .gitignore
? 584034-9_views_integration.patch
? 584034_4_feeds_view_node_item.patch
? 600584-7_batching.patch
? 641522-15_results.patch
? 641522-16_results.patch
? feeds_mapper_test_include.patch
? libraries/simplepie.inc
? tests/feeds/many_items.rss2
Index: README.txt
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/README.txt,v
retrieving revision 1.15
diff -u -p -r1.15 README.txt
--- README.txt	3 Dec 2009 21:27:40 -0000	1.15
+++ README.txt	11 Jan 2010 19:27:04 -0000
@@ -128,6 +128,11 @@ Description: The table used by FeedsData
              and the importer's id ($importer_id). This default table name can
              be overridden by defining a variable with the same name.
 
+Name:        feeds_node_batch_size
+Default:     20
+             The number of nodes feed node processor creates or deletes in one
+             page load.
+
 Glossary
 ========
 
Index: feeds.install
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.install,v
retrieving revision 1.5
diff -u -p -r1.5 feeds.install
--- feeds.install	3 Dec 2009 20:55:05 -0000	1.5
+++ feeds.install	11 Jan 2010 19:27:04 -0000
@@ -70,12 +70,27 @@ function feeds_schema() {
         'not null' => TRUE,
         'description' => t('Main source resource identifier. E. g. a path or a URL.'),
       ),
+      'batch' => array(
+        'type' => 'text',
+        'size' => 'big',
+        'not null' => FALSE,
+        'description' => t('Cache for batching.'),
+        'serialize' => TRUE,
+      ),
+      'locked' => array(
+        'type' => 'int',
+        'unsigned' => TRUE,
+        'default' => 0,
+        'not null' => TRUE,
+        'description' => 'Contains a UNIX timestamp if this source is locked for processing (e.g. importing or clearing).',
+      ),
     ),
     'primary key' => array('id', 'feed_nid'),
     'indexes' => array(
       'id' => array('id'),
       'feed_nid' => array('feed_nid'),
       'id_source' => array('id', array('source', 128)),
+      'locked' => array('locked'),
     ),
   );
   $schema['feeds_schedule'] = array(
@@ -102,27 +117,26 @@ function feeds_schema() {
         'default' => '',
         'description' => 'Callback to be invoked.',
       ),
-      'last_scheduled_time' => array(
+      'last_executed_time' => array(
         'type' => 'int',
-        'unsigned' => FALSE,
+        'unsigned' => TRUE,
         'default' => 0,
         'not null' => TRUE,
-        'description' => 'Timestamp when this feed was last scheduled to be refreshed.',
+        'description' => 'Timestamp when a job was last executed.',
       ),
       'scheduled' => array(
         'type' => 'int',
-        'unsigned' => FALSE,
-        'size' => 'tiny',
+        'unsigned' => TRUE,
         'default' => 0,
         'not null' => TRUE,
-        'description' => 'Flags whether a feed is scheduled to be refreshed or not.',
+        'description' => 'Timestamp when a job was scheduled. 0 if a job is currently not scheduled.',
       ),
     ),
     'indexes' => array(
       'feed_nid' => array('feed_nid'),
       'id' => array('id'),
       'id_callback' => array('id', 'callback'),
-      'last_scheduled_time' => array('last_scheduled_time'),
+      'last_executed_time' => array('last_executed_time'),
       'scheduled' => array('scheduled'),
     ),
   );
@@ -316,4 +330,65 @@ function feeds_update_6007() {
   db_add_field($ret, 'feeds_node_item', 'hash', $spec);
 
   return $ret;
+}
+
+/**
+ * Add batch field to feeds_source table.
+ */
+function feeds_update_6008() {
+  $ret = array();
+
+  $spec = array(
+    'type' => 'text',
+    'size' => 'big',
+    'not null' => FALSE,
+    'description' => t('Cache for batching.'),
+    'serialize' => TRUE,
+  );
+  db_add_field($ret, 'feeds_source', 'batch', $spec);
+
+  return $ret;
+}
+
+/**
+ * Add locked field to feeds_source table, fix feeds_scheduler fields.
+ *
+ */
+function feeds_update_6009() {
+  $ret = array();
+
+  // Add lock field.
+  $spec = array(
+    'type' => 'int',
+    'unsigned' => TRUE,
+    'default' => 0,
+    'not null' => TRUE,
+    'description' => 'Contains a UNIX timestamp if this source is locked for processing (e.g. importing or clearing).',
+  );
+  db_add_field($ret, 'feeds_source', 'locked', $spec);
+  db_add_index($ret, 'feeds_source', 'locked', array('locked'));
+
+  // Rename last_scheduled_time to last_executed_time, fix unsigned property.
+  $spec = array(
+    'type' => 'int',
+    'size' => 'normal',
+    'unsigned' => TRUE,
+    'default' => 0,
+    'not null' => TRUE,
+    'description' => 'Timestamp when a job was last executed.',
+  );
+  db_change_field($ret, 'feeds_schedule', 'last_scheduled_time', 'last_executed_time', $spec);
+
+  // Make scheduled flag a timestamp.
+  $spec = array(
+    'type' => 'int',
+    'size' => 'normal',
+    'unsigned' => TRUE,
+    'default' => 0,
+    'not null' => TRUE,
+    'description' => 'Timestamp when a job was scheduled. 0 if a job is currently not scheduled.',
+  );
+  db_change_field($ret, 'feeds_schedule', 'scheduled', 'scheduled', $spec);
+
+  return $ret;
 }
\ No newline at end of file
Index: feeds.module
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.module,v
retrieving revision 1.26
diff -u -p -r1.26 feeds.module
--- feeds.module	21 Dec 2009 00:32:55 -0000	1.26
+++ feeds.module	11 Jan 2010 19:27:05 -0000
@@ -31,6 +31,8 @@ define('FEEDS_SCHEDULER_QUEUE', 'feeds_q
  */
 function feeds_cron() {
   feeds_scheduler()->cron();
+  // Release source lock where the lock is older than 3 hours.
+  db_query('UPDATE {feeds_source} SET locked = 0 WHERE locked < %d', FEEDS_REQUEST_TIME - (3600 * 1));
 }
 
 /**
@@ -249,7 +251,7 @@ function feeds_nodeapi(&$node, $op, $for
             $source->addConfig($node->feeds);
             // @todo Too many indirections. Clean up.
             $batch = $source->importer->fetcher->fetch($source);
-            $source->importer->parser->parse($batch, $source);
+            $source->importer->parser->parse($source, $batch);
             // Keep the title in a static cache and populate $node->title on
             // 'presave' as node module looses any changes to $node after
             // 'validate'.
@@ -283,7 +285,7 @@ function feeds_nodeapi(&$node, $op, $for
         // Refresh feed if import on create is selected and suppress_import is
         // not set.
         if ($op == 'insert' && feeds_importer($importer_id)->config['import_on_create'] && !isset($node->feeds['suppress_import'])) {
-          $source->import();
+          feeds_batch_set(t('Importing'), 'import', $importer_id, $node->nid);
         }
         // Add import to scheduler.
         feeds_scheduler()->add($importer_id, 'import', $node->nid);
@@ -370,6 +372,71 @@ function feeds_scheduler_work($feed_info
  */
 
 /**
+ * @defgroup batch Batch functions.
+ */
+
+/**
+ * Batch helper.
+ *
+ * @param $title
+ *   Title to show to user when executing batch.
+ * @param $method
+ *   Method to execute on importer; one of 'import', 'clear' or 'expire'.
+ * @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.
+ */
+function feeds_batch_set($title, $method, $importer_id, $feed_nid = 0) {
+  $batch = array(
+    'title' => $title,
+    'operations' => array(
+      array('feeds_batch', array($method, $importer_id, $feed_nid)),
+    ),
+  );
+  batch_set($batch);
+}
+
+/**
+ * Batch callback.
+ *
+ * @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'] = FALSE;
+  if (feeds_source($importer_id, $feed_nid)->lock()) {
+    switch ($method) {
+      case 'import':
+        $batch_status = feeds_source($importer_id, $feed_nid)->import();
+        break;
+      case 'clear':
+        $batch_status = feeds_source($importer_id, $feed_nid)->clear();
+        break;
+    }
+    if ($batch_status != FEEDS_BATCH_ACTIVE) {
+      feeds_source($importer_id, $feed_nid)->release();
+      $context['finished'] = TRUE;
+    }
+  }
+  else {
+    $context['finished'] = TRUE;
+  }
+}
+
+/**
+ * @} End of "defgroup batch".
+ */
+
+/**
  * @defgroup utility Utility functions
  * @{
  */
Index: feeds.pages.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds.pages.inc,v
retrieving revision 1.8
diff -u -p -r1.8 feeds.pages.inc
--- feeds.pages.inc	21 Dec 2009 01:05:31 -0000	1.8
+++ feeds.pages.inc	11 Jan 2010 19:27:05 -0000
@@ -83,7 +83,7 @@ function feeds_import_form_submit($form,
 
   // Refresh feed if import on create is selected.
   if ($source->importer->config['import_on_create']) {
-    $source->import();
+    feeds_batch_set(t('Importing'), 'import', $form['#importer_id']);
   }
 
   // Add importer to schedule.
@@ -107,8 +107,9 @@ function feeds_import_tab_form(&$form_st
 /**
  * Submit handler for feeds_import_tab_form().
  */
-function feeds_import_tab_form_submit($form, $form_state) {
-  feeds_source($form['#importer_id'], $form['#feed_nid'])->import();
+function feeds_import_tab_form_submit($form, &$form_state) {
+  $form_state['redirect'] = $form['#redirect'];
+  feeds_batch_set(t('Importing'), 'import', $form['#importer_id'], $form['#feed_nid']);
 }
 
 /**
@@ -135,5 +136,6 @@ function feeds_delete_tab_form(&$form_st
  * Submit handler for feeds_delete_tab_form().
  */
 function feeds_delete_tab_form_submit($form, &$form_state) {
-  feeds_source($form['#importer_id'], empty($form['#feed_nid']) ? 0 : $form['#feed_nid'])->clear();
+  $form_state['redirect'] = $form['#redirect'];
+  feeds_batch_set(t('Deleting'), 'clear', $form['#importer_id'], empty($form['#feed_nid']) ? 0 : $form['#feed_nid']);
 }
Index: feeds_defaults/tests/feeds_defaults.test
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/feeds_defaults/tests/feeds_defaults.test,v
retrieving revision 1.3
diff -u -p -r1.3 feeds_defaults.test
--- feeds_defaults/tests/feeds_defaults.test	3 Dec 2009 20:55:05 -0000	1.3
+++ feeds_defaults/tests/feeds_defaults.test	11 Jan 2010 19:27:05 -0000
@@ -300,7 +300,7 @@ class FeedsDefaultsNodeTestCase extends 
       $this->assertEqual($count, 8, 'Found correct number of items.');
 
       // Import again. Feeds only updates items that haven't changed. However,
-      // there are 2 different items with the same GUID in nodes.csv. 
+      // there are 2 different items with the same GUID in nodes.csv.
       // Therefore, feeds will show updates to 2 nodes.
       $this->drupalPost('import/node/import', array(), 'Import');
       $this->assertText('Updated 2 Story nodes.');
@@ -308,14 +308,6 @@ class FeedsDefaultsNodeTestCase extends 
 
     // Remove all nodes.
     $this->drupalPost('import/node/delete-items', array(), 'Delete');
-    $this->assertText('Story Lorem ipsum has been deleted.');
-    $this->assertText('Story Ut wisi enim ad minim veniam has been deleted.');
-    $this->assertText('Story Duis autem vel eum iriure dolor has been deleted.');
-    $this->assertText('Story Typi non habent has been deleted.');
-    $this->assertText('Story Investigationes demonstraverunt has been deleted.');
-    $this->assertText('Story Claritas est etiam has been deleted.');
-    $this->assertText('Story Mirum est notare has been deleted.');
-    $this->assertText('Story Eodem modo typi has been deleted.');
     $this->assertText('Deleted 8 nodes.');
 
     // Import once again.
Index: includes/FeedsBatch.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsBatch.inc,v
retrieving revision 1.2
diff -u -p -r1.2 FeedsBatch.inc
--- includes/FeedsBatch.inc	21 Dec 2009 00:21:00 -0000	1.2
+++ includes/FeedsBatch.inc	11 Jan 2010 19:27:05 -0000
@@ -2,6 +2,22 @@
 // $Id: FeedsBatch.inc,v 1.2 2009/12/21 00:21:00 alexb Exp $
 
 /**
+ * A FeedsBatch object holds the state of an import or clear batch. Used in
+ * FeedsSource class.
+ */
+class FeedsBatch {
+  // Public counters for easier access.
+  public $created;
+  public $updated;
+  public $deleted;
+  public function __construct() {
+    $this->created = 0;
+    $this->updated = 0;
+    $this->deleted = 0;
+  }
+}
+
+/**
  * A FeedsImportBatch is the actual content retrieved from a FeedsSource. On
  * import, it is created on the fetching stage and passed through the parsing
  * and processing stage where it is normalized and consumed.
@@ -9,7 +25,7 @@
  * @see FeedsSource class
  * @see FeedsFetcher class
  */
-class FeedsImportBatch {
+class FeedsImportBatch extends FeedsBatch {
 
   protected $url;
   protected $file_path;
@@ -26,6 +42,7 @@ class FeedsImportBatch {
     $this->url = $url;
     $this->file_path = $file_path;
     $this->items = array();
+    parent::__construct();
   }
 
   /**
@@ -33,21 +50,17 @@ class FeedsImportBatch {
    *   The raw content of the feed.
    */
   public function getRaw() {
-    if (empty($this->raw)) {
-      // Prefer file.
-      if ($this->file_path) {
-        $this->raw = file_get_contents(realpath($this->file_path));
-      }
-      elseif ($this->url) {
-        feeds_include_library('http_request.inc', 'http_request');
-        $result = http_request_get($this->url);
-        if ($result->code != 200) {
-          throw new Exception(t('Download of @url failed with code !code.', array('@url' => $this->url, '!code' => $result->code)));
-        }
-        $this->raw = $result->data;
+    if ($this->file_path) {
+      return file_get_contents(realpath($this->file_path));
+    }
+    elseif ($this->url) {
+      feeds_include_library('http_request.inc', 'http_request');
+      $result = http_request_get($this->url);
+      if ($result->code != 200) {
+        throw new Exception(t('Download of @url failed with code !code.', array('@url' => $this->url, '!code' => $result->code)));
       }
+      return $result->data;
     }
-    return $this->raw;
   }
 
   /**
@@ -107,7 +120,9 @@ class FeedsImportBatch {
    *   removed from the internal array.
    */
   public function shiftItem() {
-    return array_shift($this->items);
+    if (is_array($this->items)) {
+      return array_shift($this->items);
+    }
   }
 
   /**
Index: includes/FeedsImporter.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsImporter.inc,v
retrieving revision 1.8
diff -u -p -r1.8 FeedsImporter.inc
--- includes/FeedsImporter.inc	20 Dec 2009 23:48:38 -0000	1.8
+++ includes/FeedsImporter.inc	11 Jan 2010 19:27:05 -0000
@@ -11,6 +11,9 @@ require_once(dirname(__FILE__) .'/FeedsC
 require_once(dirname(__FILE__) .'/FeedsSource.inc');
 require_once(dirname(__FILE__) .'/FeedsBatch.inc');
 
+// Batch status of a FeedsImporter operation.
+define('FEEDS_BATCH_ACTIVE', 'active');
+
 /**
  * Class defining an importer object. This is the main hub for Feeds module's
  * functionality.
@@ -65,7 +68,7 @@ class FeedsImporter extends FeedsConfigu
    */
   public function expire($time = NULL) {
     try {
-      $this->processor->expire($time);
+      return $this->processor->expire($time);
     }
     catch (Exception $e) {
       drupal_set_message($e->getMessage(), 'error');
Index: includes/FeedsScheduler.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsScheduler.inc,v
retrieving revision 1.9
diff -u -p -r1.9 FeedsScheduler.inc
--- includes/FeedsScheduler.inc	20 Dec 2009 23:48:38 -0000	1.9
+++ includes/FeedsScheduler.inc	11 Jan 2010 19:27:05 -0000
@@ -37,33 +37,32 @@ interface FeedsSchedulerInterface {
   public function remove($importer_id, $callback, $feed_nid = 0);
 
   /**
-   * Work off a given feed identified by $feed_info.
+   * Work off a given feed identified by $job.
    *
-   * @param $feed_info
+   * @param $job
    *   Array where 'importer_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($feed_info);
+  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:
+ * This scheduler uses the last_executed_time paradigm: By storing the time
+ * when a particular job was executed rather than storing when a job should be
+ * executed 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.
+ *
+ * @todo Rename $job to $job
  */
 class FeedsScheduler implements FeedsSchedulerInterface {
 
-  // Only used for debugging.
-  protected $debugTime;
-
   /**
    * Create a single instance of FeedsScheduler.
    */
@@ -90,7 +89,6 @@ class FeedsScheduler implements FeedsSch
    * 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);
@@ -98,49 +96,21 @@ class FeedsScheduler implements FeedsSch
     variable_set('feeds_scheduler_cron', TRUE);
     $start = time();
 
-    // Get feeds configuration, check whether drupal_queue is present and set
-    // parameters accordingly.
-    if ($importers = feeds_importer_load_all()) {
-
-      if ($use_queue = module_exists('drupal_queue')) {
-        drupal_queue_include();
-        $queue = drupal_queue_get(FEEDS_SCHEDULER_QUEUE);
-        $num = variable_get('feeds_schedule_queue_num', 200);
-      }
-      else {
-        $num = variable_get('feeds_schedule_num', 5);
-      }
+    // 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 configurations, pick $num feeds for each
-      // configuration, push to queue or refresh feeds.
+    // 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 (array('import', 'expire') as $callback) {
-
-          // Check whether jobs are scheduled.
           $period = $importer->getSchedulePeriod($callback);
           if ($period != FEEDS_SCHEDULE_NEVER) {
-
-            // Refresh feeds that have a refresh time older than now minus
-            // refresh period.
-            $time = $this->time() - $period;
-
-            $result = db_query_range('SELECT feed_nid, id AS importer_id, callback, last_scheduled_time FROM {feeds_schedule} WHERE id = "%s" AND callback = "%s" AND scheduled = 0 AND (last_scheduled_time < %d OR last_scheduled_time = 0) ORDER BY last_scheduled_time ASC', $importer->id, $callback, $time, 0, $num);
-            while ($feed_info = db_fetch_array($result)) {
-
-              // If drupal_queue is present, add to queue, otherwise work off
-              // immediately.
-              if ($use_queue) {
-                if ($queue->createItem($feed_info)) {
-                  $this->flag($feed_info['importer_id'], $feed_info['callback'], $feed_info['feed_nid']);
-                }
-                else {
-                  watchdog('FeedsScheduler', 'Error adding item to queue.', WATCHDOG_ALERT);
-                }
-              }
-              else {
-                $this->flag($feed_info['importer_id'], $feed_info['callback'], $feed_info['feed_nid']);
-                $this->work($feed_info);
-              }
+            $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.
             }
           }
         }
@@ -157,16 +127,15 @@ class FeedsScheduler implements FeedsSch
    *
    * Add a feed to the scheduler.
    *
-   * @todo Create optional parameter $last_scheduled_time to pass in. Set this
+   * @todo Create optional parameter $last_executed_time to pass in. Set this
    *   value if a feed is refreshed on creation.
-   * @todo Create an abstract interface for items that can be added?
    */
   public function add($importer_id, $callback, $feed_nid = 0) {
     $save = array(
       'id' => $importer_id,
       'callback' => $callback,
       'feed_nid' => $feed_nid,
-      'last_scheduled_time' => 0,
+      'last_executed_time' => 0,
       'scheduled' => 0, // Means NOT scheduled at the moment.
     );
     drupal_write_record('feeds_schedule', $save, array('id', 'callback', 'feed_nid'));
@@ -190,112 +159,102 @@ class FeedsScheduler implements FeedsSch
    * Used as worker callback invoked from feeds_scheduler_refresh() or
    * if drupal_queue is not enabled, directly from $this->cron().
    */
-  public function work($feed_info) {
-    $importer = feeds_importer($feed_info['importer_id']);
-
-    // Only refresh if feed is actually in DB or in default configuration.
+  public function work($job) {
+    $importer = feeds_importer($job['id']);
+    // feeds_importer() might have created a new importer, check.
+    // @todo Make this check a method of FeedsConfigurable and actually hit
+    // DB when invoking.
     if ($importer->export_type != FEEDS_EXPORT_NONE) {
-
-      // Remove scheduled flag, if we fail after this we'd like to try again
-      // next time around.
-      $this->unflag($feed_info['importer_id'], $feed_info['callback'], $feed_info['feed_nid']);
-
-      // There are 2 possible callbacks: expire or 'import'.
-      if ($feed_info['callback'] == 'expire') {
-        try {
-          $importer->expire();
+      try {
+        if ($job['callback'] == 'expire') {
+          if (FEEDS_BATCH_ACTIVE != $importer->expire()) {
+            $this->finished($job);
+          }
         }
-        catch (Exception $e) {
-          watchdog('FeedsScheduler', $e->getMessage(), array(), WATCHDOG_ERROR);
+        elseif ($job['callback'] == 'import') {
+          $source = feeds_source($importer->id, $job['feed_nid']);
+          if ($source->lock()) {
+            if (FEEDS_BATCH_ACTIVE != $source->import()) {
+              $this->finished($job);
+            }
+            $source->release();
+          }
         }
       }
-      elseif ($feed_info['callback'] == 'import') {
-        // Import feed if source is available.
-        $source = feeds_source($importer->id, $feed_info['feed_nid']);
-        if ($source->export_type & FEEDS_EXPORT_NONE) {
-          watchdog('FeedsScheduler', 'Expected source information in database for '. $importer->id .'/'. $feed_info['feed_nid'] .'. Could not find any.', array(), WATCHDOG_ERROR);
-          return;
-        }
-        try {
-          $source->import();
-        }
-        catch (Exception $e) {
-          watchdog('FeedsScheduler', $e->getMessage(), array(), WATCHDOG_ERROR);
-        }
+      catch (Exception $e) {
+        watchdog('FeedsScheduler', $e->getMessage(), array(), WATCHDOG_ERROR);
       }
     }
+    // Make sure job is released.
+    $this->release($job);
   }
 
   /**
-   * Set the internal time of FeedsScheduler.
-   * Use for debugging.
-   *
-   * @param $time
-   *   UNIX time that the scheduler should use for comparing the schedule. Set
-   *   this time to test the behavior of the scheduler in the future or past.
-   *   If set to 0, FeedsScheduler will use the current time.
+   * @return
+   *   Drupal Queue if available, NULL otherwise.
    */
-  public function debugSetTime($time) {
-    $this->debugTime = $time;
+  protected function queue() {
+    if (module_exists('drupal_queue')) {
+      drupal_queue_include();
+      return drupal_queue_get(FEEDS_SCHEDULER_QUEUE);
+    }
   }
 
   /**
-   * Returns the internal time that the scheduler is operating on.
+   * Attempt to reserve a job. If successful work it off or - if Drupal Queue is
+   * available - queue it.
    *
-   * Usually returns FEEDS_REQUEST_TIME, unless a debug time has been set
-   * with debugSetTime();
-   *
-   * @return
-   *   An integer that is a UNIX time.
-   */
-  public function time() {
-    return empty($this->debugTime) ? FEEDS_REQUEST_TIME : $this->debugTime;
+   * 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->release($job);
+          watchdog('FeedsScheduler', 'Error adding item to queue.', WATCHDOG_CRITICAL);
+          return;
+        }
+      }
+      else {
+        $this->work($job);
+      }
+    }
   }
 
   /**
-   * Helper function to flag a feed scheduled.
+   * Release a job.
    *
-   * This function sets the feed's scheduled bit to 1 and updates
-   * last_scheduled_time to $this->time().
+   * This function sets the source's scheduled bit to 0 and thus makes
+   * it eligible for being added to the queue again.
    *
-   * @param $id
-   *   Id of the importer configuration.
-   * @param $callback
-   *   Callback of the job.
-   * @param $feed_nid
-   *   Identifier of the feed node.
+   * @param $job
+   *   A job array.
    */
-  protected function flag($id, $callback, $feed_nid) {
-    $save = array(
-      'id' => $id,
-      'callback' => $callback,
-      'feed_nid' => $feed_nid,
-      'last_scheduled_time' => $this->time(),
-      'scheduled' => 1,
-    );
-    drupal_write_record('feeds_schedule', $save, array('id', 'callback', 'feed_nid'));
+  protected function release($job) {
+    unset($job['last_executed_time']);
+    $job = array(
+      'scheduled' => 0,
+    ) + $job;
+    drupal_write_record('feeds_schedule', $job, array('id', 'callback', 'feed_nid'));
   }
 
   /**
-   * Helper function to flag a feed unscheduled.
-   *
-   * This function sets the feed's scheduled bit to 0 and thus makes
-   * it eligible for being added to the queue again.
+   * Release a job and set its last_executed_time flag.
    *
-   * @param $id
-   *   Id of the importer configuration.
-   * @param $callback
-   *   Callback of the job.
-   * @param $feed_nid
-   *   Identifier of the feed node.
+   * @param $job
+   *   A job array.
    */
-  protected function unflag($id, $callback, $feed_nid) {
-    $save = array(
-      'id' => $id,
-      'callback' => $callback,
-      'feed_nid' => $feed_nid,
+  protected function finished($job) {
+    $job = array(
       'scheduled' => 0,
-    );
-    drupal_write_record('feeds_schedule', $save, array('id', 'callback', 'feed_nid'));
+      'last_executed_time' => FEEDS_REQUEST_TIME,
+    ) + $job;
+    drupal_write_record('feeds_schedule', $job, array('id', 'callback', 'feed_nid'));
   }
 }
Index: includes/FeedsSource.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsSource.inc,v
retrieving revision 1.6
diff -u -p -r1.6 FeedsSource.inc
--- includes/FeedsSource.inc	20 Dec 2009 23:48:38 -0000	1.6
+++ includes/FeedsSource.inc	11 Jan 2010 19:27:05 -0000
@@ -76,6 +76,9 @@ class FeedsSource extends FeedsConfigura
   // The FeedsImporter object that this source is expected to be used with.
   protected $importer;
 
+  // A FeedsBatch object. NULL if there is no active batch.
+  protected $batch;
+
   /**
    * Instantiate a unique object per class/id/feed_nid. Don't use
    * directly, use feeds_source() instead.
@@ -102,16 +105,24 @@ class FeedsSource extends FeedsConfigura
   /**
    * Import a feed: execute, fetching, parsing and processing stage.
    *
+   * Lock a source before importing by using FeedsSource::lock(), after
+   * importing, release with FeedsSource::release().
+   *
    * @todo Iron out and document potential Exceptions.
-   * @todo Support batching.
    * @todo catch exceptions outside of import(), clear() and expire().
    */
   public function import() {
     try {
-      $feed = $this->importer->fetcher->fetch($this);
-      $this->importer->parser->parse($feed, $this);
-      $this->importer->processor->process($feed, $this);
-      unset($feed);
+      if (!$this->batch || !($this->batch instanceof FeedsImportBatch)) {
+        $this->batch = $this->importer->fetcher->fetch($this);
+        $this->importer->parser->parse($this, $this->batch);
+      }
+      if (FEEDS_BATCH_ACTIVE == $this->importer->processor->process($this, $this->batch)) {
+        $this->save();
+        return FEEDS_BATCH_ACTIVE;
+      }
+      unset($this->batch);
+      $this->save();
     }
     catch (Exception $e) {
       drupal_set_message($e->getMessage(), 'error');
@@ -121,12 +132,23 @@ class FeedsSource extends FeedsConfigura
 
   /**
    * Remove all items from a feed.
+   *
+   * Lock a source before clearing by using FeedsSource::lock(), after clearing,
+   * release with FeedsSource::release().
    */
   public function clear() {
     try {
       $this->importer->fetcher->clear($this);
       $this->importer->parser->clear($this);
-      $this->importer->processor->clear($this);
+      if (!$this->batch) {
+        $this->batch = new FeedsBatch();
+      }
+      if (FEEDS_BATCH_ACTIVE == $this->importer->processor->clear($this, $this->batch)) {
+        $this->save();
+        return FEEDS_BATCH_ACTIVE;
+      }
+      unset($this->batch);
+      $this->save();
     }
     catch (Exception $e) {
       drupal_set_message($e->getMessage(), 'error');
@@ -134,6 +156,25 @@ class FeedsSource extends FeedsConfigura
   }
 
   /**
+   * Lock a source for importing or clearing. A source SHOULD be only imported
+   * or cleared if a lock could be obtained.
+   *
+   * @return
+   *   TRUE if lock could be obtained, FALSE otherwise.
+   */
+  public function lock() {
+    db_query("UPDATE {feeds_source} SET locked = %d WHERE id = '%s' AND feed_nid = %d", FEEDS_REQUEST_TIME, $this->id, $this->feed_nid);
+    return db_affected_rows() ? TRUE : FALSE;
+  }
+
+  /**
+   * Release a source after locking it.
+   */
+  public function release() {
+    db_query("UPDATE {feeds_source} SET locked = %d WHERE id = '%s' AND feed_nid = %d", 0, $this->id, $this->feed_nid);
+  }
+
+  /**
    * Save configuration.
    */
   public function save() {
@@ -149,6 +190,7 @@ class FeedsSource extends FeedsConfigura
       'feed_nid' => $this->feed_nid,
       'config' => $config,
       'source' => $source,
+      'batch' => isset($this->batch) ? $this->batch : FALSE,
     );
     // Make sure a source record is present at all time, try to update first,
     // then insert.
@@ -164,12 +206,13 @@ class FeedsSource extends FeedsConfigura
    * @todo Patch CTools to move constants from export.inc to ctools.module.
    */
   public function load() {
-    if ($config = db_result(db_query('SELECT config FROM {feeds_source} WHERE id = "%s" AND feed_nid = %d', $this->id, $this->feed_nid))) {
+    if ($record = db_fetch_object(db_query('SELECT config, batch FROM {feeds_source} WHERE id = "%s" AND feed_nid = %d', $this->id, $this->feed_nid))) {
       // While FeedsSource cannot be exported, we still use CTool's export.inc
       // export definitions.
       ctools_include('export');
       $this->export_type = EXPORT_IN_DATABASE;
-      $this->config = unserialize($config);
+      $this->config = unserialize($record->config);
+      $this->batch = unserialize($record->batch);
     }
   }
 
Index: plugins/FeedsCSVParser.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsCSVParser.inc,v
retrieving revision 1.5
diff -u -p -r1.5 FeedsCSVParser.inc
--- plugins/FeedsCSVParser.inc	20 Dec 2009 23:54:44 -0000	1.5
+++ plugins/FeedsCSVParser.inc	11 Jan 2010 19:27:05 -0000
@@ -9,7 +9,7 @@ class FeedsCSVParser extends FeedsParser
   /**
    * Implementation of FeedsParser::parse().
    */
-  public function parse(FeedsImportBatch $batch, FeedsSource $source) {
+  public function parse(FeedsSource $source, FeedsImportBatch $batch) {
 
     // Parse.
     feeds_include_library('ParserCSV.inc', 'ParserCSV');
Index: plugins/FeedsDataProcessor.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsDataProcessor.inc,v
retrieving revision 1.7
diff -u -p -r1.7 FeedsDataProcessor.inc
--- plugins/FeedsDataProcessor.inc	20 Dec 2009 23:48:38 -0000	1.7
+++ plugins/FeedsDataProcessor.inc	11 Jan 2010 19:27:05 -0000
@@ -14,7 +14,7 @@ class FeedsDataProcessor extends FeedsPr
   /**
    * Implementation of FeedsProcessor::process().
    */
-  public function process(FeedsImportBatch $batch, FeedsSource $source) {
+  public function process(FeedsSource $source, FeedsImportBatch $batch) {
 
     // Count number of created and updated nodes.
     $inserted  = $updated = 0;
@@ -57,7 +57,7 @@ class FeedsDataProcessor extends FeedsPr
    *
    * Delete all data records for feed_nid in this table.
    */
-  public function clear(FeedsSource $source) {
+  public function clear(FeedsSource $source, FeedsBatch $batch) {
     $clause = array(
       'feed_nid' => $source->feed_nid,
     );
Index: plugins/FeedsFeedNodeProcessor.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsFeedNodeProcessor.inc,v
retrieving revision 1.7
diff -u -p -r1.7 FeedsFeedNodeProcessor.inc
--- plugins/FeedsFeedNodeProcessor.inc	20 Dec 2009 23:48:38 -0000	1.7
+++ plugins/FeedsFeedNodeProcessor.inc	11 Jan 2010 19:27:05 -0000
@@ -15,11 +15,7 @@ class FeedsFeedNodeProcessor extends Fee
   /**
    * Implementation of FeedsProcessor::process().
    */
-  public function process(FeedsImportBatch $batch, FeedsSource $source) {
-
-    // Count number of created and updated nodes.
-    $created  = $updated = 0;
-
+  public function process(FeedsSource $source, FeedsImportBatch $batch) {
     while ($item = $batch->shiftItem()) {
 
       // If the target item does not exist OR if update_existing is enabled,
@@ -39,20 +35,20 @@ class FeedsFeedNodeProcessor extends Fee
         node_save($node);
 
         if ($nid) {
-          $updated++;
+          $batch->updated++;
         }
         else {
-          $created++;
+          $batch->created++;
         }
       }
     }
 
     // Set messages.
-    if ($created) {
-      drupal_set_message(t('Created !number !type nodes.', array('!number' => $created, '!type' => $this->config['content_type'])));
+    if ($batch->created) {
+      drupal_set_message(t('Created !number !type nodes.', array('!number' => $batch->created, '!type' => $this->config['content_type'])));
     }
-    elseif ($updated) {
-      drupal_set_message(t('Updated !number !type nodes.', array('!number' => $updated, '!type' => $this->config['content_type'])));
+    elseif ($batch->updated) {
+      drupal_set_message(t('Updated !number !type nodes.', array('!number' => $batch->updated, '!type' => $this->config['content_type'])));
     }
     else {
       drupal_set_message(t('There is no new content.'));
@@ -62,7 +58,7 @@ class FeedsFeedNodeProcessor extends Fee
   /**
    * Implementation of FeedsProcessor::clear().
    */
-  public function clear(FeedsSource $source) {
+  public function clear(FeedsSource $source, FeedsBatch $batch) {
     // Do not support deleting imported items as we would have to delete all
     // items of the content type we imported which may contain nodes that a
     // user created by hand.
Index: plugins/FeedsNodeProcessor.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsNodeProcessor.inc,v
retrieving revision 1.20
diff -u -p -r1.20 FeedsNodeProcessor.inc
--- plugins/FeedsNodeProcessor.inc	21 Dec 2009 01:20:05 -0000	1.20
+++ plugins/FeedsNodeProcessor.inc	11 Jan 2010 19:27:05 -0000
@@ -6,6 +6,9 @@
  * Class definition of FeedsNodeProcessor.
  */
 
+// Create or delete FEEDS_NODE_BATCH_SIZE at a time.
+define('FEEDS_NODE_BATCH_SIZE', 50);
+
 /**
  * Creates nodes from feed items.
  */
@@ -14,10 +17,10 @@ class FeedsNodeProcessor extends FeedsPr
   /**
    * Implementation of FeedsProcessor::process().
    */
-  public function process(FeedsImportBatch $batch, FeedsSource $source) {
+  public function process(FeedsSource $source, FeedsImportBatch $batch) {
 
     // Count number of created and updated nodes.
-    $created = $updated = 0;
+    $processed = 0;
 
     while ($item = $batch->shiftItem()) {
 
@@ -38,10 +41,10 @@ class FeedsNodeProcessor extends FeedsPr
           // If updating populate nid and vid avoiding an expensive node_load().
           $node->nid = $nid;
           $node->vid = db_result(db_query('SELECT vid FROM {node} WHERE nid = %d', $nid));
-          $updated++;
+          $batch->updated++;
         }
         else {
-          $created++;
+          $batch->created++;
         }
 
         // Populate and prepare node object.
@@ -68,14 +71,21 @@ class FeedsNodeProcessor extends FeedsPr
         // Save the node.
         node_save($node);
       }
+
+      // Return FEEDS_BATCH_ACTIVE batch size is reached and items are not
+      // completely consumed yet.
+      $processed++;
+      if ($processed >= variable_get('feeds_node_batch_size', FEEDS_NODE_BATCH_SIZE)) {
+        return FEEDS_BATCH_ACTIVE;
+      }
     }
 
     // Set messages.
-    if ($created) {
-      drupal_set_message(t('Created !number !type nodes.', array('!number' => $created, '!type' => node_get_types('name', $this->config['content_type']))));
+    if ($batch->created) {
+      drupal_set_message(t('Created !number !type nodes.', array('!number' => $batch->created, '!type' => node_get_types('name', $this->config['content_type']))));
     }
-    elseif ($updated) {
-      drupal_set_message(t('Updated !number !type nodes.', array('!number' => $updated, '!type' => node_get_types('name', $this->config['content_type']))));
+    elseif ($batch->updated) {
+      drupal_set_message(t('Updated !number !type nodes.', array('!number' => $batch->updated, '!type' => node_get_types('name', $this->config['content_type']))));
     }
     else {
       drupal_set_message(t('There is no new content.'));
@@ -85,20 +95,21 @@ class FeedsNodeProcessor extends FeedsPr
   /**
    * Implementation of FeedsProcessor::clear().
    */
-  public function clear(FeedsSource $source) {
-    // Count number of deleted nodes.
-    $deleted = 0;
-
-    $result = db_query('SELECT nid FROM {feeds_node_item} WHERE feed_nid = %d', $source->feed_nid);
+  public function clear(FeedsSource $source, FeedsBatch $batch) {
+    $result = db_query_range('SELECT nid FROM {feeds_node_item} WHERE feed_nid = %d', $source->feed_nid, 0, variable_get('feeds_node_batch_size', FEEDS_NODE_BATCH_SIZE));
     while ($node = db_fetch_object($result)) {
       _feeds_node_delete($node->nid);
-      $deleted++;
+      $batch->deleted++;
+    }
+    // If there is still content to be deleted return FALSE.
+    if (db_result(db_query_range('SELECT nid FROM {feeds_node_item} WHERE feed_nid = %d', $source->feed_nid, 0, 1))) {
+      return FEEDS_BATCH_ACTIVE;
     }
 
     // Set message.
     drupal_get_messages('status');
-    if ($deleted) {
-      drupal_set_message(t('Deleted !number nodes.', array('!number' => $deleted)));
+    if ($batch->deleted) {
+      drupal_set_message(t('Deleted !number nodes.', array('!number' => $batch->deleted)));
     }
     else {
       drupal_set_message(t('There is no content to be deleted.'));
Index: plugins/FeedsOPMLParser.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsOPMLParser.inc,v
retrieving revision 1.4
diff -u -p -r1.4 FeedsOPMLParser.inc
--- plugins/FeedsOPMLParser.inc	20 Dec 2009 23:54:44 -0000	1.4
+++ plugins/FeedsOPMLParser.inc	11 Jan 2010 19:27:05 -0000
@@ -14,7 +14,7 @@ class FeedsOPMLParser extends FeedsParse
   /**
    * Implementation of FeedsParser::parse().
    */
-  public function parse(FeedsImportBatch $batch, FeedsSource $source) {
+  public function parse(FeedsSource $source, FeedsImportBatch $batch) {
     feeds_include_library('opml_parser.inc', 'opml_parser');
     $result = opml_parser_parse($batch->getRaw());
     $batch->setTitle($result['title']);
Index: plugins/FeedsParser.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsParser.inc,v
retrieving revision 1.4
diff -u -p -r1.4 FeedsParser.inc
--- plugins/FeedsParser.inc	20 Dec 2009 23:54:44 -0000	1.4
+++ plugins/FeedsParser.inc	11 Jan 2010 19:27:05 -0000
@@ -11,12 +11,12 @@ abstract class FeedsParser extends Feeds
    *
    * Extending classes must implement this method.
    *
-   * @param $batch
-   *   FeedsImportBatch returned by fetcher.
    * @param FeedsSource $source
    *   Source information.
+   * @param $batch
+   *   FeedsImportBatch returned by fetcher.
    */
-  public abstract function parse(FeedsImportBatch $batch, FeedsSource $source);
+  public abstract function parse(FeedsSource $source, FeedsImportBatch $batch);
 
   /**
    * Clear all caches for results for given source.
Index: plugins/FeedsProcessor.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsProcessor.inc,v
retrieving revision 1.6
diff -u -p -r1.6 FeedsProcessor.inc
--- plugins/FeedsProcessor.inc	20 Dec 2009 23:48:38 -0000	1.6
+++ plugins/FeedsProcessor.inc	11 Jan 2010 19:27:05 -0000
@@ -10,12 +10,15 @@ abstract class FeedsProcessor extends Fe
    * Process the result of the parser or previous processors.
    * Extending classes must implement this method.
    *
-   * @param FeedsImportBatch $batch
-   *   The current feed import data passed in from the parsing stage.
    * @param FeedsSource $source
    *   Source information about this import.
+   * @param FeedsImportBatch $batch
+   *   The current feed import data passed in from the parsing stage.
+   *
+   * @return
+   *   NULL if all items have been cleared, FEEDS_BATCH_ACTIVE if not.
    */
-  public abstract function process(FeedsImportBatch $batch, FeedsSource $source);
+  public abstract function process(FeedsSource $source, FeedsImportBatch $batch);
 
   /**
    * Remove all stored results or stored results up to a certain time for this
@@ -27,8 +30,14 @@ abstract class FeedsProcessor extends Fe
    *   item pertains to a certain souce is by using $source->feed_nid. It is the
    *   processor's responsibility to store the feed_nid of an imported item in
    *   the processing stage.
+   * @param FeedsBatch $batch
+   *   A FeedsBatch object for tracking information such as how many
+   *   items have been deleted total between page loads.
+   *
+   * @return
+   *   NULL if all items have been cleared, FEEDS_BATCH_ACTIVE if not.
    */
-  public abstract function clear(FeedsSource $source);
+  public abstract function clear(FeedsSource $source, FeedsBatch $batch);
 
   /**
    * Delete feed items younger than now - $time.
@@ -39,8 +48,12 @@ abstract class FeedsProcessor extends Fe
    *   If implemented, all items produced by this configuration that are older
    *   than FEEDS_REQUEST_TIME - $time
    *   If $time === NULL processor should use internal configuration.
+   *
+   * @return
+   *   NULL if all items have been expired, FEEDS_BATCH_ACTIVE if not.
    */
-  public function expire($time = NULL) {}
+  public function expire($time = NULL) {
+  }
 
   /**
    * Execute mapping on an item.
Index: plugins/FeedsSimplePieParser.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsSimplePieParser.inc,v
retrieving revision 1.7
diff -u -p -r1.7 FeedsSimplePieParser.inc
--- plugins/FeedsSimplePieParser.inc	20 Dec 2009 23:54:44 -0000	1.7
+++ plugins/FeedsSimplePieParser.inc	11 Jan 2010 19:27:06 -0000
@@ -11,7 +11,7 @@ class FeedsSimplePieParser extends Feeds
   /**
    * Implementation of FeedsParser::parse().
    */
-  public function parse(FeedsImportBatch $batch, FeedsSource $source) {
+  public function parse(FeedsSource $source, FeedsImportBatch $batch) {
     feeds_include_library('simplepie.inc', 'simplepie');
 
     // Initialize SimplePie.
Index: plugins/FeedsSyndicationParser.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsSyndicationParser.inc,v
retrieving revision 1.10
diff -u -p -r1.10 FeedsSyndicationParser.inc
--- plugins/FeedsSyndicationParser.inc	20 Dec 2009 23:54:44 -0000	1.10
+++ plugins/FeedsSyndicationParser.inc	11 Jan 2010 19:27:06 -0000
@@ -11,7 +11,7 @@ class FeedsSyndicationParser extends Fee
   /**
    * Implementation of FeedsParser::parse().
    */
-  public function parse(FeedsImportBatch $batch, FeedsSource $source) {
+  public function parse(FeedsSource $source, FeedsImportBatch $batch) {
     feeds_include_library('common_syndication_parser.inc', 'common_syndication_parser');
     $result = common_syndication_parser_parse($batch->getRaw());
     $batch->setTitle($result['title']);
Index: plugins/FeedsTermProcessor.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsTermProcessor.inc,v
retrieving revision 1.3
diff -u -p -r1.3 FeedsTermProcessor.inc
--- plugins/FeedsTermProcessor.inc	20 Dec 2009 23:27:28 -0000	1.3
+++ plugins/FeedsTermProcessor.inc	11 Jan 2010 19:27:06 -0000
@@ -14,7 +14,7 @@ class FeedsTermProcessor extends FeedsPr
   /**
    * Implementation of FeedsProcessor::process().
    */
-  public function process(FeedsImportBatch $batch, FeedsSource $source) {
+  public function process(FeedsSource $source, FeedsImportBatch $batch) {
 
     if (empty($this->config['vocabulary'])) {
       throw new Exception(t('You must define a vocabulary for Taxonomy term processor before importing.'));
@@ -72,13 +72,9 @@ class FeedsTermProcessor extends FeedsPr
   }
 
   /**
-   * Implement clear.
-   *
-   * @param $source
-   *   FeedsSource of this term. FeedsTermProcessor does not heed this
-   *   parameter, it deletes all terms from a vocabulary.
+   * Implementation of FeedsProcessor::clear().
    */
-  public function clear(FeedsSource $source) {
+  public function clear(FeedsSource $source, FeedsBatch $batch) {
     $deleted = 0;
 
     $result = db_query('SELECT tid FROM {term_data} WHERE vid = %d', $this->config['vocabulary']);
Index: plugins/FeedsUserProcessor.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsUserProcessor.inc,v
retrieving revision 1.4
diff -u -p -r1.4 FeedsUserProcessor.inc
--- plugins/FeedsUserProcessor.inc	20 Dec 2009 23:48:38 -0000	1.4
+++ plugins/FeedsUserProcessor.inc	11 Jan 2010 19:27:06 -0000
@@ -14,7 +14,7 @@ class FeedsUserProcessor extends FeedsPr
   /**
    * Implementation of FeedsProcessor::process().
    */
-  public function process(FeedsImportBatch $batch, FeedsSource $source) {
+  public function process(FeedsSource $source, FeedsImportBatch $batch) {
 
     // Count number of created and updated nodes.
     $created  = $updated = $failed = 0;
@@ -66,13 +66,9 @@ class FeedsUserProcessor extends FeedsPr
   }
 
   /**
-   * Implement clear.
-   *
-   * @param $source
-   *   FeedsSource of this term. FeedsTermProcessor does not heed this
-   *   parameter, it deletes all terms from a vocabulary.
+   * Implementation of FeedsProcessor::clear().
    */
-  public function clear(FeedsSource $source) {
+  public function clear(FeedsSource $source, FeedsBatch $batch) {
     // Do not support deleting users as we have no way of knowing which ones we
     // imported.
     throw new Exception(t('User processor does not support deleting users.'));
Index: tests/feeds.test
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/tests/feeds.test,v
retrieving revision 1.7
diff -u -p -r1.7 feeds.test
--- tests/feeds.test	20 Dec 2009 23:48:38 -0000	1.7
+++ tests/feeds.test	11 Jan 2010 19:27:06 -0000
@@ -592,7 +592,6 @@ class FeedsSchedulerTestCase extends Fee
    */
   public function setUp() {
     parent::setUp('feeds', 'feeds_ui', 'ctools');
-
     $this->drupalLogin(
       $this->drupalCreateUser(
         array(
@@ -600,13 +599,6 @@ class FeedsSchedulerTestCase extends Fee
         )
       )
     );
-  }
-
-  /**
-   * Test scheduling on cron.
-   */
-  public function testScheduling() {
-    // Create default configuration.
     $this->createFeedConfiguration();
     $this->addMappings('syndication',
       array(
@@ -637,7 +629,12 @@ class FeedsSchedulerTestCase extends Fee
         ),
       )
     );
+  }
 
+  /**
+   * Test scheduling on cron.
+   */
+  public function testScheduling() {
     // Create 10 feed nodes. Turn off import on create before doing that.
     $edit = array(
       'import_on_create' => FALSE,
@@ -656,7 +653,7 @@ class FeedsSchedulerTestCase extends Fee
 
     // There should be feeds_schedule_num (= 10) feeds updated now.
     $schedule = array();
-    $count = db_result(db_query('select COUNT(*) from {feeds_schedule} WHERE last_scheduled_time <> 0'));
+    $count = db_result(db_query('select COUNT(*) from {feeds_schedule} WHERE last_executed_time <> 0'));
     $this->assertEqual($count, 10, '10 feeds refreshed on cron.');
 
     // There should be 100 story nodes in the database.
@@ -669,7 +666,7 @@ class FeedsSchedulerTestCase extends Fee
 
     // There should be feeds_schedule_num X 2 (= 20) feeds updated now.
     $schedule = array();
-    $result = db_query('select feed_nid, last_scheduled_time, scheduled from {feeds_schedule} WHERE last_scheduled_time <> 0');
+    $result = db_query('select feed_nid, last_executed_time, scheduled from {feeds_schedule} WHERE last_executed_time <> 0');
     while ($row = db_fetch_object($result)) {
       $schedule[$row->feed_nid] = $row;
     }
@@ -691,9 +688,9 @@ class FeedsSchedulerTestCase extends Fee
     // The import_period setting of the feed configuration is 1800, there
     // shouldn't be any change to the database now.
     $equal = TRUE;
-    $result = db_query('select feed_nid, last_scheduled_time, scheduled from {feeds_schedule} WHERE last_scheduled_time <> 0');
+    $result = db_query('select feed_nid, last_executed_time, scheduled from {feeds_schedule} WHERE last_executed_time <> 0');
     while ($row = db_fetch_object($result)) {
-      $equal = $equal && ($row->last_scheduled_time == $schedule[$row->feed_nid]->last_scheduled_time);
+      $equal = $equal && ($row->last_executed_time == $schedule[$row->feed_nid]->last_executed_time);
     }
     $this->assertTrue($equal, 'Schedule did not change.');
 
@@ -712,19 +709,18 @@ class FeedsSchedulerTestCase extends Fee
     $this->assertText('Refresh: as often as possible');
 
     // Hit cron again, 4 times now.
-    $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
-    $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
-    $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
-    $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
+    for ($i = 0; $i < 4; $i++) {
+      $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
+    }
 
     // Refresh period is set to 'as often as possible'. All scheduled times
     // should have changed now.
     // There should not be more nodes than before.
     $equal = FALSE;
     $output = '';
-    $result = db_query('select feed_nid, last_scheduled_time, scheduled from {feeds_schedule} WHERE last_scheduled_time <> 0');
+    $result = db_query('select feed_nid, last_executed_time, scheduled from {feeds_schedule} WHERE last_executed_time <> 0');
     while ($row = db_fetch_object($result)) {
-      $equal = $equal || ($row->last_scheduled_time == $schedule[$row->feed_nid]->last_scheduled_time);
+      $equal = $equal || ($row->last_executed_time == $schedule[$row->feed_nid]->last_executed_time);
     }
     $this->assertFalse($equal, 'Every feed schedule time changed.');
 
@@ -735,6 +731,31 @@ class FeedsSchedulerTestCase extends Fee
     // @todo Use debug time feature in FeedsScheduler and test behavior in future.
     // @todo How do I call an API function on the test system from the test script?
   }
+
+  /**
+   * Test batching on cron.
+   */
+  function testBatching() {
+    $nid = $this->createFeedNode('syndication', $GLOBALS['base_url'] .'/'. drupal_get_path('module', 'feeds') .'/tests/feeds/many_items.rss2');
+    $this->assertText('Created 1000 Story nodes.');
+    $this->drupalPost('node/'. $nid .'/delete-items', array(), 'Delete');
+    $this->assertText('Deleted 1000 nodes.');
+
+    // Hit cron 20 times, assert correct number of story nodes.
+    for ($i = 0; $i < 20; $i++) {
+      $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
+      $this->assertEqual(50 * ($i + 1), db_result(db_query("SELECT COUNT(*) FROM {node} WHERE type = 'story'")));
+    }
+
+    // Delete a couple of nodes, then hit cron again. They should not be replaced
+    // as the minimum update time is 30 minutes.
+    for ($i = 1010; $i < 1015; $i++) {
+      $this->drupalPost("node/$i/delete", array(), 'Delete');
+    }
+    $this->assertEqual(995, db_result(db_query("SELECT COUNT(*) FROM {node} WHERE type = 'story'")));
+    $this->drupalGet($GLOBALS['base_url'] .'/cron.php');
+    $this->assertEqual(995, db_result(db_query("SELECT COUNT(*) FROM {node} WHERE type = 'story'")));
+  }
 }
 
 /**
Index: tests/feeds.test.inc
===================================================================
RCS file: /cvs/drupal-contrib/contributions/modules/feeds/tests/feeds.test.inc,v
retrieving revision 1.6
diff -u -p -r1.6 feeds.test.inc
--- tests/feeds.test.inc	20 Dec 2009 23:48:38 -0000	1.6
+++ tests/feeds.test.inc	11 Jan 2010 19:27:07 -0000
@@ -182,9 +182,9 @@ class FeedsWebTestCase extends DrupalWeb
     $this->assertEqual($config['FeedsHTTPFetcher']['source'], $feed_url, t('URL in DB correct.'));
 
     // Check whether feed got properly added to scheduler.
-    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND feed_nid = %d AND callback = "import" AND last_scheduled_time = 0 AND scheduled = 0', $id, $nid)));
+    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND feed_nid = %d AND callback = "import" AND last_executed_time = 0 AND scheduled = 0', $id, $nid)));
     // There must be only one entry for 'expire' - no matter how many actual feed nodes exist.
-    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND callback = "expire" AND last_scheduled_time = 0 AND scheduled = 0', $id)));
+    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND callback = "expire" AND last_executed_time = 0 AND scheduled = 0', $id)));
 
     return $nid;
   }
@@ -248,9 +248,9 @@ class FeedsWebTestCase extends DrupalWeb
     $this->assertEqual($config['FeedsHTTPFetcher']['source'], $feed_url, t('URL in DB correct.'));
 
     // Check whether feed got properly added to scheduler.
-    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND feed_nid = 0 AND callback = "import" AND last_scheduled_time = 0 AND scheduled = 0', $id)));
+    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND feed_nid = 0 AND callback = "import" AND last_executed_time = 0 AND scheduled = 0', $id)));
     // There must be only one entry for callback 'expire' - no matter what the feed_nid is.
-    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND callback = "expire" AND last_scheduled_time = 0 AND scheduled = 0', $id)));
+    $this->assertEqual(1, db_result(db_query('SELECT COUNT(*) FROM {feeds_schedule} WHERE id = "%s" AND callback = "expire" AND last_executed_time = 0 AND scheduled = 0', $id)));
   }
 
   /**