feat: chunked processing (streaming)

Author: ovflowd

bin/commands/generate.mjs

@@ -79,7 +79,7 @@ export default {
       prompt: {
         type: 'text',
         message: 'Items per worker thread',
-        initialValue: '10',
+        initialValue: '20',
       },
     },
     version: {

src/generators.mjs

@@ -1,6 +1,7 @@
 'use strict';
 
 import { allGenerators } from './generators/index.mjs';
+import { isAsyncGenerator, createStreamingCache } from './streaming.mjs';
 import WorkerPool from './threading/index.mjs';
 import createParallelWorker from './threading/parallel.mjs';
 
@@ -26,12 +27,38 @@ const createGenerator = input => {
   /**
    * We store all the registered generators to be processed
    * within a Record, so we can access their results at any time whenever needed
-   * (we store the Promises of the generator outputs)
+   * (we store the Promises of the generator outputs, or AsyncGenerators for streaming)
    *
    * @type {{ [K in keyof AllGenerators]: ReturnType<AllGenerators[K]['generate']> }}
    */
   const cachedGenerators = { ast: Promise.resolve(input) };
 
+  /**
+   * Cache for collected async generator results.
+   * When a streaming generator is first consumed, we collect all chunks
+   * and store the promise here so subsequent consumers share the same result.
+   */
+  const streamingCache = createStreamingCache();
+
+  /**
+   * Gets the dependency input, handling both regular promises and async generators.
+   * For async generators, ensures only one collection happens and result is cached.
+   * @param {string} dependsOn - Name of the dependency generator
+   * @returns {Promise<any>}
+   */
+  const getDependencyInput = async dependsOn => {
+    // First, await the cached promise to get the actual result
+    const result = await cachedGenerators[dependsOn];
+
+    // Check if the result is an async generator (streaming)
+    if (isAsyncGenerator(result)) {
+      return streamingCache.getOrCollect(dependsOn, result);
+    }
+
+    // Regular result - return it directly
+    return result;
+  };
+
   /**
    * Runs the Generator engine with the provided top-level input and the given generator options
    *
@@ -57,18 +84,20 @@ const createGenerator = input => {
 
       // Ensure dependency is scheduled (but don't await its result yet)
       if (dependsOn && !(dependsOn in cachedGenerators)) {
-        await runGenerators({ ...options, generators: [dependsOn] });
+        // Recursively schedule - don't await, just ensure it's in cachedGenerators
+        runGenerators({ ...options, generators: [dependsOn] });
       }
 
       // Create a ParallelWorker for this generator
+      // The worker supports both batch (map) and streaming (stream) modes
       const worker = createParallelWorker(generatorName, chunkPool, options);
 
       /**
        * Schedule the generator - it awaits its dependency internally
-       * his allows multiple generators with the same dependency to run in parallel
+       * This allows multiple generators with the same dependency to run in parallel
        */
       const scheduledGenerator = async () => {
-        const input = await cachedGenerators[dependsOn];
+        const input = await getDependencyInput(dependsOn);
 
         return generate(input, { ...options, worker });
       };

src/generators/api-links/__tests__/fixtures.test.mjs

@@ -26,12 +26,17 @@ describe('api links', () => {
           chunkSize: 10,
         });
 
-        const astJsResult = await astJs.generate(undefined, {
+        // Collect results from the async generator
+        const astJsResults = [];
+
+        for await (const chunk of astJs.generate(undefined, {
           input: [sourceFile],
           worker,
-        });
+        })) {
+          astJsResults.push(...chunk);
+        }
 
-        const actualOutput = await apiLinks.generate(astJsResult, {
+        const actualOutput = await apiLinks.generate(astJsResults, {
           gitRef: 'https://github.com/nodejs/node/tree/HEAD',
         });
 

src/generators/ast-js/index.mjs

@@ -50,15 +50,20 @@ export default {
   },
 
   /**
-   * @param {Input} _
+   * @param {Input} i
    * @param {Partial<GeneratorOptions>} options
+   * @returns {AsyncGenerator<Array<object>>}
    */
-  async generate(_, { input = [], worker }) {
+  async *generate(i, { input = [], worker }) {
     const sourceFiles = globSync(input).filter(
       filePath => extname(filePath) === '.js'
     );
 
+    const deps = { input: sourceFiles };
+
     // Parse the Javascript sources into ASTs in parallel using worker threads
-    return worker.map(sourceFiles, _, { input: sourceFiles });
+    for await (const chunkResult of worker.stream(sourceFiles, i, deps)) {
+      yield chunkResult;
+    }
   },
 };

src/generators/jsx-ast/index.mjs

@@ -1,42 +1,9 @@
-import { OVERRIDDEN_POSITIONS } from './constants.mjs';
 import { buildSideBarProps } from './utils/buildBarProps.mjs';
 import buildContent from './utils/buildContent.mjs';
+import { getSortedHeadNodes } from './utils/getSortedHeadNodes.mjs';
 import { groupNodesByModule } from '../../utils/generators.mjs';
 import { getRemarkRecma } from '../../utils/remark.mjs';
 
-/**
- * Sorts entries by OVERRIDDEN_POSITIONS and then heading name.
- * @param {Array<ApiDocMetadataEntry>} entries
- */
-const getSortedHeadNodes = entries => {
-  /**
-   * Sorts entries by OVERRIDDEN_POSITIONS and then heading name.
-   * @param {ApiDocMetadataEntry} a
-   * @param {ApiDocMetadataEntry} b
-   * @returns {number}
-   */
-  const headingSortFn = (a, b) => {
-    const ai = OVERRIDDEN_POSITIONS.indexOf(a.api);
-    const bi = OVERRIDDEN_POSITIONS.indexOf(b.api);
-
-    if (ai !== -1 && bi !== -1) {
-      return ai - bi;
-    }
-
-    if (ai !== -1) {
-      return -1;
-    }
-
-    if (bi !== -1) {
-      return 1;
-    }
-
-    return a.heading.data.name.localeCompare(b.heading.data.name);
-  };
-
-  return entries.filter(node => node.heading.depth === 1).sort(headingSortFn);
-};
-
 /**
  * Generator for converting MDAST to JSX AST.
  *
@@ -97,11 +64,15 @@ export default {
    *
    * @param {Input} entries
    * @param {Partial<GeneratorOptions>} options
-   * @returns {Promise<Array<string>>} Array of generated content
+   * @returns {AsyncGenerator<Array<string>>}
    */
-  async generate(entries, { index, releases, version, worker }) {
+  async *generate(entries, { index, releases, version, worker }) {
     const headNodes = entries.filter(node => node.heading.depth === 1);
 
-    return worker.map(headNodes, entries, { index, releases, version });
+    const deps = { index, releases, version };
+
+    for await (const chunkResult of worker.stream(headNodes, entries, deps)) {
+      yield chunkResult;
+    }
   },
 };

src/generators/jsx-ast/utils/getSortedHeadNodes.mjs

@@ -0,0 +1,36 @@
+'use strict';
+
+import { OVERRIDDEN_POSITIONS } from '../constants.mjs';
+
+/**
+ * Sorts entries by OVERRIDDEN_POSITIONS and then heading name.
+ * @param {ApiDocMetadataEntry} a
+ * @param {ApiDocMetadataEntry} b
+ * @returns {number}
+ */
+const headingSortFn = (a, b) => {
+  const ai = OVERRIDDEN_POSITIONS.indexOf(a.api);
+  const bi = OVERRIDDEN_POSITIONS.indexOf(b.api);
+
+  if (ai !== -1 && bi !== -1) {
+    return ai - bi;
+  }
+
+  if (ai !== -1) {
+    return -1;
+  }
+
+  if (bi !== -1) {
+    return 1;
+  }
+
+  return a.heading.data.name.localeCompare(b.heading.data.name);
+};
+
+/**
+ * Filters and sorts entries by OVERRIDDEN_POSITIONS and then heading name.
+ * @param {Array<ApiDocMetadataEntry>} entries
+ * @returns {Array<ApiDocMetadataEntry>}
+ */
+export const getSortedHeadNodes = entries =>
+  entries.filter(node => node.heading.depth === 1).sort(headingSortFn);

src/generators/legacy-html-all/index.mjs

@@ -6,7 +6,7 @@ import { join, resolve } from 'node:path';
 import HTMLMinifier from '@minify-html/node';
 
 import { getRemarkRehype } from '../../utils/remark.mjs';
-import dropdowns from '../legacy-html/utils/buildDropdowns.mjs';
+import { replaceTemplateValues } from '../legacy-html/utils/replaceTemplateValues.mjs';
 import tableOfContents from '../legacy-html/utils/tableOfContents.mjs';
 
 /**
@@ -40,67 +40,109 @@ export default {
 
   dependsOn: 'legacy-html',
 
+  /**
+   * Process a chunk of template values from the dependency.
+   * Extracts toc and content from each entry for aggregation.
+   * @param {Input} fullInput
+   * @param {number[]} itemIndices
+   */
+  processChunk(fullInput, itemIndices) {
+    const results = [];
+
+    for (const idx of itemIndices) {
+      const entry = fullInput[idx];
+
+      // Skip the index entry
+      if (entry.api === 'index') {
+        continue;
+      }
+
+      results.push({
+        api: entry.api,
+        section: entry.section,
+        toc: entry.toc,
+        content: entry.content,
+      });
+    }
+
+    return results;
+  },
+
   /**
    * Generates the `all.html` file from the `legacy-html` generator
    * @param {Input} input
    * @param {Partial<GeneratorOptions>} options
+   * @returns {AsyncGenerator<Array<{api: string; section: string; toc: string; content: string}>>}
    */
-  async generate(input, { version, releases, output }) {
-    const inputWithoutIndex = input.filter(entry => entry.api !== 'index');
-
-    // Gets a Remark Processor that parses Markdown to minified HTML
-    const remarkWithRehype = getRemarkRehype();
-
-    // Current directory path relative to the `index.mjs` file
-    // from the `legacy-html` generator, as all the assets are there
-    const baseDir = resolve(import.meta.dirname, '..', 'legacy-html');
-
-    // Reads the API template.html file to be used as a base for the HTML files
-    const apiTemplate = await readFile(join(baseDir, 'template.html'), 'utf-8');
-
-    // Aggregates all individual Table of Contents into one giant string
-    const aggregatedToC = inputWithoutIndex.map(entry => entry.toc).join('\n');
-
-    // Aggregates all individual content into one giant string
-    const aggregatedContent = inputWithoutIndex
-      .map(entry => entry.content)
-      .join('\n');
-
-    // Creates a "mimic" of an `ApiDocMetadataEntry` which fulfils the requirements
-    // for generating the `tableOfContents` with the `tableOfContents.parseNavigationNode` parser
-    const sideNavigationFromValues = inputWithoutIndex.map(entry => ({
-      api: entry.api,
-      heading: { data: { depth: 1, name: entry.section } },
-    }));
-
-    // Generates the global Table of Contents (Sidebar Navigation)
-    const parsedSideNav = remarkWithRehype.processSync(
-      tableOfContents(sideNavigationFromValues, {
-        maxDepth: 1,
-        parser: tableOfContents.parseNavigationNode,
-      })
-    );
-
-    const generatedAllTemplate = apiTemplate
-      .replace('__ID__', 'all')
-      .replace(/__FILENAME__/g, 'all')
-      .replace('__SECTION__', 'All')
-      .replace(/__VERSION__/g, `v${version.version}`)
-      .replace(/__TOC__/g, tableOfContents.wrapToC(aggregatedToC))
-      .replace(/__GTOC__/g, parsedSideNav)
-      .replace('__CONTENT__', aggregatedContent)
-      .replace(/__TOC_PICKER__/g, dropdowns.buildToC(aggregatedToC))
-      .replace(/__GTOC_PICKER__/g, '')
-      .replace('__ALTDOCS__', dropdowns.buildVersions('all', '', releases))
-      .replace('__EDIT_ON_GITHUB__', '');
-
-    // We minify the html result to reduce the file size and keep it "clean"
-    const minified = HTMLMinifier.minify(Buffer.from(generatedAllTemplate), {});
+  async *generate(input, { version, releases, output, worker }) {
+    // Collect all chunks as they stream in
+    const allChunks = [];
 
+    for await (const chunkResult of worker.stream(input, input, {})) {
+      allChunks.push(...chunkResult);
+
+      yield chunkResult;
+    }
+
+    // After all chunks are collected, build and write the final file
     if (output) {
+      // Gets a Remark Processor that parses Markdown to minified HTML
+      const remarkWithRehype = getRemarkRehype();
+
+      // Current directory path relative to the `index.mjs` file
+      // from the `legacy-html` generator, as all the assets are there
+      const baseDir = resolve(import.meta.dirname, '..', 'legacy-html');
+
+      // Reads the API template.html file to be used as a base for the HTML files
+      const apiTemplate = await readFile(
+        join(baseDir, 'template.html'),
+        'utf-8'
+      );
+
+      // Aggregates all individual Table of Contents into one giant string
+      const aggregatedToC = allChunks.map(entry => entry.toc).join('\n');
+
+      // Aggregates all individual content into one giant string
+      const aggregatedContent = allChunks
+        .map(entry => entry.content)
+        .join('\n');
+
+      // Creates a "mimic" of an `ApiDocMetadataEntry` which fulfils the requirements
+      // for generating the `tableOfContents` with the `tableOfContents.parseNavigationNode` parser
+      const sideNavigationFromValues = allChunks.map(entry => ({
+        api: entry.api,
+        heading: { data: { depth: 1, name: entry.section } },
+      }));
+
+      // Generates the global Table of Contents (Sidebar Navigation)
+      const parsedSideNav = remarkWithRehype.processSync(
+        tableOfContents(sideNavigationFromValues, {
+          maxDepth: 1,
+          parser: tableOfContents.parseNavigationNode,
+        })
+      );
+
+      const templateValues = {
+        api: 'all',
+        added: '',
+        section: 'All',
+        version: `v${version.version}`,
+        toc: aggregatedToC,
+        nav: String(parsedSideNav),
+        content: aggregatedContent,
+      };
+
+      const generatedAllTemplate = replaceTemplateValues(
+        apiTemplate,
+        templateValues,
+        releases,
+        { skipGitHub: true, skipGtocPicker: true }
+      );
+
+      // We minify the html result to reduce the file size and keep it "clean"
+      const minified = HTMLMinifier.minify(Buffer.from(generatedAllTemplate));
+
       await writeFile(join(output, 'all.html'), minified);
     }
-
-    return minified;
   },
 };