? 744660-44_full_batching.patch ? 744660-45_full_batching.patch ? 849986-4_clean_batching.patch ? 850652-1_column_names.patch ? libraries/simplepie.inc Index: includes/FeedsBatch.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsBatch.inc,v retrieving revision 1.12 diff -u -p -r1.12 FeedsBatch.inc --- includes/FeedsBatch.inc 20 Jun 2010 01:10:29 -0000 1.12 +++ includes/FeedsBatch.inc 11 Jul 2010 13:31:08 -0000 @@ -1,26 +1,86 @@ 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 + * A float between 0 and .9999 that indicates the progress on the given + * stage or FEEDS_BATCH_COMPLETE if the stage is complete. + */ + public function setProgress($stage, $progress = FEEDS_BATCH_COMPLETE) { + $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) { + $total = $progress = 0; + if ($stage) { + $total = $this->total[$stage]; + $progress = $this->progress[$stage]; + if ($progress == FEEDS_BATCH_COMPLETE) { + return FEEDS_BATCH_COMPLETE; + } + } + else { + foreach ($this->total as $t) { + $total += $t; + } + $complete = TRUE; + foreach ($this->progress as $p) { + $progress += $p; + $complete &= $p == FEEDS_BATCH_COMPLETE; + } + if ($complete) { + return FEEDS_BATCH_COMPLETE; + } + } + $progress = (1.0 / $total) * $progress; + return $progress == FEEDS_BATCH_COMPLETE ? 0.999 : $progress; } } @@ -52,10 +112,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,8 +148,18 @@ class FeedsImportBatch extends FeedsBatc protected $link; protected $items; protected $raw; + public $created; + public $updated; public function __construct($raw = '') { + parent::__construct(); + $this->progress = array( + FEEDS_FETCHING => FEEDS_BATCH_COMPLETE, + FEEDS_PARSING => FEEDS_BATCH_COMPLETE, + FEEDS_PROCESSING => FEEDS_BATCH_COMPLETE, + ); + $this->created = 0; + $this->updated = 0; $this->raw = $raw; $this->title = ''; $this->description = ''; @@ -75,6 +168,57 @@ class FeedsImportBatch extends FeedsBatc } /** + * Determine whether batch needs fetching. + * + * A batch needs fetching when the fetching stage is not complete and the + * parsing stage and the processing stage are complete. + * + * @return + * TRUE if batch needs fetching, FALSE if it doesn't. + */ + public function needsFetching() { + if ($this->progress[FEEDS_FETCHING] != FEEDS_BATCH_COMPLETE && + $this->progress[FEEDS_PARSING] == FEEDS_BATCH_COMPLETE && + $this->progress[FEEDS_PROCESSING] == FEEDS_BATCH_COMPLETE) { + return TRUE; + } + return FALSE; + } + + /** + * Determine whether batch needs parsing. + * + * A batch needs parsing if the parsing stage is not complete and the + * processing stage is complete. + * + * @return + * TRUE if batch needs parsing, FALSE if it doesn't. + */ + public function needsParsing() { + if ($this->progress[FEEDS_PARSING] != FEEDS_BATCH_COMPLETE && + $this->progress[FEEDS_PROCESSING] == FEEDS_BATCH_COMPLETE) { + return TRUE; + } + return FALSE; + } + + /** + * Determine whether batch needs processing. + * + * A batch needs processing when the processing stage is not complete, no + * matter what status the fetching or parsing state are in. + * + * @return + * TRUE if batch needs processing, FALSE if it doesn't. + */ + public function needsProcessing() { + if ($this->progress[FEEDS_PROCESSING] != FEEDS_BATCH_COMPLETE) { + return TRUE; + } + return FALSE; + } + + /** * @return * The raw content from the source as a string. * @@ -170,7 +314,6 @@ class FeedsImportBatch extends FeedsBatc */ public function setItems($items) { $this->items = $items; - $this->total = count($this->items); } /** @@ -178,6 +321,27 @@ class FeedsImportBatch extends FeedsBatc */ 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; } } Index: includes/FeedsSource.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/includes/FeedsSource.inc,v retrieving revision 1.14 diff -u -p -r1.14 FeedsSource.inc --- includes/FeedsSource.inc 11 Jul 2010 01:10:52 -0000 1.14 +++ includes/FeedsSource.inc 11 Jul 2010 13:31:08 -0000 @@ -140,11 +140,22 @@ class FeedsSource extends FeedsConfigura */ public function import() { try { - if (!$this->batch || !($this->batch instanceof FeedsImportBatch)) { + if (!$this->batch || + !($this->batch instanceof FeedsImportBatch) || + $this->batch->needsFetching()) { $this->batch = $this->importer->fetcher->fetch($this); + $this->batch->setProgress(FEEDS_PARSING, FEEDS_BATCH_ACTIVE); + } + if ($this->batch->needsParsing()) { + $this->batch->setProgress(FEEDS_PARSING, FEEDS_BATCH_COMPLETE); $this->importer->parser->parse($this->batch, $this); + $this->batch->setProgress(FEEDS_PROCESSING, FEEDS_BATCH_ACTIVE); + } + if ($this->batch->needsProcessing()) { + $this->batch->setProgress(FEEDS_PROCESSING, FEEDS_BATCH_COMPLETE); + $this->importer->processor->process($this->batch, $this); } - $result = $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); @@ -173,10 +184,11 @@ class FeedsSource extends FeedsConfigura 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); } Index: plugins/FeedsCSVParser.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsCSVParser.inc,v retrieving revision 1.12 diff -u -p -r1.12 FeedsCSVParser.inc --- plugins/FeedsCSVParser.inc 11 Jul 2010 01:36:39 -0000 1.12 +++ plugins/FeedsCSVParser.inc 11 Jul 2010 13:31:08 -0000 @@ -1,6 +1,9 @@ setColumnNames($header); - // Set line limit to 0 and start byte to last position and parse rest. - $parser->setLineLimit(0); - $parser->setStartByte($parser->lastLinePos()); + // Set line limit and start byte to last position and parse rest. + $parser->setLineLimit(FEEDS_PARSER_CSV_BATCH_SIZE); + $parser->setStartByte($batch->parser_pos ? $batch->parser_pos : $parser->lastLinePos()); $rows = $parser->parse($iterator); + // Remember last parser position if we did not get to the end of the file. + if ($pos = $parser->lastLinePos()) { + $batch->parser_pos = $pos; + $batch->setTotal(FEEDS_PARSING, filesize(realpath($batch->getFilePath()))); + $batch->setProgress(FEEDS_PARSING, $pos); + } + else { + $batch->setProgress(FEEDS_PARSING, FEEDS_BATCH_COMPLETE); + } + // Populate batch. $batch->setItems($rows); } Index: plugins/FeedsDataProcessor.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsDataProcessor.inc,v retrieving revision 1.14 diff -u -p -r1.14 FeedsDataProcessor.inc --- plugins/FeedsDataProcessor.inc 8 Jul 2010 17:19:16 -0000 1.14 +++ plugins/FeedsDataProcessor.inc 11 Jul 2010 13:31:08 -0000 @@ -60,8 +60,6 @@ class FeedsDataProcessor extends FeedsPr else { drupal_set_message(t('There are no new items.')); } - - return FEEDS_BATCH_COMPLETE; } /** @@ -75,7 +73,6 @@ class FeedsDataProcessor extends FeedsPr ); $num = $this->handler()->delete($clause); drupal_set_message(format_plural($num, 'Deleted @number item.', 'Deleted @number items.', array('@number' => $num))); - return FEEDS_BATCH_COMPLETE; } /** Index: plugins/FeedsFeedNodeProcessor.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsFeedNodeProcessor.inc,v retrieving revision 1.12 diff -u -p -r1.12 FeedsFeedNodeProcessor.inc --- plugins/FeedsFeedNodeProcessor.inc 28 Apr 2010 22:18:30 -0000 1.12 +++ plugins/FeedsFeedNodeProcessor.inc 11 Jul 2010 13:31:08 -0000 @@ -53,8 +53,6 @@ class FeedsFeedNodeProcessor extends Fee else { drupal_set_message(t('There is no new content.')); } - - return FEEDS_BATCH_COMPLETE; } /** Index: plugins/FeedsNodeProcessor.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsNodeProcessor.inc,v retrieving revision 1.42 diff -u -p -r1.42 FeedsNodeProcessor.inc --- plugins/FeedsNodeProcessor.inc 10 Jul 2010 22:41:47 -0000 1.42 +++ plugins/FeedsNodeProcessor.inc 11 Jul 2010 13:31:09 -0000 @@ -24,8 +24,11 @@ class FeedsNodeProcessor extends FeedsPr */ 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()) { @@ -59,30 +62,33 @@ class FeedsNodeProcessor extends FeedsPr $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; } } + $batch->setProgress(FEEDS_PROCESSING, FEEDS_BATCH_COMPLETE); // Set messages. - if ($batch->created) { - drupal_set_message(format_plural($batch->created, 'Created @number @type node.', 'Created @number @type nodes.', array('@number' => $batch->created, '@type' => node_get_types('name', $this->config['content_type'])))); - } - elseif ($batch->updated) { - drupal_set_message(format_plural($batch->updated, 'Updated @number @type node.', '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.')); + if ($batch->getProgress() == FEEDS_BATCH_COMPLETE) { + if ($batch->created) { + drupal_set_message(format_plural($batch->created, 'Created @number @type node.', 'Created @number @type nodes.', array('@number' => $batch->created, '@type' => node_get_types('name', $this->config['content_type'])))); + } + elseif ($batch->updated) { + drupal_set_message(format_plural($batch->updated, 'Updated @number @type node.', '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.')); + } } - - return 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)) { @@ -90,7 +96,8 @@ class FeedsNodeProcessor extends FeedsPr $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. @@ -101,7 +108,7 @@ class FeedsNodeProcessor extends FeedsPr else { drupal_set_message(t('There is no content to be deleted.')); } - return FEEDS_BATCH_COMPLETE; + $batch->setProgress(FEEDS_CLEARING, FEEDS_BATCH_COMPLETE); } /** Index: plugins/FeedsProcessor.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsProcessor.inc,v retrieving revision 1.12 diff -u -p -r1.12 FeedsProcessor.inc --- plugins/FeedsProcessor.inc 8 Jul 2010 17:19:16 -0000 1.12 +++ plugins/FeedsProcessor.inc 11 Jul 2010 13:31:09 -0000 @@ -14,10 +14,6 @@ abstract class FeedsProcessor extends Fe * 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); @@ -34,10 +30,6 @@ 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. - * - * @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); Index: plugins/FeedsTermProcessor.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsTermProcessor.inc,v retrieving revision 1.9 diff -u -p -r1.9 FeedsTermProcessor.inc --- plugins/FeedsTermProcessor.inc 28 Apr 2010 22:18:30 -0000 1.9 +++ plugins/FeedsTermProcessor.inc 11 Jul 2010 13:31:09 -0000 @@ -77,8 +77,6 @@ class FeedsTermProcessor extends FeedsPr else { drupal_set_message(t('There are no new terms.')); } - - return FEEDS_BATCH_COMPLETE; } /** @@ -103,7 +101,6 @@ class FeedsTermProcessor extends FeedsPr else { drupal_set_message(t('No terms to be deleted.')); } - return FEEDS_BATCH_COMPLETE; } /** Index: plugins/FeedsUserProcessor.inc =================================================================== RCS file: /cvs/drupal-contrib/contributions/modules/feeds/plugins/FeedsUserProcessor.inc,v retrieving revision 1.12 diff -u -p -r1.12 FeedsUserProcessor.inc --- plugins/FeedsUserProcessor.inc 19 Jun 2010 15:57:13 -0000 1.12 +++ plugins/FeedsUserProcessor.inc 11 Jul 2010 13:31:09 -0000 @@ -80,8 +80,6 @@ class FeedsUserProcessor extends FeedsPr else { drupal_set_message(t('There are no new users.')); } - - return FEEDS_BATCH_COMPLETE; } /**