#849986 lyricnz, alex_b: Cleaner batch support.

CHANGELOG.txt

@@ -3,6 +3,7 @@
 Feeds 6.x 1.0 XXXXXXXXXXXXXXXXXX
 --------------------------------
 
+- #849986 lyricnz, alex_b: Cleaner batch support.
 - #866492 lyricnz: Clean up tests.
 - #862444 pounard: Do not name files after their enclosure class.
 - #851570 morningtime: Avoid trailing slashes when passing file paths to

includes/FeedsBatch.inc

 <?php
 // $Id$
 
+// Batch stages.
+define('FEEDS_FETCHING', 'fetching');
+define('FEEDS_PARSING', 'parsing');
+define('FEEDS_PROCESSING', 'processing');
+define('FEEDS_CLEARING', 'clearing');
+
 /**
  * A FeedsBatch object holds the state of an import or clear batch.
  *
  * Used in FeedsSource class. Counter variables are public for easier access.
  */
 class FeedsBatch {
-  // Maximum number of items in this batch. This is not necessarily the current
-  // number of items in this batch
-  public $total;
-  // Number of items created.
-  public $created;
-  // Number of items updated or replaced.
-  public $updated;
-  // Number of items deleted.
-  public $deleted;
+  // Total of each stage of this batch.
+  protected $total;
+  // Progress of each stage of this batch.
+  protected $progress;
   public function __construct() {
-    $this->total = 0;
-    $this->created = 0;
-    $this->updated = 0;
-    $this->deleted = 0;
+    $this->total = array();
+    $this->progress = array();
+  }
+
+  /**
+   * Set the total for a stage.
+   */
+  public function setTotal($stage, $total) {
+    $this->total[$stage] = $total;
+  }
+
+  /**
+   * Get the total for a stage.
+   */
+  public function getTotal($stage) {
+    return $this->total[$stage];
+  }
+
+  /**
+   * Set progress for a stage.
+   *
+   * @param $stage
+   *   The stage to set the progress for. One of FEEDS_FETCHING, FEEDS_PARSING,
+   *   FEEDS_PROCESING or FEEDS_CLEARING.
+   * @param $progress
+   *   The number of items worked off for the given stage. This should be the
+   *   number of items worked off across all page loads, not just the present
+   *   page load.
+   */
+  public function setProgress($stage, $progress) {
+    $this->progress[$stage] = $progress;
+  }
+
+  /**
+   * Report progress.
+   *
+   * @param $stage
+   *   The stage to set the progress for. One of FEEDS_FETCHING, FEEDS_PARSING,
+   *   FEEDS_PROCESING or FEEDS_CLEARING.
+   */
+  public function getProgress($stage = NULL) {
+    if ($stage) {
+      $progress = $this->progress[$stage];
+      if ($progress == FEEDS_BATCH_COMPLETE) {
+        return FEEDS_BATCH_COMPLETE;
+      }
+      $total = $this->total[$stage];
+    }
+    else {
+      $complete = TRUE;
+      $progress = 0;
+      foreach ($this->progress as $p) {
+        $progress += $p;
+        $complete &= $p == FEEDS_BATCH_COMPLETE;
+      }
+      if ($complete) {
+        return FEEDS_BATCH_COMPLETE;
+      }
+      $total = array_sum($this->total);
+    }
+    $progress = (1.0 / $total) * $progress;
+    return $progress == FEEDS_BATCH_COMPLETE ? 0.999 : $progress;
   }
 }
 
@@ -52,10 +111,33 @@ class FeedsBatch {
  * }
  * @endcode
  *
+ * If a processing task is very slow, it can be batched over multiple page
+ * loads. For batching the consumer loop can be left while the current progress
+ * is set on the batch object. If the current progress is not
+ * FEEDS_BATCH_COMPLETE the processor will be called again on a subsequent page
+ * load to continue where it has left off. For an example, see
+ * FeedsNodeProcessor::process().
+ *
+ * @code
+ * $created = 0;
+ * while ($item = $batch->shiftItem()) {
+ *   $object = $this->map($item);
+ *   $object->save();
+ *   $created++; // Created in this page load.
+ *   $batch->created++; // Created total.
+ *   if ($created > MAX_CREATED) {
+ *     $batch->setProgress(FEEDS_PROCESSING, $batch->created);
+ *     return;
+ *   }
+ * }
+ * $batch->setProgress(FEEDS_PROCESSING, FEEDS_BATCH_COMPLETE);
+ * @endcode
+ *
  * Note: Knowledge of the internal structure of a single item in the $items
  * array is managed by the mapping API specified in FeedsParser class and
  * FeedsProcessor class.
  *
+ * @see FeedsBatch
  * @see FeedsFileBatch
  * @see FeedsHTTPBatch
  */
@@ -65,13 +147,23 @@ class FeedsImportBatch extends FeedsBatch {
   protected $link;
   protected $items;
   protected $raw;
+  public $created;
+  public $updated;
 
   public function __construct($raw = '') {
-    $this->raw = $raw;
+    parent::__construct();
+    $this->progress = array(
+      FEEDS_FETCHING => FEEDS_BATCH_COMPLETE,
+      FEEDS_PARSING => FEEDS_BATCH_COMPLETE,
+      FEEDS_PROCESSING => FEEDS_BATCH_COMPLETE,
+    );
     $this->title = '';
     $this->description = '';
     $this->link = '';
     $this->items = array();
+    $this->raw = $raw;
+    $this->created = 0;
+    $this->updated = 0;
   }
 
   /**
@@ -170,7 +262,6 @@ class FeedsImportBatch extends FeedsBatch {
    */
   public function setItems($items) {
     $this->items = $items;
-    $this->total = count($this->items);
   }
 
   /**
@@ -178,6 +269,27 @@ class FeedsImportBatch extends FeedsBatch {
    */
   public function addItem($item) {
     $this->items[] = $item;
-    $this->total = count($this->items);
+  }
+
+  /**
+   * Get number of items.
+   */
+  public function getItemCount() {
+    return count($this->items);
+  }
+}
+
+/**
+ * Batch class for batched deleting of items.
+ */
+class FeedsClearBatch extends FeedsBatch {
+  // Number of items deleted.
+  public $deleted;
+  public function __construct() {
+    parent::__construct();
+    $this->progress = array(
+      FEEDS_CLEARING => FEEDS_BATCH_COMPLETE,
+    );
+    $this->deleted = 0;
   }
 }

includes/FeedsSource.inc

@@ -141,7 +141,8 @@ class FeedsSource extends FeedsConfigurable {
         $this->batch = $this->importer->fetcher->fetch($this);
         $this->importer->parser->parse($this->batch, $this);
       }
-      $result = $this->importer->processor->process($this->batch, $this);
+      $this->importer->processor->process($this->batch, $this);
+      $result = $this->batch->getProgress();
       if ($result == FEEDS_BATCH_COMPLETE) {
         unset($this->batch);
         module_invoke_all('feeds_after_import', $this->importer, $this);
@@ -170,10 +171,11 @@ class FeedsSource extends FeedsConfigurable {
     try {
       $this->importer->fetcher->clear($this);
       $this->importer->parser->clear($this);
-      if (!$this->batch) {
-        $this->batch = new FeedsBatch();
+      if (!$this->batch || !($this->batch instanceof FeedsClearBatch)) {
+        $this->batch = new FeedsClearBatch();
       }
-      $result = $this->importer->processor->clear($this->batch, $this);
+      $this->importer->processor->clear($this->batch, $this);
+      $result = $this->batch->getProgress();
       if ($result == FEEDS_BATCH_COMPLETE) {
         unset($this->batch);
       }

plugins/FeedsDataProcessor.inc

@@ -60,8 +60,6 @@ class FeedsDataProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('There are no new items.'));
     }
-
-    return FEEDS_BATCH_COMPLETE;
   }
 
   /**
@@ -75,7 +73,6 @@ class FeedsDataProcessor extends FeedsProcessor {
     );
     $num = $this->handler()->delete($clause);
     drupal_set_message(format_plural($num, 'Deleted @number item.', 'Deleted @number items.', array('@number' => $num)));
-    return FEEDS_BATCH_COMPLETE;
   }
 
   /**

plugins/FeedsFeedNodeProcessor.inc

@@ -53,8 +53,6 @@ class FeedsFeedNodeProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('There is no new content.'));
     }
-
-    return FEEDS_BATCH_COMPLETE;
   }
 
   /**

plugins/FeedsNodeProcessor.inc

@@ -25,8 +25,11 @@ class FeedsNodeProcessor extends FeedsProcessor {
    */
   public function process(FeedsImportBatch $batch, FeedsSource $source) {
 
-    // Keep track of processed items in this pass.
+    // Keep track of processed items in this pass, set total number of items.
     $processed = 0;
+    if (!$batch->getTotal(FEEDS_PROCESSING)) {
+      $batch->setTotal(FEEDS_PROCESSING, $batch->getItemCount());
+    }
 
     while ($item = $batch->shiftItem()) {
 
@@ -60,7 +63,8 @@ class FeedsNodeProcessor extends FeedsProcessor {
 
       $processed++;
       if ($processed >= variable_get('feeds_node_batch_size', FEEDS_NODE_BATCH_SIZE)) {
-        return (1.0 / ($batch->total + 1)) * ($batch->updated + $batch->created); // Add + 1 to make sure that result is not 1.0 = finished.
+        $batch->setProgress(FEEDS_PROCESSING, $batch->created + $batch->updated);
+        return;
       }
     }
 
@@ -74,16 +78,16 @@ class FeedsNodeProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('There is no new content.'));
     }
-
-    return FEEDS_BATCH_COMPLETE;
+    $batch->setProgress(FEEDS_PROCESSING, FEEDS_BATCH_COMPLETE);
   }
 
   /**
    * Implementation of FeedsProcessor::clear().
    */
   public function clear(FeedsBatch $batch, FeedsSource $source) {
-    if (empty($batch->total)) {
-      $batch->total = db_result(db_query("SELECT COUNT(nid) FROM {feeds_node_item} WHERE id = '%s' AND feed_nid = %d", $source->id, $source->feed_nid));
+    if (!$batch->getTotal(FEEDS_CLEARING)) {
+      $total = db_result(db_query("SELECT COUNT(nid) FROM {feeds_node_item} WHERE id = '%s' AND feed_nid = %d", $source->id, $source->feed_nid));
+      $batch->setTotal(FEEDS_CLEARING, $total);
     }
     $result = db_query_range("SELECT nid FROM {feeds_node_item} WHERE id = '%s' AND feed_nid = %d", $source->id, $source->feed_nid, 0, variable_get('feeds_node_batch_size', FEEDS_NODE_BATCH_SIZE));
     while ($node = db_fetch_object($result)) {
@@ -91,7 +95,8 @@ class FeedsNodeProcessor extends FeedsProcessor {
       $batch->deleted++;
     }
     if (db_result(db_query_range("SELECT nid FROM {feeds_node_item} WHERE id = '%s' AND feed_nid = %d", $source->id, $source->feed_nid, 0, 1))) {
-      return (1.0 / ($batch->total + 1)) * $batch->deleted;
+      $batch->setProgress(FEEDS_CLEARING, $batch->deleted);
+      return;
     }
 
     // Set message.
@@ -102,7 +107,7 @@ class FeedsNodeProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('There is no content to be deleted.'));
     }
-    return FEEDS_BATCH_COMPLETE;
+    $batch->setProgress(FEEDS_CLEARING, FEEDS_BATCH_COMPLETE);
   }
 
   /**

plugins/FeedsProcessor.inc

@@ -20,10 +20,6 @@ abstract class FeedsProcessor extends FeedsPlugin {
    *   The current feed import data passed in from the parsing stage.
    * @param FeedsSource $source
    *   Source information about this import.
-   *
-   * @return
-   *   FEEDS_BATCH_COMPLETE if all items have been processed, a float between 0
-   *   and 0.99* indicating progress otherwise.
    */
   public abstract function process(FeedsImportBatch $batch, FeedsSource $source);
 
@@ -40,10 +36,6 @@ abstract class FeedsProcessor extends FeedsPlugin {
    *   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.
-   *
-   * @return
-   *   FEEDS_BATCH_COMPLETE if all items have been processed, a float between 0
-   *   and 0.99* indicating progress otherwise.
    */
   public abstract function clear(FeedsBatch $batch, FeedsSource $source);
 

plugins/FeedsTermProcessor.inc

@@ -82,8 +82,6 @@ class FeedsTermProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('There are no new terms.'));
     }
-
-    return FEEDS_BATCH_COMPLETE;
   }
 
   /**
@@ -108,7 +106,6 @@ class FeedsTermProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('No terms to be deleted.'));
     }
-    return FEEDS_BATCH_COMPLETE;
   }
 
   /**

plugins/FeedsUserProcessor.inc

@@ -80,8 +80,6 @@ class FeedsUserProcessor extends FeedsProcessor {
     else {
       drupal_set_message(t('There are no new users.'));
     }
-
-    return FEEDS_BATCH_COMPLETE;
   }
 
   /**