[WIP] Streaming indexing of the import file

e2e/playground.blueprint.json

@@ -5,6 +5,11 @@
         "php": "8.3",
         "wp": "latest"
     },
+    "constants": {
+        "WP_DEBUG": true,
+        "WP_DEBUG_LOG": true,
+        "WP_DEBUG_DISPLAY": true
+    },
     "steps": [
         {
             "step": "login"

src/class-wp-import.php

@@ -6,6 +6,9 @@
  * @subpackage Importer
  */
 
+use WordPress\ByteStream\ReadStream\FileReadStream;
+use WordPress\DataLiberation\BlockMarkup\BlockMarkupUrlProcessor;
+use WordPress\DataLiberation\Importer\StreamImporter;
 use WordPress\DataLiberation\URL\WPURL;
 use function WordPress\DataLiberation\URL\wp_rewrite_urls;
 
@@ -65,8 +68,10 @@ public function dispatch() {
 				break;
 			case 1:
 				check_admin_referer( 'import-upload' );
-				if ( $this->handle_upload() ) {
-					$this->import_options();
+				$file = $this->handle_upload();
+				if ( $file ) {
+					$this->index_uploaded_import_file( $file );
+					$this->display_import_options_form();
 				}
 				break;
 			case 2:
@@ -242,24 +247,77 @@ public function handle_upload() {
 			return false;
 		}
 
-		$this->id    = (int) $file['id'];
-		$import_data = $this->parse( $file['file'] );
-		if ( is_wp_error( $import_data ) ) {
-			/** @var WP_Error $import_error */
-			$import_error = $import_data;
-			echo '<p><strong>' . __( 'Sorry, there has been an error.', 'wordpress-importer' ) . '</strong><br />';
-			echo esc_html( $import_error->get_error_message() ) . '</p>';
+		return $file;
+	}
+
+	public function index_uploaded_import_file( $file ) {
+		$this->id        = (int) $file['id'];
+		$stream_importer = StreamImporter::create(
+			function ( $cursor = null ) use ( $file ) {
+				return WXR_Parser_XML_Processor::create_wxr_entity_reader( $file['file'], $cursor );
+			}
+		);
+
+		// We're at STAGE_INITIAL, let's advance to STAGE_INDEX_ENTITIES
+		$stream_importer->next_step();
+		$stream_importer->advance_to_next_stage();
+		if ( $stream_importer->get_stage() !== StreamImporter::STAGE_INDEX_ENTITIES ) {
+			_doing_it_wrong( __METHOD__, 'StreamImporter is not at STAGE_INDEX_ENTITIES', '1.0' );
 			return false;
 		}
 
-		$this->version = $import_data['version'];
-		if ( $this->version > $this->max_wxr_version ) {
-			echo '<div class="error"><p><strong>';
-			printf( __( 'This WXR file (version %s) may not be supported by this version of the importer. Please consider updating.', 'wordpress-importer' ), esc_html( $import_data['version'] ) );
-			echo '</strong></p></div>';
-		}
+		// @TODO: Pause before the 30 seconds timeout, display some nice info to the user, resume
+		//        on the next page load.
+		while ( $stream_importer->next_step() ) {
+			$entity = $stream_importer->get_current_entity();
+			if ( ! $entity ) {
+				continue;
+			}
 
-		$this->get_authors_from_import( $import_data );
+			$data = $entity->get_data();
+			switch ( $entity->get_type() ) {
+				case 'wxr_version':
+					$wxr_version = $data['wxr_version'];
+					// @TODO: Store this across resumed sessions without relying on memory
+					$this->version = $wxr_version;
+					if ( $wxr_version > $this->max_wxr_version ) {
+						echo '<div class="error"><p><strong>';
+						printf( __( 'This WXR file (version %s) may not be supported by this version of the importer. Please consider updating.', 'wordpress-importer' ), esc_html( $wxr_version ) );
+						echo '</strong></p></div>';
+					}
+					break;
+				case 'user':
+					$key                   = isset( $data['author_login'] ) ? $data['author_login'] : (
+						isset( $data['author_email'] ) ? $data['author_email'] : (
+							isset( $data['author_id'] ) ? $data['author_id'] : count( $this->authors )
+						)
+					);
+					$this->authors[ $key ] = $data;
+					// @TODO: Record the author without relying on memory, e.g.:
+					// $this->import_session->record_author( $key, $data );
+					break;
+				case 'post':
+					if ( empty( $data['post_author'] ) ) {
+						printf( __( 'Failed to import author %s. Their posts will be attributed to the current user.', 'wordpress-importer' ), esc_html( $data['post_author'] ?? '' ) );
+						echo '<br />';
+						continue 2;
+					}
+
+					$login = sanitize_user( $data['post_author'], true );
+					if ( ! isset( $this->authors[ $login ] ) ) {
+						$this->authors[ $login ] = array(
+							'author_login'        => $login,
+							'author_display_name' => $data['post_author'],
+						);
+					}
+					break;
+			}
+
+			// @TODO: Should we record referenced domains and ask for per-domain mapping?
+			// $stream_importer->get_indexed_assets_urls();
+
+			// @TODO: Store the cursor in the database for resuming.
+		}
 
 		return true;
 	}
@@ -299,7 +357,7 @@ public function get_authors_from_import( $import_data ) {
 	 * Display pre-import options, author importing/mapping and option to
 	 * fetch attachments
 	 */
-	public function import_options() {
+	public function display_import_options_form() {
 		$j = 0;
 		// phpcs:disable Generic.WhiteSpace.ScopeIndent.Incorrect
 		?>

src/parsers/class-wxr-parser-xml-processor.php

@@ -43,7 +43,7 @@ public function parse( $file ) {
 		$wxr_version         = '';
 
 		try {
-			$reader = $this->create_wxr_entity_reader( $file );
+			$reader = self::create_wxr_entity_reader( $file );
 			// Parse the XML document
 			$last_term = null;
 			while ( $reader->next_entity() ) {
@@ -202,7 +202,7 @@ public function parse( $file ) {
 		);
 	}
 
-	private function create_wxr_entity_reader( $file ) {
+	public static function create_wxr_entity_reader( $file ) {
 		// Every XML element is a combination of a long-form namespace and a
 		// local element name, e.g. a syntax <wp:post_id> could actually refer
 		// to a (https://wordpress.org/export/1.0/, post_id) element.

src/php-toolkit/DataLiberation/EntityReader/class-entityreaderiterator.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace WordPress\DataLiberation\EntityReader;
+
+use Iterator;
+use ReturnTypeWillChange;
+use WordPress\DataLiberation\DataLiberationException;
+
+/**
+ * An iterator that reads entities from a WP_Entity_Reader.
+ */
+class EntityReaderIterator implements Iterator {
+
+	/**
+	 * @var EntityReader
+	 */
+	private $entity_reader;
+	private $is_initialized = false;
+	private $key            = 0;
+
+	public function __construct( EntityReader $entity_reader ) {
+		$this->entity_reader = $entity_reader;
+	}
+
+	public function get_entity_reader() {
+		return $this->entity_reader;
+	}
+
+	#[ReturnTypeWillChange]
+	public function current() {
+		$this->ensure_initialized();
+
+		return $this->entity_reader->get_entity();
+	}
+
+	#[ReturnTypeWillChange]
+	public function next() {
+		$this->ensure_initialized();
+		$this->advance_to_next_entity();
+	}
+
+	#[ReturnTypeWillChange]
+	public function key() {
+		$this->ensure_initialized();
+
+		return $this->key;
+	}
+
+	#[ReturnTypeWillChange]
+	public function valid() {
+		$this->ensure_initialized();
+		if ( $this->entity_reader->is_finished() ) {
+			return false;
+		}
+		// @TODO: Remove these checks once we figure out why.
+		// WXREntityReader says next_entity() succeeds.
+		// one time once the data stream is exhausted.
+		$entity = $this->entity_reader->get_entity();
+		if ( ! $entity ) {
+			return false;
+		}
+		if ( ! $entity->get_type() ) {
+			return false;
+		}
+
+		return true;
+	}
+
+	#[ReturnTypeWillChange]
+	public function rewind() {
+		// rewind is not supported except for the first rewind call that initializes the iterator.
+		if ( $this->is_initialized ) {
+			throw new DataLiberationException( 'EntityReaderIterator does not support rewinding.' );
+		}
+		$this->is_initialized = true;
+		$this->advance_to_next_entity();
+	}
+
+	private function ensure_initialized() {
+		if ( ! $this->is_initialized ) {
+			$this->is_initialized = true;
+			$this->advance_to_next_entity();
+		}
+	}
+
+	private function advance_to_next_entity() {
+		if ( $this->entity_reader->next_entity() ) {
+			++$this->key;
+		}
+	}
+
+	public function get_reentrancy_cursor() {
+		return $this->entity_reader->get_reentrancy_cursor();
+	}
+}

src/php-toolkit/DataLiberation/EntityReader/class-wxrentityreader.php

@@ -941,6 +941,13 @@ private function emit_entity() {
 		}
 		$this->entity_finished = true;
 		++$this->entities_read_so_far;
+
+		// Trim all the XML whitespace.
+		foreach ( $this->entity_data as $k => $v ) {
+			if ( is_string( $v ) ) {
+				$this->entity_data[ $k ] = trim( $v );
+			}
+		}
 	}
 
 	/**