Fake renaming.

src/AsStream.js

@@ -1,150 +0,0 @@
-'use strict';
-
-const {Duplex} = require('stream');
-const defs = require('./defs');
-
-class AsStream extends Duplex {
-
-  constructor(fn, options) {
-    super(Object.assign({}, {writableObjectMode: true, readableObjectMode: true}, options));
-    if (typeof fn == 'function') {
-      this._fn = fn;
-    } else if (fn) {
-      if (typeof fn[Symbol.asyncIterator] == 'function') {
-        this._fn = fn[Symbol.asyncIterator].bind(fn);
-      } else if (typeof fn[Symbol.iterator] == 'function') {
-        this._fn = fn[Symbol.iterator].bind(fn);
-      }
-    }
-    if (!this._fn)
-      throw TypeError(
-        'Only a function or an object with an iterator is accepted as the first argument'
-      );
-
-    // pump variables
-    this._paused = Promise.resolve();
-    this._resolvePaused = null;
-    this._queue = [];
-  }
-
-  _write(chunk, encoding, callback) {
-    this._processChunk(chunk, encoding).then(
-      () => callback(null),
-      error => callback(error)
-    );
-  }
-  _final(callback) {
-    if (!defs.isFlushable(this._fn)) {
-      this.push(null);
-      callback(null);
-      return;
-    }
-    this._processChunk(defs.none, null).then(
-      () => (this.push(null), callback(null)),
-      error => callback(error)
-    );
-  }
-  _read() {
-    this._resume();
-  }
-
-  // pause/resume
-  _resume() {
-    if (!this._resolvePaused) return;
-    this._resolvePaused();
-    this._resolvePaused = null;
-    this._paused = Promise.resolve();
-  }
-  _pause() {
-    if (this._resolvePaused) return;
-    this._paused = new Promise(resolve => (this._resolvePaused = resolve));
-  }
-
-  // data processing
-  _pushResults(values) {
-    if (values && typeof values.next == 'function') {
-      // generator
-      this._queue.push(values);
-      return;
-    }
-    // array
-    this._queue.push(values[Symbol.iterator]());
-  }
-  async _pump() {
-    const queue = this._queue;
-    while (queue.length) {
-      await this._paused;
-      const gen = queue[queue.length - 1];
-      let result = gen.next();
-      if (result && typeof result.then == 'function') {
-        result = await result;
-      }
-      if (result.done) {
-        queue.pop();
-        continue;
-      }
-      const value = result.value;
-      if (value && typeof value.then == 'function') {
-        value = await value;
-      }
-      await this._sanitize(value);
-    }
-  }
-  async _sanitize(value) {
-    if (value === undefined || value === null || value === defs.none) return;
-    if (value === defs.stop) throw new defs.Stop();
-
-    if (defs.isMany(value)) {
-      this._pushResults(defs.getManyValues(value));
-      return this._pump();
-    }
-
-    if (defs.isFinalValue(value)) {
-      // a final value is not supported, it is treated as a regular value
-      value = defs.getFinalValue(value);
-      return this._processValue(value);
-    }
-
-    if (!this.push(value)) {
-      this._pause();
-    }
-  }
-  async _processChunk(chunk, encoding) {
-    try {
-      const value = this._fn(chunk, encoding);
-      await this._processValue(value);
-    } catch (error) {
-      if (error instanceof defs.Stop) {
-        this.push(null);
-        this.destroy();
-        return;
-      }
-      throw error;
-    }
-  }
-  async _processValue(value) {
-    if (value && typeof value.then == 'function') {
-      // thenable
-      return value.then(value => this._processValue(value));
-    }
-    if (value && typeof value.next == 'function') {
-      // generator
-      this._pushResults(value);
-      return this._pump();
-    }
-    return this._sanitize(value);
-  }
-
-  static make(fn, options) {
-    return new AsStream(fn, options);
-  }
-}
-
-AsStream.asStream = AsStream.make;
-AsStream.make.Constructor = AsStream;
-
-module.exports = AsStream;
-
-// to keep ESM happy:
-module.exports.asStream = AsStream.make;
-module.exports.make = AsStream.make;

src/aasStream.js

@@ -0,0 +1,142 @@
+'use strict';
+
+const {Duplex} = require('stream');
+const defs = require('./defs');
+
+const asStream = (fn, options) => {
+  if (fn && typeof fn != 'function') {
+    if (typeof fn[Symbol.asyncIterator] == 'function') {
+      fn = fn[Symbol.asyncIterator].bind(fn);
+    } else if (typeof fn[Symbol.iterator] == 'function') {
+      fn = fn[Symbol.iterator].bind(fn);
+    } else {
+      fn = null;
+    }
+  }
+  if (!fn)
+    throw TypeError(
+      'Only a function or an object with an iterator is accepted as the first argument'
+    );
+
+  // pump variables
+  let paused = Promise.resolve(),
+    resolvePaused = null;
+  const queue = [];
+
+  // pause/resume
+  const resume = () => {
+    if (!resolvePaused) return;
+    resolvePaused();
+    resolvePaused = null;
+    paused = Promise.resolve();
+  };
+  const pause = () => {
+    if (resolvePaused) return;
+    paused = new Promise(resolve => (resolvePaused = resolve));
+  };
+
+  let stream = null; // will be assigned later
+
+  // data processing
+  const pushResults = values => {
+    if (values && typeof values.next == 'function') {
+      // generator
+      queue.push(values);
+      return;
+    }
+    // array
+    queue.push(values[Symbol.iterator]());
+  };
+  const pump = async () => {
+    while (queue.length) {
+      await paused;
+      const gen = queue[queue.length - 1];
+      let result = gen.next();
+      if (result && typeof result.then == 'function') {
+        result = await result;
+      }
+      if (result.done) {
+        queue.pop();
+        continue;
+      }
+      const value = result.value;
+      if (value && typeof value.then == 'function') {
+        value = await value;
+      }
+      await sanitize(value);
+    }
+  };
+  const sanitize = async value => {
+    if (value === undefined || value === null || value === defs.none) return;
+    if (value === defs.stop) throw new defs.Stop();
+
+    if (defs.isMany(value)) {
+      pushResults(defs.getManyValues(value));
+      return pump();
+    }
+
+    if (defs.isFinalValue(value)) {
+      // a final value is not supported, it is treated as a regular value
+      value = defs.getFinalValue(value);
+      return processValue(value);
+    }
+
+    if (!stream.push(value)) {
+      pause();
+    }
+  };
+  const processChunk = async (chunk, encoding) => {
+    try {
+      const value = fn(chunk, encoding);
+      await processValue(value);
+    } catch (error) {
+      if (error instanceof defs.Stop) {
+        stream.push(null);
+        stream.destroy();
+        return;
+      }
+      throw error;
+    }
+  };
+  const processValue = async value => {
+    if (value && typeof value.then == 'function') {
+      // thenable
+      return value.then(value => processValue(value));
+    }
+    if (value && typeof value.next == 'function') {
+      // generator
+      pushResults(value);
+      return pump();
+    }
+    return sanitize(value);
+  };
+
+  stream = new Duplex(
+    Object.assign({}, {writableObjectMode: true, readableObjectMode: true}, options, {
+      write(chunk, encoding, callback) {
+        processChunk(chunk, encoding).then(
+          () => callback(null),
+          error => callback(error)
+        );
+      },
+      final(callback) {
+        if (!defs.isFlushable(fn)) {
+          stream.push(null);
+          callback(null);
+          return;
+        }
+        processChunk(defs.none, null).then(
+          () => (stream.push(null), callback(null)),
+          error => callback(error)
+        );
+      },
+      read() {
+        resume();
+      }
+    })
+  );
+
+  return stream;
+};
+
+module.exports = asStream;