diff -urN ../feeds.orig/includes/FeedsBatch.inc ./includes/FeedsBatch.inc --- ../feeds.orig/includes/FeedsBatch.inc 2010-06-20 03:10:29.000000000 +0200 +++ ./includes/FeedsBatch.inc 2010-07-19 16:48:57.000000000 +0200 @@ -1,26 +1,124 @@ limit; + } + + /** + * Set items number to process on each batch page. + * + * @param int $limit + * Positive integer to give an arbitrary limit. + * Give FeedsBatch::ITEM_LIMIT_NONE for no limit. + */ + public function setLimit($limit) { + $this->limit = $limit; + } + + // 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 + * 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 +150,33 @@ * } * @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 +186,19 @@ protected $link; protected $items; protected $raw; + protected $offset = 0; + 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 +207,111 @@ } /** + * 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; + } + + /** + * Process items for the current batch page. If the internal limit is no + * limit, then process all items. + * + * @param FeedsProcessor $processor + * Current processor to use. + * @param FeedsSource $source + * Current feeds source. + */ + public final function process(FeedsProcessor $processor, FeedsSource $source) { + // Keep track of processed items in this pass, set total number of items. + if (!$this->total[FEEDS_PROCESSING]) { + $this->total[FEEDS_PROCESSING] = $this->getItemCount(); + } + + $limit = $this->limit > 0 ? $this->limit : count($this->items); + + for ($processed = $this->offset; $processed < $limit; $processed++) { + switch ($processor->process($item, $source)) { + case FeedsProcessor::FEEDS_PROCESSOR_CREATED: + $batch->created++; + break; + + case FeedsProcessor::FEEDS_PROCESSOR_UPDATED: + $this->updated++; + break; + + default: + // FIXME: An error happened, treat it. + } + + if ($processed < count($this->items)) { + $this->setProgress(FEEDS_PROCESSING, $this->created + $this->updated); + return; + } + } + + $this->setProgress(FEEDS_PROCESSING, FEEDS_BATCH_COMPLETE); + + // Set messages. + // FIXME: Find a way to let the processor advertise the imported item type. + if ($batch->getProgress() == FEEDS_BATCH_COMPLETE) { + if ($batch->created) { + drupal_set_message(format_plural($this->created, 'Created 1 item.', 'Created @count items.')); + } + elseif ($batch->updated) { + drupal_set_message(format_plural($this->updated, 'Updated 1 items.', 'Updated @count items.')); + } + else { + drupal_set_message(t('There is no new item.')); + } + } + } + + /** * @return * The raw content from the source as a string. * @@ -138,7 +375,10 @@ * removed from the internal array. */ public function shiftItem() { - return array_shift($this->items); + if ($this->offset < count($this->items)) { + return $this->items[$this->offset++]; + } + return NULL; } /** @@ -170,7 +410,6 @@ */ public function setItems($items) { $this->items = $items; - $this->total = count($this->items); } /** @@ -178,6 +417,27 @@ */ 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; } } diff -urN ../feeds.orig/includes/FeedsSource.inc ./includes/FeedsSource.inc --- ../feeds.orig/includes/FeedsSource.inc 2010-07-11 15:53:17.000000000 +0200 +++ ./includes/FeedsSource.inc 2010-07-19 16:32:57.000000000 +0200 @@ -137,11 +137,22 @@ */ 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->batch->process($this->importer->processor, $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); @@ -170,10 +181,11 @@ 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); } diff -urN ../feeds.orig/includes/FeedsSource.inc.orig ./includes/FeedsSource.inc.orig --- ../feeds.orig/includes/FeedsSource.inc.orig 1970-01-01 01:00:00.000000000 +0100 +++ ./includes/FeedsSource.inc.orig 2010-07-19 14:54:38.000000000 +0200 @@ -0,0 +1,303 @@ +feed_nid = $feed_nid; + $this->importer = feeds_importer($importer_id); + parent::__construct($importer_id); + $this->load(); + } + + /** + * Preview = fetch and parse a feed. + * + * @return + * FeedsImportBatch object, fetched and parsed. + * + * @throws + * Throws Exception if an error occurs when fetching or parsing. + */ + public function preview() { + $batch = $this->importer->fetcher->fetch($this); + $this->importer->parser->parse($batch, $this); + return $batch; + } + + /** + * Import a feed: execute fetching, parsing and processing stage. + * + * @return + * FEEDS_BATCH_COMPLETE if the import process finished. A decimal between + * 0.0 and 0.9 periodic if import is still in progress. + * + * @throws + * Throws Exception if an error occurs when importing. + */ + public function import() { + try { + if (!$this->batch || !($this->batch instanceof FeedsImportBatch)) { + $this->batch = $this->importer->fetcher->fetch($this); + $this->importer->parser->parse($this->batch, $this); + } + $result = $this->importer->processor->process($this->batch, $this); + if ($result == FEEDS_BATCH_COMPLETE) { + unset($this->batch); + module_invoke_all('feeds_after_import', $this->importer, $this); + } + } + catch (Exception $e) { + unset($this->batch); + $this->save(); + throw $e; + } + $this->save(); + return $result; + } + + /** + * Remove all items from a feed. + * + * @return + * FEEDS_BATCH_COMPLETE if the clearing process finished. A decimal between + * 0.0 and 0.9 periodic if clearing is still in progress. + * + * @throws + * Throws Exception if an error occurs when clearing. + */ + public function clear() { + try { + $this->importer->fetcher->clear($this); + $this->importer->parser->clear($this); + if (!$this->batch) { + $this->batch = new FeedsBatch(); + } + $result = $this->importer->processor->clear($this->batch, $this); + if ($result == FEEDS_BATCH_COMPLETE) { + unset($this->batch); + } + } + catch (Exception $e) { + unset($this->batch); + $this->save(); + throw $e; + } + $this->save(); + return $result; + } + + /** + * Save configuration. + */ + public function save() { + $config = $this->getConfig(); + // Alert implementers of FeedsSourceInterface to the fact that we're saving. + foreach ($this->importer->plugin_types as $type) { + $this->importer->$type->sourceSave($this); + } + // Store the source property of the fetcher in a separate column so that we + // can do fast lookups on it. + $source = ''; + if (isset($config[get_class($this->importer->fetcher)]['source'])) { + $source = $config[get_class($this->importer->fetcher)]['source']; + } + $object = array( + 'id' => $this->id, + 'feed_nid' => $this->feed_nid, + 'config' => $config, + 'source' => $source, + 'batch' => isset($this->batch) ? $this->batch : FALSE, + ); + if (db_result(db_query_range("SELECT 1 FROM {feeds_source} WHERE id = '%s' AND feed_nid = %d", $this->id, $this->feed_nid, 0, 1))) { + drupal_write_record('feeds_source', $object, array('id', 'feed_nid')); + } + else { + drupal_write_record('feeds_source', $object); + } + } + + /** + * Load configuration and unpack. + * + * @todo Patch CTools to move constants from export.inc to ctools.module. + */ + public function load() { + 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($record->config); + $this->batch = unserialize($record->batch); + } + } + + /** + * Delete configuration. Removes configuration information + * from database, does not delete configuration itself. + */ + public function delete() { + // Alert implementers of FeedsSourceInterface to the fact that we're + // deleting. + foreach ($this->importer->plugin_types as $type) { + $this->importer->$type->sourceDelete($this); + } + db_query("DELETE FROM {feeds_source} WHERE id = '%s' AND feed_nid = %d", $this->id, $this->feed_nid); + } + + /** + * Convenience function. Returns the configuration for a specific class. + * + * @param FeedsSourceInterface $client + * An object that is an implementer of FeedsSourceInterface. + * + * @return + * An array stored for $client. + */ + public function getConfigFor(FeedsSourceInterface $client) { + return $this->config[get_class($client)]; + } + + /** + * Return defaults for feed configuration. + */ + public function configDefaults() { + // Collect information from plugins. + $defaults = array(); + foreach ($this->importer->plugin_types as $type) { + if ($this->importer->$type->hasSourceConfig()) { + $defaults[get_class($this->importer->$type)] = $this->importer->$type->sourceDefaults(); + } + } + return $defaults; + } + + /** + * Override parent::configForm(). + */ + public function configForm(&$form_state) { + // Collect information from plugins. + $form = array(); + foreach ($this->importer->plugin_types as $type) { + if ($this->importer->$type->hasSourceConfig()) { + $class = get_class($this->importer->$type); + $form[$class] = $this->importer->$type->sourceForm($this->config[$class]); + $form[$class]['#tree'] = TRUE; + } + } + return $form; + } + + /** + * Override parent::configFormValidate(). + */ + public function configFormValidate(&$values) { + foreach ($this->importer->plugin_types as $type) { + $class = get_class($this->importer->$type); + if (isset($values[$class]) && $this->importer->$type->hasSourceConfig()) { + $this->importer->$type->sourceFormValidate($values[$class]); + } + } + } +} diff -urN ../feeds.orig/plugins/FeedsCSVParser.inc ./plugins/FeedsCSVParser.inc --- ../feeds.orig/plugins/FeedsCSVParser.inc 2010-07-11 03:36:39.000000000 +0200 +++ ./plugins/FeedsCSVParser.inc 2010-07-19 14:54:38.000000000 +0200 @@ -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); } diff -urN ../feeds.orig/plugins/FeedsDataProcessor.inc ./plugins/FeedsDataProcessor.inc --- ../feeds.orig/plugins/FeedsDataProcessor.inc 2010-07-13 21:32:06.000000000 +0200 +++ ./plugins/FeedsDataProcessor.inc 2010-07-19 14:54:38.000000000 +0200 @@ -60,8 +60,6 @@ else { drupal_set_message(t('There are no new items.')); } - - return FEEDS_BATCH_COMPLETE; } /** @@ -75,7 +73,6 @@ ); $num = $this->handler()->delete($clause); drupal_set_message(format_plural($num, 'Deleted @number item.', 'Deleted @number items.', array('@number' => $num))); - return FEEDS_BATCH_COMPLETE; } /** diff -urN ../feeds.orig/plugins/FeedsFeedNodeProcessor.inc ./plugins/FeedsFeedNodeProcessor.inc --- ../feeds.orig/plugins/FeedsFeedNodeProcessor.inc 2010-07-13 21:32:06.000000000 +0200 +++ ./plugins/FeedsFeedNodeProcessor.inc 2010-07-19 14:54:38.000000000 +0200 @@ -53,8 +53,6 @@ else { drupal_set_message(t('There is no new content.')); } - - return FEEDS_BATCH_COMPLETE; } /** diff -urN ../feeds.orig/plugins/FeedsNodeProcessor.inc ./plugins/FeedsNodeProcessor.inc --- ../feeds.orig/plugins/FeedsNodeProcessor.inc 2010-07-18 20:33:43.000000000 +0200 +++ ./plugins/FeedsNodeProcessor.inc 2010-07-19 16:33:52.000000000 +0200 @@ -23,67 +23,43 @@ /** * Implementation of FeedsProcessor::process(). */ - public function process(FeedsImportBatch $batch, FeedsSource $source) { - - // Keep track of processed items in this pass. - $processed = 0; - - while ($item = $batch->shiftItem()) { - - // Create/update if item does not exist or update existing is enabled. - if (!($nid = $this->existingItemId($item, $source)) || ($this->config['update_existing'] != FEEDS_SKIP_EXISTING)) { - $node = $this->buildNode($nid, $source->feed_nid); - - // Only proceed if item has actually changed. - $hash = $this->hash($item); - if (!empty($nid) && $hash == $this->getHash($nid)) { - continue; - } - $node->feeds_node_item->hash = $hash; + public function process(array $item, FeedsSource $source) { + // Create/update if item does not exist or update existing is enabled. + if (!($nid = $this->existingItemId($item, $source)) || ($this->config['update_existing'] != FEEDS_SKIP_EXISTING)) { + $node = $this->buildNode($nid, $source->feed_nid); + + // Only proceed if item has actually changed. + $hash = $this->hash($item); + if (!empty($nid) && $hash == $this->getHash($nid)) { + return; + } + $node->feeds_node_item->hash = $hash; - // Map and save node. If errors occur don't stop but report them. - try { - $this->map($item, $node); - node_save($node); - if (!empty($nid)) { - $batch->updated++; - } - else { - $batch->created++; - } + // Map and save node. If errors occur don't stop but report them. + try { + $this->map($item, $node); + node_save($node); + if (!empty($nid)) { + return FeedsProcessor::FEEDS_PROCESSOR_UPDATED; } - catch (Exception $e) { - drupal_set_message($e->getMessage(), 'warning'); - watchdog('feeds', $e->getMessage(), array(), WATCHDOG_WARNING); + else { + return FeedsProcessor::FEEDS_PROCESSOR_CREATED; } } - - $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. + catch (Exception $e) { + drupal_set_message($e->getMessage(), 'warning'); + watchdog('feeds', $e->getMessage(), array(), WATCHDOG_WARNING); } } - - // 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.')); - } - - 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)) { @@ -91,7 +67,8 @@ $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 +79,7 @@ else { drupal_set_message(t('There is no content to be deleted.')); } - return FEEDS_BATCH_COMPLETE; + $batch->setProgress(FEEDS_CLEARING, FEEDS_BATCH_COMPLETE); } /** diff -urN ../feeds.orig/plugins/FeedsProcessor.inc ./plugins/FeedsProcessor.inc --- ../feeds.orig/plugins/FeedsProcessor.inc 2010-07-18 20:33:43.000000000 +0200 +++ ./plugins/FeedsProcessor.inc 2010-07-19 16:27:30.000000000 +0200 @@ -13,19 +13,24 @@ abstract class FeedsProcessor extends FeedsPlugin { /** - * Process the result of the parser or previous processors. - * Extending classes must implement this method. + * Process function create an item. + */ + const FEEDS_PROCESSOR_CREATED = 1; + + /** + * Process function updated an item. + */ + const FEEDS_PROCESSOR_UPDATED = 2; + + /** + * Process the given item.. * - * @param FeedsImportBatch $batch - * The current feed import data passed in from the parsing stage. + * @param array $item + * Item to process. * @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); + public abstract function process(array $item, FeedsSource $source); /** * Remove all stored results or stored results up to a certain time for this @@ -40,10 +45,6 @@ * 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); diff -urN ../feeds.orig/plugins/FeedsProcessor.inc.orig ./plugins/FeedsProcessor.inc.orig --- ../feeds.orig/plugins/FeedsProcessor.inc.orig 1970-01-01 01:00:00.000000000 +0100 +++ ./plugins/FeedsProcessor.inc.orig 2010-07-19 14:54:38.000000000 +0200 @@ -0,0 +1,301 @@ +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); + + /** + * Delete feed items younger than now - $time. Do not invoke expire on a + * processor directly, but use FeedsImporter::expire() instead. + * + * @see FeedsImporter::expire(). + * @see FeedsDataProcessor::expire(). + * + * @param $time + * If implemented, all items produced by this configuration that are older + * than FEEDS_REQUEST_TIME - $time should be deleted. + * If $time === NULL processor should use internal configuration. + * + * @return + * FEEDS_BATCH_COMPLETE if all items have been processed, a float between 0 + * and 0.99* indicating progress otherwise. + */ + public function expire($time = NULL) { + return FEEDS_BATCH_COMPLETE; + } + + /** + * Execute mapping on an item. + * + * This method encapsulates the central mapping functionality. When an item is + * processed, it is passed through map() where the properties of $source_item + * are mapped onto $target_item following the processor's mapping + * configuration. + * + * For each mapping FeedsParser::getSourceElement() is executed to retrieve + * the source element, then FeedsProcessor::setTargetElement() is invoked + * to populate the target item properly. Alternatively a + * hook_x_targets_alter() may have specified a callback for a mapping target + * in which case the callback is asked to populate the target item instead of + * FeedsProcessor::setTargetElement(). + * + * @ingroup mappingapi + * + * @see hook_feeds_data_processor_targets_alter() + * @see hook_feeds_node_processor_targets_alter() + * @see hook_feeds_term_processor_targets_alter() + * @see hook_feeds_user_processor_targets_alter() + */ + protected function map($source_item, $target_item = NULL) { + + // Static cache $targets as getMappingTargets() may be an expensive method. + static $targets; + if (!isset($targets[$this->id])) { + $targets[$this->id] = $this->getMappingTargets(); + } + $parser = feeds_importer($this->id)->parser; + if (empty($target_item)) { + $target_item = array(); + } + + // Many mappers add to existing fields rather than replacing them. Hence we + // need to clear target elements of each item before mapping in case we are + // mapping on a prepopulated item such as an existing node. + if (is_array($target_item)) { + $target_item = (object)$target_item; + $convert_to_array = TRUE; + } + foreach ($this->config['mappings'] as $mapping) { + if (isset($targets[$mapping['target']]['real_target'])) { + unset($target_item->{$targets[$mapping['target']]['real_target']}); + } + elseif (isset($target_item->{$mapping['target']})) { + unset($target_item->{$mapping['target']}); + } + } + if ($convert_to_array) { + $target_item = (array)$target_item; + } + + /* + This is where the actual mapping happens: For every mapping we envoke + the parser's getSourceElement() method to retrieve the value of the source + element and pass it to the processor's setTargetElement() to stick it + on the right place of the target item. + + If the mapping specifies a callback method, use the callback instead of + setTargetElement(). + */ + foreach ($this->config['mappings'] as $mapping) { + $value = $parser->getSourceElement($source_item, $mapping['source']); + + if (is_array($targets[$this->id][$mapping['target']]) && isset($targets[$this->id][$mapping['target']]['callback']) && function_exists($targets[$this->id][$mapping['target']]['callback'])) { + $callback = $targets[$this->id][$mapping['target']]['callback']; + $callback($target_item, $mapping['target'], $value); + } + else { + $this->setTargetElement($target_item, $mapping['target'], $value); + } + } + return $target_item; + } + + /** + * Per default, don't support expiry. If processor supports expiry of imported + * items, return the time after which items should be removed. + */ + public function expiryTime() { + return FEEDS_EXPIRE_NEVER; + } + + /** + * Declare default configuration. + */ + public function configDefaults() { + return array('mappings' => array()); + } + + /** + * Add a mapping to existing mappings. + * + * @param $source + * A string that identifies a source element. + * @param $target + * A string that identifies a target element. + * @param $unique + * A boolean that defines whether the target value should be unique. If + * TRUE only one item with a given target value can exist on the local + * system. Compare with existingItemId() and uniqueTargets(). + */ + public function addMapping($source, $target, $unique = FALSE) { + if (!empty($source) && !empty($target)) { + $this->config['mappings'][] = array( + 'source' => $source, + 'target' => $target, + 'unique' => $unique, + ); + } + } + + /** + * Set unique state of a mapping target. + */ + public function setUnique($source, $target, $unique) { + if (!empty($source) && !empty($target)) { + foreach ($this->config['mappings'] as $k => $mapping) { + if ($mapping['source'] == $source && $mapping['target'] == $target) { + $this->config['mappings'][$k]['unique'] = $unique; + } + } + } + } + + /** + * Remove a mapping. + */ + public function removeMapping($source, $target) { + foreach ($this->config['mappings'] as $k => $mapping) { + if ($mapping['source'] == $source && $mapping['target'] == $target) { + unset($this->config['mappings'][$k]); + } + } + // Keep or keys clean. + $this->config['mappings'] = array_values($this->config['mappings']); + } + + /** + * Get mappings. + */ + public function getMappings() { + return isset($this->config['mappings']) ? $this->config['mappings'] : array(); + } + + /** + * Declare possible mapping targets that this processor exposes. + * + * @ingroup mappingapi + * + * @return + * An array of mapping targets. Keys are paths to targets + * separated by ->, values are TRUE if target can be unique, + * FALSE otherwise. + */ + public function getMappingTargets() { + return array(); + } + + /** + * Set a concrete target element. Invoked from FeedsProcessor::map(). + * + * @ingroup mappingapi + */ + public function setTargetElement(&$target_item, $target_element, $value) { + $target_item[$target_element] = $value; + } + + /** + * Retrieve the target item's existing id if available. Otherwise return 0. + * + * @ingroup mappingapi + * + * @param $source_item + * A single item that has been aggregated from a feed. + * @param FeedsSource $source + * The source information about this import. + */ + protected function existingItemId($source_item, FeedsSource $source) { + return 0; + } + + /** + * Utility function that iterates over a target array and retrieves all + * sources that are unique. + * + * @param $source_item + * A feed item from a FeedsImportBatch. + * + * @return + * An array where the keys are target field names and the values are the + * elements from the source item mapped to these targets. + */ + protected function uniqueTargets($source_item) { + $parser = feeds_importer($this->id)->parser; + $targets = array(); + foreach ($this->config['mappings'] as $mapping) { + if ($mapping['unique']) { + // Invoke the parser's getSourceElement to retrieve the value for this + // mapping's source. + $targets[$mapping['target']] = $parser->getSourceElement($source_item, $mapping['source']); + } + } + return $targets; + } + + /** + * Loads on-behalf implementations from mappers/ directory. + * + * FeedsProcessor::map() does not load from mappers/ as only node and user + * processor ship with on-behalf implementations. + * + * @see FeedsNodeProcessor::map() + * @see FeedsUserProcessor::map() + */ + protected static function loadMappers() { + static $loaded = FALSE; + if (!$loaded) { + $path = drupal_get_path('module', 'feeds') .'/mappers'; + $files = drupal_system_listing('.*\.inc$', $path, 'name', 0); + foreach ($files as $file) { + if (strstr($file->filename, '/mappers/')) { + require_once("./$file->filename"); + } + } + // Rebuild cache. + module_implements('', FALSE, TRUE); + } + $loaded = TRUE; + } +} diff -urN ../feeds.orig/plugins/FeedsTermProcessor.inc ./plugins/FeedsTermProcessor.inc --- ../feeds.orig/plugins/FeedsTermProcessor.inc 2010-07-13 21:32:06.000000000 +0200 +++ ./plugins/FeedsTermProcessor.inc 2010-07-19 14:54:38.000000000 +0200 @@ -77,8 +77,6 @@ else { drupal_set_message(t('There are no new terms.')); } - - return FEEDS_BATCH_COMPLETE; } /** @@ -103,7 +101,6 @@ else { drupal_set_message(t('No terms to be deleted.')); } - return FEEDS_BATCH_COMPLETE; } /** diff -urN ../feeds.orig/plugins/FeedsUserProcessor.inc ./plugins/FeedsUserProcessor.inc --- ../feeds.orig/plugins/FeedsUserProcessor.inc 2010-07-18 21:47:37.000000000 +0200 +++ ./plugins/FeedsUserProcessor.inc 2010-07-19 14:54:38.000000000 +0200 @@ -80,8 +80,6 @@ else { drupal_set_message(t('There are no new users.')); } - - return FEEDS_BATCH_COMPLETE; } /**