feat!: drop streams support COMPASS-7124 (#208)

We don't really need streams usage anymore and can just drop it. That's
an easy way to ensure that we don't have any dependencies that require
Node.js-specific builtins or polyfills of those.

Author: addaleax

.github/workflows/unit-tests.yml

@@ -21,6 +21,8 @@ jobs:
         run: npm ci
       - name: Check
         run: npm run check
+      - name: Build
+        run: npm run build
       - name: Test
         run: npm test
       - name: Coverage

examples/parse-from-file.ts

@@ -4,8 +4,7 @@ import { pipeline as callbackPipeline, PassThrough, Transform } from 'stream';
 import path from 'path';
 import fs from 'fs';
 import { promisify } from 'util';
-
-import stream from '../src/stream';
+import { parseSchema } from '../src';
 
 const schemaFileName = path.join(__dirname, './fanclub.json');
 
@@ -44,12 +43,10 @@ async function parseFromFile(fileName: string) {
   });
 
   const dest = new PassThrough({ objectMode: true });
+  const resultPromise = parseSchema(dest);
   const pipeline = promisify(callbackPipeline);
-  await pipeline(fileReadStream, createFileStreamLineParser(), stream(), dest);
-  let res;
-  for await (const result of dest) {
-    res = result;
-  }
+  await pipeline(fileReadStream, createFileStreamLineParser(), dest);
+  const res = await resultPromise;
 
   const dur = Date.now() - startTime;
   console.log(res);

src/index.ts

@@ -1,9 +1,3 @@
-import type { AggregationCursor, Document, FindCursor } from 'mongodb';
-import { Readable, PassThrough } from 'stream';
-import { pipeline } from 'stream/promises';
-
-import stream from './stream';
-import type { ParseStreamOptions } from './stream';
 import { SchemaAnalyzer } from './schema-analyzer';
 import type {
   ArraySchemaType,
@@ -24,72 +18,55 @@ import type {
 } from './schema-analyzer';
 import * as schemaStats from './stats';
 
-type MongoDBCursor = AggregationCursor | FindCursor;
+type AnyIterable<T = any> = Iterable<T> | AsyncIterable<T>;
 
-function getStreamSource(
-  source: Document[] | MongoDBCursor | Readable
-): Readable {
-  let streamSource: Readable;
-  if ('stream' in source) {
-    // MongoDB Cursor.
-    streamSource = source.stream();
-  } else if ('pipe' in source) {
-    // Document stream.
-    streamSource = source;
-  } else if (Array.isArray(source)) {
-    // Array of documents.
-    streamSource = Readable.from(source);
-  } else {
+function verifyStreamSource(
+  source: AnyIterable
+): AnyIterable {
+  if (!(Symbol.iterator in source) && !(Symbol.asyncIterator in source)) {
     throw new Error(
       'Unknown input type for `docs`. Must be an array, ' +
         'stream or MongoDB Cursor.'
     );
   }
 
-  return streamSource;
+  return source;
 }
 
-async function schemaStream(
-  source: Document[] | MongoDBCursor | Readable,
-  options?: ParseStreamOptions
-) {
-  const streamSource = getStreamSource(source);
-
-  const dest = new PassThrough({ objectMode: true });
-  await pipeline(streamSource, stream(options), dest);
-  for await (const result of dest) {
-    return result;
+async function getCompletedSchemaAnalyzer(
+  source: AnyIterable,
+  options?: SchemaParseOptions
+): Promise<SchemaAnalyzer> {
+  const analyzer = new SchemaAnalyzer(options);
+  for await (const doc of verifyStreamSource(source)) {
+    analyzer.analyzeDoc(doc);
   }
-  throw new Error('unreachable'); // `dest` always emits exactly one doc.
+  return analyzer;
 }
 
 /**
  * Convenience shortcut for parsing schemas. Accepts an array, stream or
  * MongoDB cursor object to parse documents` from.
  */
 async function parseSchema(
-  source: Document[] | MongoDBCursor | Readable,
+  source: AnyIterable,
   options?: SchemaParseOptions
 ): Promise<Schema> {
-  return await schemaStream(source, options);
+  return (await getCompletedSchemaAnalyzer(source, options)).getResult();
 }
 
 // Convenience shortcut for getting schema paths.
 async function getSchemaPaths(
-  source: Document[] | MongoDBCursor | Readable
+  source: AnyIterable
 ): Promise<string[][]> {
-  return await schemaStream(source, {
-    schemaPaths: true
-  });
+  return (await getCompletedSchemaAnalyzer(source)).getSchemaPaths();
 }
 
 // Convenience shortcut for getting the simplified schema.
 async function getSimplifiedSchema(
-  source: Document[] | MongoDBCursor | Readable
+  source: AnyIterable
 ): Promise<SimplifiedSchema> {
-  return await schemaStream(source, {
-    simplifiedSchema: true
-  });
+  return (await getCompletedSchemaAnalyzer(source)).getSimplifiedSchema();
 }
 
 export default parseSchema;
@@ -113,8 +90,6 @@ export type {
 };
 
 export {
-  stream,
-  getStreamSource,
   parseSchema,
   getSchemaPaths,
   getSimplifiedSchema,

src/stream.ts

@@ -0,0 +1,72 @@
+import assert from 'assert';
+import vm from 'vm';
+import fs from 'fs';
+import path from 'path';
+
+function createMockModuleSystem() {
+  const context = vm.createContext(Object.create(null));
+  class Module {
+    exports = {};
+  }
+  const modules = new Map<string, Module>();
+  // Tiny (incomplete) CommonJS module system mock
+  function makeRequire(basename: string) {
+    return function require(identifier: string): any {
+      if (!identifier.startsWith('./') && !identifier.startsWith('../') && !path.isAbsolute(identifier)) {
+        let current = path.dirname(basename);
+        let previous: string;
+        do {
+          const nodeModulesEntry = path.resolve(current, 'node_modules', identifier);
+          previous = current;
+          current = path.dirname(current);
+          if (fs.existsSync(nodeModulesEntry)) {
+            return require(nodeModulesEntry);
+          }
+        } while (previous !== current);
+        throw new Error(`mock require() does not support Node.js built-ins (${identifier})`);
+      }
+      let file = path.resolve(path.dirname(basename), identifier);
+      if (!fs.existsSync(file) && fs.existsSync(`${file}.js`)) {
+        file = `${file}.js`;
+      } else if (fs.statSync(file).isDirectory()) {
+        if (fs.existsSync(`${file}/package.json`)) {
+          const pjson = JSON.parse(fs.readFileSync(`${file}/package.json`, 'utf8'));
+          file = path.resolve(file, pjson.main || 'index.js');
+        } else if (fs.existsSync(`${file}/index.js`)) {
+          file = path.resolve(file, 'index.js');
+        }
+      }
+      const existing = modules.get(file);
+      if (existing) {
+        return existing.exports;
+      }
+      const module = new Module();
+      const source = fs.readFileSync(file);
+      vm.runInContext(`(function(require, module, exports, __filename, __dirname) {\n${source}\n})`, context)(
+        makeRequire(file), module, module.exports, file, path.dirname(file)
+      );
+      modules.set(file, module);
+      return module.exports;
+    };
+  }
+  return makeRequire;
+}
+
+describe('getSchema should work in plain JS environment without Node.js or browser dependencies', function() {
+  const docs = [
+    { foo: 'bar' },
+    { country: 'Croatia' },
+    { country: 'Croatia' },
+    { country: 'England' }
+  ];
+
+  it('Check if return value is a promise', async function() {
+    const makeRequire = createMockModuleSystem();
+
+    const { parseSchema } = makeRequire(__filename)('../lib/index.js') as typeof import('..');
+
+    const result = await parseSchema(docs);
+    assert.strictEqual(result.count, 4);
+    assert.strictEqual(result.fields.map(f => f.name).join(','), 'country,foo');
+  });
+});