Review.

Author: RubenVerborgh

asynciterator.ts

@@ -25,6 +25,7 @@ export function setTaskScheduler(scheduler: TaskScheduler): void {
   taskScheduler = scheduler;
 }
 
+// Returns a function that calls `fn` with `self` as `this` pointer. */
 function bind<T extends Function>(fn: T, self?: object): T {
   return self ? fn.bind(self) : fn;
 }
@@ -537,8 +538,6 @@ export class AsyncIterator<T> extends EventEmitter {
   */
   range(start: number, end: number): AsyncIterator<T> {
     return this.transform({ offset: start, limit: Math.max(end - start + 1, 0) });
-    // TODO: Implement as following when .take is optimised
-    // return this.skip(start).take(Math.max(end - start + 1, 0));
   }
 
   /**
@@ -582,7 +581,8 @@ function addSingleListener(source: EventEmitter, eventName: string,
     source.on(eventName, listener);
 }
 
-function _validateSource<S>(source?: AsyncIterator<S>, allowDestination = false) {
+// Validates an AsyncIterator for use as a source within another AsyncIterator
+function ensureSourceAvailable<S>(source?: AsyncIterator<S>, allowDestination = false) {
   if (!source || !isFunction(source.read) || !isFunction(source.on))
     throw new Error(`Invalid source: ${source}`);
   if (!allowDestination && (source as any)._destination)
@@ -798,65 +798,66 @@ export class IntegerIterator extends AsyncIterator<number> {
 export type MapFunction<S, D = S> = (item: S) => D | null;
 
 /**
- An iterator that calls a synchronous mapping function
- on every item from its source iterator.
+ * Function that maps an element to itself.
+ */
+export function identity<S>(item: S): typeof item {
+  return item;
+}
+
+/**
+ An iterator that synchronously transforms every item from its source
+ by applying a mapping function.
  @extends module:asynciterator.AsyncIterator
 */
 export class MappingIterator<S, D = S> extends AsyncIterator<D> {
-  private readonly _destroySource: boolean;
-  protected _source: InternalSource<S>;
+  protected readonly _map: MapFunction<S, D>;
+  protected readonly _source: InternalSource<S>;
+  protected readonly _destroySource: boolean;
 
   /**
    * Applies the given mapping to the source iterator.
    */
   constructor(
-    _source: AsyncIterator<S>,
-    private _map: MapFunction<S, D>,
+    source: AsyncIterator<S>,
+    map: MapFunction<S, D> = identity as MapFunction<S, D>,
     options: SourcedIteratorOptions = {}
   ) {
     super();
-    this._source = _validateSource(_source);
+    this._map = map;
+    this._source = ensureSourceAvailable(source);
     this._destroySource = options.destroySource !== false;
 
-    if (_source.done) {
+    // Close if the source is already empty
+    if (source.done) {
       this.close();
     }
+    // Otherwise, wire up the source for reading
     else {
       this._source._destination = this;
       this._source.on('end', destinationClose);
       this._source.on('error', destinationEmitError);
       this._source.on('readable', destinationSetReadable);
-      // If we are given a source that is already readable
-      // then we need to set the state of this iterable to readable
-      // also as there is no guarantee that the is no forthcoming
-      // readable event from the source
       this.readable = this._source.readable;
     }
   }
 
+  /** Tries to read the next item from the iterator. */
   read(): D | null {
-    // Do not read the source if the current iterator is ended
-    if (this.done)
-      return null;
-
-    // A source should only be read from if readable is true
-    if (!this._source.readable) {
+    if (!this.done) {
+      // Try to read an item that maps to a non-null value
+      if (this._source.readable) {
+        let item: S | null, mapped: D | null;
+        while ((item = this._source.read()) !== null) {
+          if ((mapped = this._map(item)) !== null)
+            return mapped;
+        }
+      }
       this.readable = false;
+
+      // Close this iterator if the source is empty
       if (this._source.done)
         this.close();
-      return null;
     }
-
-    let item: S | null;
-    while ((item = this._source.read()) !== null) {
-      const mapped = this._map(item);
-      if (mapped !== null)
-        return mapped;
-    }
-
-    // This will set readable to false on the current iterator
-    // if the source is no longer readable
-    this.readable = false;
     return null;
   }
 
@@ -1247,7 +1248,7 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
   protected _validateSource(source?: AsyncIterator<S>, allowDestination = false): InternalSource<S> {
     if (this._source || typeof this._createSource !== 'undefined')
       throw new Error('The source cannot be changed after it has been set');
-    return _validateSource(source, allowDestination);
+    return ensureSourceAvailable(source, allowDestination);
   }
 
   /**
@@ -1331,10 +1332,7 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
 }
 
 function destinationSetReadable<S>(this: InternalSource<S>) {
-  if (this._destination!.readable)
-    this._destination!.emit('readable');
-  else
-    this._destination!.readable = true;
+  this._destination!.readable = true;
 }
 function destinationEmitError<S>(this: InternalSource<S>, error: Error) {
   this._destination!.emit('error', error);
@@ -2064,10 +2062,9 @@ export interface BufferedIteratorOptions {
   autoStart?: boolean;
 }
 
-export interface TransformIteratorOptions<S> extends BufferedIteratorOptions {
+export interface TransformIteratorOptions<S> extends SourcedIteratorOptions, BufferedIteratorOptions {
   source?: SourceExpression<S>;
   optional?: boolean;
-  destroySource?: boolean;
 }
 
 export interface TransformOptions<S, D> extends TransformIteratorOptions<S> {

package.json

@@ -31,7 +31,7 @@
     "test:microtask": "npm run mocha",
     "test:immediate": "npm run mocha -- --require test/config/useSetImmediate.js",
     "mocha": "c8 mocha",
-    "lint": "eslint asynciterator.ts test",
+    "lint": "eslint asynciterator.ts test perf",
     "docs": "npm run build:module && npm run jsdoc",
     "jsdoc": "jsdoc -c jsdoc.json"
   },

perf/.eslintrc

@@ -0,0 +1,6 @@
+{
+  rules: {
+    no-console: off,
+  },
+}
+

perf/MappingIterator-perf.js

@@ -1,22 +1,27 @@
-import { ArrayIterator, range } from '../dist/asynciterator.js'
+import { ArrayIterator, range } from '../dist/asynciterator.js';
 
-async function perf(warmupIterator, iterator, description) { 
-  return new Promise((res) => {
-    // Then run experiment
-    const now = performance.now()
-    iterator.on('data', () => { }).on('end', () => {
+function noop() {
+  // empty function to drain an iterator
+}
+
+async function perf(warmupIterator, iterator, description) {
+  return new Promise(res => {
+    const now = performance.now();
+    iterator.on('data', noop);
+    iterator.on('end', () => {
       console.log(description, performance.now() - now);
       res();
     });
-  })
+  });
 }
 
 function run(iterator) {
-  return new Promise((res) => {
-    iterator.on('data', () => { }).on('end', () => {
+  return new Promise(res => {
+    iterator.on('data', noop);
+    iterator.on('end', () => {
       res();
     });
-  })
+  });
 }
 
 function baseIterator() {
@@ -27,28 +32,30 @@ function baseIterator() {
 function createMapped(filter) {
   let iterator = baseIterator();
   for (let j = 0; j < 20; j++) {
-    iterator = iterator.map((item) => item)
-    if (filter) {
-      iterator = iterator.filter((item) => item % (j + 2) === 0)
-    }
+    iterator = iterator.map(item => item);
+    if (filter)
+      iterator = iterator.filter(item => item % (j + 2) === 0);
   }
   return iterator;
 }
 
 (async () => {
-  // Warming up
-  await run(baseIterator());
+  await run(baseIterator()); // warm-up run
 
   await perf(baseIterator(), createMapped(), '20000000 elems 20 maps\t\t\t\t\t');
   await perf(createMapped(true), createMapped(true), '20000000 elems 20 maps 20 filter\t\t\t');
 
-  let now = performance.now();
+  const now = performance.now();
   for (let j = 0; j < 100_000; j++) {
     let it = range(1, 100);
-    for (let k = 0; k < 5; k++) {
+    for (let k = 0; k < 5; k++)
       it = it.map(item => item);
-    }
-    await new Promise((res) => { it.on('data', (d) => { }).on('end', () => { res() }) })
+
+    await new Promise((resolve, reject) => {
+      it.on('data', () => null);
+      it.on('end', resolve);
+      it.on('error', reject);
+    });
   }
   console.log('100_000 iterators each with 5 maps and 100 elements\t', performance.now() - now);
 })();