Issue #2531858 by twistor: Add a FeedsSource::pushImport() method

includes/FeedsConfigurable.inc

@@ -55,7 +55,9 @@ abstract class FeedsConfigurable {
     if (!strlen($id)) {
       throw new InvalidArgumentException(t('Empty configuration identifier.'));
     }
-    static $instances = array();
+
+    $instances = &drupal_static(__METHOD__, array());
+
     if (!isset($instances[$class][$id])) {
       $instances[$class][$id] = new $class($id);
     }

includes/FeedsSource.inc

@@ -197,7 +197,9 @@ class FeedsSource extends FeedsConfigurable {
    */
   public static function instance($importer_id, $feed_nid) {
     $class = variable_get('feeds_source_class', 'FeedsSource');
-    static $instances = array();
+
+    $instances = &drupal_static(__METHOD__, array());
+
     if (!isset($instances[$class][$importer_id][$feed_nid])) {
       $instances[$class][$importer_id][$feed_nid] = new $class($importer_id, $feed_nid);
     }
@@ -406,23 +408,80 @@ class FeedsSource extends FeedsConfigurable {
       // occurred in hook_feeds_after_import().
       $this->exception = $e;
     }
-    $this->releaseLock();
 
     // Clean up.
     $result = $this->progressImporting();
     if ($result == FEEDS_BATCH_COMPLETE || isset($e)) {
       $this->imported = time();
-      $this->log('import', 'Imported in !s s', array('!s' => $this->imported - $this->state[FEEDS_START]), WATCHDOG_INFO);
+      $this->log('import', 'Imported in @s seconds.', array('@s' => $this->imported - $this->state[FEEDS_START]), WATCHDOG_INFO);
       module_invoke_all('feeds_after_import', $this);
       unset($this->fetcher_result, $this->state);
     }
     $this->save();
+
+    $this->releaseLock();
+
     if (isset($e)) {
       throw $e;
     }
+
     return $result;
   }
 
+  /**
+   * Imports a fetcher result all at once in memory.
+   *
+   * @param FeedsFetcherResult $fetcher_result
+   *   The fetcher result to process.
+   *
+   * @throws Exception
+   *   Thrown if an error occurs when importing.
+   */
+  public function pushImport(FeedsFetcherResult $fetcher_result) {
+    // Since locks only work during a request, check if an import is active.
+    if (!empty($this->fetcher_result) || !empty($this->state)) {
+      throw new RuntimeException('The feed is currently importing.');
+    }
+
+    $this->acquireLock();
+    $start = time();
+
+    try {
+      module_invoke_all('feeds_before_import', $this);
+
+      // Parse.
+      do {
+        $parser_result = $this->importer->parser->parse($this, $fetcher_result);
+        module_invoke_all('feeds_after_parse', $this, $parser_result);
+
+        // Process.
+        $this->importer->processor->process($this, $parser_result);
+
+      } while ($this->progressParsing() !== FEEDS_BATCH_COMPLETE);
+    }
+    catch (Exception $e) {
+      // $e is stored and re-thrown once we've had a chance to log our progress.
+      // Set the exception so that other modules can check if an exception
+      // occurred in hook_feeds_after_import().
+      $this->exception = $e;
+    }
+
+    module_invoke_all('feeds_after_import', $this);
+
+    $this->imported = time();
+    $this->log('import', 'Imported in @s seconds.', array('@s' => $this->imported - $start), WATCHDOG_INFO);
+
+    unset($this->fetcher_result, $this->state);
+
+    $this->save();
+
+    $this->releaseLock();
+
+    if (isset($e)) {
+      throw $e;
+    }
+  }
+
   /**
    * Remove all items from a feed.
    *

plugins/FeedsPlugin.inc

@@ -53,7 +53,7 @@ abstract class FeedsPlugin extends FeedsConfigurable implements FeedsSourceInter
       throw new InvalidArgumentException(t('Empty configuration identifier.'));
     }
 
-    static $instances = array();
+    $instances = &drupal_static(__METHOD__, array());
 
     if (!isset($instances[$class][$id])) {
       $instance = new $class($id);

tests/feeds_processor_node.test

@@ -12,8 +12,8 @@ class FeedsRSStoNodesTest extends FeedsWebTestCase {
 
   public static function getInfo() {
     return array(
-      'name' => 'RSS import to nodes',
-      'description' => 'Tests a feed configuration that is attached to a content type, uses HTTP fetcher, common syndication parser and a node processor. Repeats the same test for an importer configuration that is not attached to a content type and for a configuration that is attached to a content type and uses the file fetcher.',
+      'name' => 'Processor: Node',
+      'description' => 'Tests for the node processor.',
       'group' => 'Feeds',
     );
   }
@@ -577,8 +577,6 @@ class FeedsRSStoNodesTest extends FeedsWebTestCase {
 
     // The feed should still be scheduled because it is being processed.
     // @see https://drupal.org/node/2275893
-    feeds_source('syndication', 0)->scheduleImport();
-
     $this->cronRun();
     $this->assertEqual(86, db_query("SELECT COUNT(*) FROM {node}")->fetchField());
   }
@@ -658,4 +656,16 @@ class FeedsRSStoNodesTest extends FeedsWebTestCase {
     $this->assertNoText('Updated 2 nodes.');
   }
 
+  /**
+   * Tests the FeedsSource::pushImport() method.
+   */
+  public function testPushImport() {
+    // Attach to standalone importer.
+    $this->setSettings('syndication', NULL, array('content_type' => ''));
+
+    $raw = file_get_contents(dirname(__FILE__) . '/feeds/developmentseed.rss2');
+    feeds_source('syndication', 0)->pushImport(new FeedsFetcherResult($raw));
+    $this->assertEqual(10, db_query("SELECT COUNT(*) FROM {node}")->fetchField());
+  }
+
 }