WIP: Review.

Author: RubenVerborgh

asynciterator.ts

@@ -463,7 +463,7 @@ export class AsyncIterator<T> extends EventEmitter {
     @returns {module:asynciterator.AsyncIterator} A new iterator that maps the items from this iterator
   */
   map<D>(map: (item: T, it: AsyncIterator<any>) => D | null, self?: any): AsyncIterator<D> {
-    return new MappingIterator<T, D>(this, [bind(map, self)]);
+    return new MappingIterator<T, D>(this, bind(map, self));
   }
 
   /**
@@ -786,8 +786,112 @@ export class IntegerIterator extends AsyncIterator<number> {
 }
 
 
+export type MapItem<T, D = T, I extends AsyncIterator<D> = AsyncIterator<D>> =
+  (item: T, iterator: I) => D;
+
+/**
+  An iterator that calls a synchronous function
+  on every item from its source iterator.
+  @extends module:asynciterator.AsyncIterator
+*/
+export class MappingIterator<T, D = T> extends AsyncIterator<D> {
+  protected _source: AsyncIterator<T>;
+  private readonly _destroySource: boolean;
+  private readonly _mappings: MapItem<any, any, MappingIterator<T, D>>[];
+  private readonly _mappingRoot: InternalSource<any>;
+
+  // This is wrong: readable should be set by listening to source events
+  get readable() {
+    return (this._state < CLOSED) && this._source.readable;
+  }
+
+  /**
+   * Applies the given mapping to the source iterator.
+   */
+  constructor(
+    source: AsyncIterator<T>,
+    mapping?: MapItem<T, D, MappingIterator<T, D>>,
+    options?: SourcedIteratorOptions,
+  );
+
+  /**
+   * Applies the given list mappings to the mapping root.
+   */
+  constructor(
+    source: AsyncIterator<T>,
+    mappings: MapItem<any, any, MappingIterator<T, D>>[],
+    mappingRoot: AsyncIterator<any>,
+    options?: SourcedIteratorOptions,
+  );
+
+  constructor(
+    source: AsyncIterator<T>,
+    mappings: MapItem<T, D, MappingIterator<T, D>> |
+              MapItem<any, any, MappingIterator<T, D>>[] = [],
+    mappingRoot?: AsyncIterator<any> | SourcedIteratorOptions,
+    options: SourcedIteratorOptions = {},
+  ) {
+    super();
+    // Resolve optional parameters
+    if (!isEventEmitter(mappingRoot)) {
+      if (mappingRoot)
+        options = mappingRoot;
+      mappingRoot = source;
+    }
+    this._source = source;
+    this._mappings = isFunction(mappings) ? [mappings] : mappings;
+    this._mappingRoot = mappingRoot as InternalSource<any>;
+    this._destroySource = options.destroySource !== false;
+
+    if (mappingRoot.done) {
+      this.close();
+    }
+    else {
+      _validateSource(mappingRoot);
+      this._mappingRoot._destination = this;
+      this._mappingRoot.on('end', destinationClose);
+      this._mappingRoot.on('error', destinationEmitError);
+      this._mappingRoot.on('readable', destinationEmitReadable);
+    }
+  }
+
+  read(): D | null {
+    let mapped : any = null;
+    while (mapped === null && (mapped = this._source.read()) !== null) {
+      for (let i = 0; i < this._mappings.length; i++) {
+        mapped = this._mappings[i](mapped, this);
+        if (mapped === null)
+          break;
+      }
+    }
+    return mapped;
+  }
+
+  map<K>(map: (item: D, it: AsyncIterator<any>) => K | null, self?: any): AsyncIterator<K> {
+    return new MappingIterator<T, K>(this._source, [...this._mappings, bind(map, self)], this);
+  }
+
+  public close() {
+    if (this._destroySource)
+      this._mappingRoot.destroy();
+    super.close();
+  }
+
+  /* Cleans up the source iterator and ends. */
+  protected _end(destroy: boolean) {
+    this._mappingRoot.removeListener('end', destinationClose);
+    this._mappingRoot.removeListener('error', destinationEmitError);
+    this._mappingRoot.removeListener('readable', destinationEmitReadable);
+    delete this._mappingRoot._destination;
+    if (this._destroySource)
+      this._mappingRoot.destroy();
+    super._end(destroy);
+  }
+}
+
+
 /**
-  A iterator that maintains an internal buffer of items.
+  An iterator that maintains an internal buffer of items.
   This class serves as a base class for other iterators
   with a typically complex item generation process.
   @extends module:asynciterator.AsyncIterator
@@ -1252,102 +1356,23 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
   }
 }
 
-function destinationEmitError<S>(this: InternalSource<S>, error: Error) {
-  this._destination.emit('error', error);
+function destinationClose<S>(this: InternalSource<S>) {
+  this._destination.close();
 }
 function destinationCloseWhenDone<S>(this: InternalSource<S>) {
   (this._destination as any)._closeWhenDone();
 }
+function destinationEmitReadable<S>(this: InternalSource<S>) {
+  this._destination.emit('readable');
+}
+function destinationEmitError<S>(this: InternalSource<S>, error: Error) {
+  this._destination.emit('error', error);
+}
 function destinationFillBuffer<S>(this: InternalSource<S>) {
   if ((this._destination as any)._sourceStarted !== false)
     (this._destination as any)._fillBuffer();
 }
 
-export class MappingIterator<T, D = T> extends AsyncIterator<D> {
-  private _destroySource: boolean;
-
-  get readable() {
-    return this.source.readable;
-  }
-
-  set readable(readable) {
-    this.source.readable = readable;
-  }
-
-  constructor(
-    protected source: AsyncIterator<T>,
-    private transforms: ((item: any, iterator: AsyncIterator<any>) => any)[],
-    private upstream: AsyncIterator<any> = source,
-    options: { destroySource?: boolean } = {}
-  ) {
-    // Subscribe the iterator directly upstream rather than the original source to avoid over-subscribing
-    // listeners to the original source
-    super();
-    this._destroySource = options.destroySource !== false;
-    if (upstream.done) {
-      this.close();
-    }
-    else {
-      _validateSource(upstream);
-      // @ts-ignore
-      upstream._destination = this;
-      upstream.on('end', onSourceEnd);
-      upstream.on('error', onSourceError);
-      upstream.on('readable', onSourceReadable);
-    }
-  }
-
-  read(): D | null {
-    const { source, transforms } = this;
-    let item, i;
-    while ((item = source.read()) !== null) {
-      i = transforms.length;
-      // Applies each of the transforms in sequence, and terminates
-      // early if a transform returns null
-      //
-      // Do not use a for-of loop here, it slows down transformations
-      // by approximately a factor of 2.
-      while (i-- >= 1 && (item = transforms[i](item, this)) !== null)
-        ;
-      if (item !== null)
-        return item;
-    }
-    return null;
-  }
-
-  map<K>(map: (item: D, it: AsyncIterator<any>) => K | null, self?: any): AsyncIterator<K> {
-    return new MappingIterator<T, K>(this.source, [bind(map, self), ...this.transforms], this);
-  }
-
-  destroy(cause?: Error): void {
-    this.upstream.destroy(cause);
-    super.destroy(cause);
-  }
-
-  public close() {
-    this.upstream.removeListener('end', onSourceEnd);
-    this.upstream.removeListener('error', onSourceError);
-    this.upstream.removeListener('readable', onSourceReadable);
-    if (this._destroySource)
-      this.upstream.destroy();
-    scheduleTask(() => {
-      // @ts-ignore
-      delete this.upstream._destination;
-      delete this.source;
-    });
-    super.close();
-  }
-}
-
-function onSourceError<S>(this: InternalSource<S>, error: Error) {
-  this._destination.emit('error', error);
-}
-function onSourceEnd<S>(this: InternalSource<S>) {
-  this._destination.close();
-}
-function onSourceReadable<S>(this: InternalSource<S>) {
-  this._destination.emit('readable');
-}
 
 /**
   An iterator that generates items based on a source iterator
@@ -2023,15 +2048,18 @@ function isSourceExpression<T>(object: any): object is SourceExpression<T> {
   return object && (isEventEmitter(object) || isPromise(object) || isFunction(object));
 }
 
+export interface SourcedIteratorOptions {
+  destroySource?: boolean;
+}
+
 export interface BufferedIteratorOptions {
   maxBufferSize?: number;
   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> {