From 42cae481e2e8cb96d41b3dfe28c43719cb902e60 Mon Sep 17 00:00:00 2001
From: Jesse Wright <jesse.wright@anu.edu.au>
Date: Sun, 26 Jun 2022 11:24:13 +1000
Subject: [PATCH] Add MappingIterator.

---
 asynciterator.ts                     | 128 +++-
 package.json                         |   2 +-
 perf/.eslintrc                       |   6 +
 perf/MappingIterator-perf.js         |  61 ++
 test/MappingIterator-test.js         | 989 +++++++++++++++++++++++++++
 test/SimpleTransformIterator-test.js | 149 ----
 6 files changed, 1171 insertions(+), 164 deletions(-)
 create mode 100644 perf/.eslintrc
 create mode 100644 perf/MappingIterator-perf.js
 create mode 100644 test/MappingIterator-test.js

diff --git a/asynciterator.ts b/asynciterator.ts
index e7df6bb8..908b60b1 100644
--- a/asynciterator.ts
+++ b/asynciterator.ts
@@ -25,6 +25,11 @@ 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;
+}
+
 /**
   ID of the INIT state.
   An iterator is initializing if it is preparing main item generation.
@@ -161,7 +166,7 @@ export class AsyncIterator<T> extends EventEmitter {
     @param {object?} self The `this` pointer for the callback
   */
   forEach(callback: (item: T) => void, self?: object) {
-    this.on('data', self ? callback.bind(self) : callback);
+    this.on('data', bind(callback, self));
   }
 
   /**
@@ -455,8 +460,8 @@ export class AsyncIterator<T> extends EventEmitter {
     @param {object?} self The `this` pointer for the mapping function
     @returns {module:asynciterator.AsyncIterator} A new iterator that maps the items from this iterator
   */
-  map<D>(map: (item: T) => D, self?: any): AsyncIterator<D> {
-    return this.transform({ map: self ? map.bind(self) : map });
+  map<D>(map: MapFunction<T, D>, self?: any): AsyncIterator<D> {
+    return new MappingIterator(this, bind(map, self));
   }
 
   /**
@@ -469,7 +474,8 @@ export class AsyncIterator<T> extends EventEmitter {
   filter<K extends T>(filter: (item: T) => item is K, self?: any): AsyncIterator<K>;
   filter(filter: (item: T) => boolean, self?: any): AsyncIterator<T>;
   filter(filter: (item: T) => boolean, self?: any): AsyncIterator<T> {
-    return this.transform({ filter: self ? filter.bind(self) : filter });
+    filter = bind(filter, self);
+    return this.map(item => filter(item) ? item : null);
   }
 
   /**
@@ -510,7 +516,7 @@ export class AsyncIterator<T> extends EventEmitter {
     @returns {module:asynciterator.AsyncIterator} A new iterator that skips the given number of items
   */
   skip(offset: number): AsyncIterator<T> {
-    return this.transform({ offset });
+    return this.map(item => offset-- > 0 ? null : item);
   }
 
   /**
@@ -575,6 +581,14 @@ function addSingleListener(source: EventEmitter, eventName: string,
     source.on(eventName, listener);
 }
 
+// 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)
+    throw new Error('The source already has a destination');
+  return source as InternalSource<S>;
+}
 
 /**
   An iterator that doesn't emit any items.
@@ -777,9 +791,90 @@ export class IntegerIterator extends AsyncIterator<number> {
   }
 }
 
+/**
+ * A synchronous mapping function from one element to another.
+ * A return value of `null` means that nothing should be emitted for a particular item.
+ */
+export type MapFunction<S, D = S> = (item: S) => D | null;
+
+/**
+ * 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> {
+  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>,
+    map: MapFunction<S, D> = identity as MapFunction<S, D>,
+    options: SourcedIteratorOptions = {}
+  ) {
+    super();
+    this._map = map;
+    this._source = ensureSourceAvailable(source);
+    this._destroySource = options.destroySource !== false;
+
+    // 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);
+      this.readable = this._source.readable;
+    }
+  }
+
+  /** Tries to read the next item from the iterator. */
+  read(): D | null {
+    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;
+  }
+
+  /* Cleans up the source iterator and ends. */
+  protected _end(destroy: boolean) {
+    this._source.removeListener('end', destinationClose);
+    this._source.removeListener('error', destinationEmitError);
+    this._source.removeListener('readable', destinationSetReadable);
+    delete this._source._destination;
+    if (this._destroySource)
+      this._source.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
@@ -1150,14 +1245,10 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
     @param {object} source The source to validate
     @param {boolean} allowDestination Whether the source can already have a destination
   */
-  protected _validateSource(source?: AsyncIterator<S>, allowDestination = false) {
+  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');
-    if (!source || !isFunction(source.read) || !isFunction(source.on))
-      throw new Error(`Invalid source: ${source}`);
-    if (!allowDestination && (source as any)._destination)
-      throw new Error('The source already has a destination');
-    return source as InternalSource<S>;
+    return ensureSourceAvailable(source, allowDestination);
   }
 
   /**
@@ -1240,9 +1331,15 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
   }
 }
 
+function destinationSetReadable<S>(this: InternalSource<S>) {
+  this._destination!.readable = true;
+}
 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();
 }
@@ -1956,15 +2053,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> {
diff --git a/package.json b/package.json
index 575842b0..9ef1028c 100644
--- a/package.json
+++ b/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"
   },
diff --git a/perf/.eslintrc b/perf/.eslintrc
new file mode 100644
index 00000000..4f9a6d18
--- /dev/null
+++ b/perf/.eslintrc
@@ -0,0 +1,6 @@
+{
+  rules: {
+    no-console: off,
+  },
+}
+
diff --git a/perf/MappingIterator-perf.js b/perf/MappingIterator-perf.js
new file mode 100644
index 00000000..92190a52
--- /dev/null
+++ b/perf/MappingIterator-perf.js
@@ -0,0 +1,61 @@
+import { ArrayIterator, range } from '../dist/asynciterator.js';
+
+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', noop);
+    iterator.on('end', () => {
+      res();
+    });
+  });
+}
+
+function baseIterator() {
+  let i = 0;
+  return new ArrayIterator(new Array(20000000).fill(true).map(() => i++));
+}
+
+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);
+  }
+  return iterator;
+}
+
+(async () => {
+  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');
+
+  const now = performance.now();
+  for (let j = 0; j < 100_000; j++) {
+    let it = range(1, 100);
+    for (let k = 0; k < 5; k++)
+      it = it.map(item => item);
+
+    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);
+})();
diff --git a/test/MappingIterator-test.js b/test/MappingIterator-test.js
new file mode 100644
index 00000000..590c3dd2
--- /dev/null
+++ b/test/MappingIterator-test.js
@@ -0,0 +1,989 @@
+import {
+  AsyncIterator,
+  ArrayIterator,
+  IntegerIterator,
+  MappingIterator,
+  range,
+  fromArray,
+  wrap,
+  ENDED,
+} from '../dist/asynciterator.js';
+
+import { EventEmitter } from 'events';
+import { expect } from 'chai';
+
+describe('MappingIterator', () => {
+  describe('The MappingIterator function', () => {
+    describe('the result when called with `new`', () => {
+      let instance;
+
+      before(() => {
+        instance = new MappingIterator(new ArrayIterator([]));
+      });
+
+      it('should be a MappingIterator object', () => {
+        instance.should.be.an.instanceof(MappingIterator);
+      });
+
+      it('should be an AsyncIterator object', () => {
+        instance.should.be.an.instanceof(AsyncIterator);
+      });
+
+      it('should be an EventEmitter object', () => {
+        instance.should.be.an.instanceof(EventEmitter);
+      });
+    });
+  });
+
+  describe('A MappingIterator with an array source', () => {
+    let iterator, source;
+
+    before(() => {
+      source = new ArrayIterator([0, 1, 2, 3, 4, 5, 6]);
+      iterator = new MappingIterator(source);
+    });
+
+    describe('when reading items', () => {
+      const items = [];
+
+      before(done => {
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', done);
+      });
+
+      it('should return all items', () => {
+        items.should.deep.equal([0, 1, 2, 3, 4, 5, 6]);
+      });
+    });
+  });
+
+  describe('A MappingIterator with a source that emits 0 items', () => {
+    it('should not return any items', done => {
+      const items = [];
+      const iterator = new MappingIterator(new ArrayIterator([]));
+      iterator.on('data', item => { items.push(item); });
+      iterator.on('end', () => {
+        items.should.deep.equal([]);
+        done();
+      });
+    });
+  });
+
+  describe('A MappingIterator with a source that is already ended', () => {
+    it('should not return any items', done => {
+      const items = [];
+      const source = new ArrayIterator([]);
+      source.on('end', () => {
+        const iterator = new MappingIterator(source);
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', () => {
+          items.should.deep.equal([]);
+          done();
+        });
+      });
+    });
+  });
+
+  describe('A TransformIterator with destroySource set to its default', () => {
+    let iterator, source;
+
+    before(() => {
+      source = new ArrayIterator([1, 2, 3]);
+      iterator = new MappingIterator(source);
+    });
+
+    describe('after being closed', () => {
+      before(done => {
+        iterator.read();
+        iterator.close();
+        iterator.on('end', done);
+      });
+
+      it('should have destroyed the source', () => {
+        expect(source).to.have.property('destroyed', true);
+      });
+    });
+  });
+
+  describe('A TransformIterator with destroySource set to false', () => {
+    let iterator, source;
+
+    before(() => {
+      source = new ArrayIterator([1, 2, 3]);
+      iterator = new MappingIterator(source, x => x, { destroySource: false });
+    });
+
+    describe('after being closed', () => {
+      before(done => {
+        iterator.read();
+        iterator.close();
+        iterator.on('end', done);
+      });
+
+      it('should not have destroyed the source', () => {
+        expect(source).to.have.property('destroyed', false);
+      });
+    });
+  });
+
+  describe('A TransformIterator with destroySource set to false', () => {
+    let iterator, source;
+
+    before(() => {
+      source = new ArrayIterator([1, 2, 3]);
+      captureEvents(source, 'readable', 'end');
+    });
+
+    describe('after being closed', () => {
+      it('should read an element from the original source', () => {
+        expect(source.read()).to.equal(1);
+      });
+
+      it('should read the next elements from the mapping iterator', () => {
+        iterator = new MappingIterator(source, x => x, { destroySource: false });
+        captureEvents(iterator, 'readable', 'end');
+
+        expect(iterator.read()).to.equal(2);
+        expect(iterator.read()).to.equal(3);
+        expect(iterator.read()).to.equal(null);
+      });
+
+      it('should have emitted the `end` event', () => {
+        iterator._eventCounts.end.should.equal(1);
+      });
+
+      it('should not be readable', () => {
+        iterator.readable.should.be.false;
+      });
+
+      it('should have ended', () => {
+        iterator.ended.should.be.true;
+      });
+
+      it('source should have emitted the `end` event', () => {
+        source._eventCounts.end.should.equal(1);
+      });
+
+      it('source should not be readable', () => {
+        source.readable.should.be.false;
+      });
+
+      it('source should have ended', () => {
+        source.ended.should.be.true;
+      });
+    });
+  });
+
+  describe('A MappingIterator with a map function', () => {
+    let iterator, source, map;
+    before(() => {
+      let i = 0;
+      source = new ArrayIterator(['a', 'b', 'c']);
+      map = sinon.spy(item => item + (++i));
+      iterator = new MappingIterator(source, map);
+    });
+
+    describe('when reading items', () => {
+      const items = [];
+      before(done => {
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', done);
+      });
+
+      it('should execute the map function on all items in order', () => {
+        items.should.deep.equal(['a1', 'b2', 'c3']);
+      });
+
+      it('should have called the map function once for each item', () => {
+        map.should.have.been.calledThrice;
+      });
+
+      it('should have called the map function with the iterator as `this`', () => {
+        map.alwaysCalledOn(iterator).should.be.true;
+      });
+    });
+  });
+
+  describe('A MappingIterator with a map function that returns null', () => {
+    let iterator, source, map;
+    before(() => {
+      let i = 0;
+      source = new ArrayIterator(['a', 'b', 'c']);
+      map = sinon.spy(item => {
+        if (++i === 2)
+          return null;
+        return item + i;
+      });
+      iterator = new MappingIterator(source, map);
+    });
+
+    describe('when reading items', () => {
+      const items = [];
+      before(done => {
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', done);
+      });
+
+      it('should execute the map function on all items in order, skipping null', () => {
+        items.should.deep.equal(['a1', 'c3']);
+      });
+
+      it('should have called the map function once for each item', () => {
+        map.should.have.been.calledThrice;
+      });
+
+      it('should have called the map function with the iterator as `this`', () => {
+        map.alwaysCalledOn(iterator).should.be.true;
+      });
+    });
+  });
+
+  describe('A MappingIterator with a source that returns null and sets readable false', () => {
+    let iterator, source, map;
+    before(() => {
+      let i = 0;
+      source = new AsyncIterator();
+      source._readable = true;
+      source.read = () => {
+        if (i % 2 === 0) {
+          i += 1;
+          return i;
+        }
+        source.readable = false;
+        return null;
+      };
+      map = sinon.spy(item => item);
+      iterator = new MappingIterator(source, map);
+    });
+
+    describe('when reading items', () => {
+      it('should be readable to start', () => {
+        iterator.readable.should.be.true;
+      });
+
+      it('should return 0 for the first item', () => {
+        iterator.read().should.equal(1);
+      });
+
+      it('should return null for the next item', () => {
+        expect(iterator.read()).to.equal(null);
+      });
+
+      it('source should should not be readable after null item', () => {
+        source.readable.should.be.false;
+      });
+
+      it('iterator should should not be readable after null item', () => {
+        iterator.readable.should.be.false;
+      });
+    });
+  });
+
+  describe('A MappingIterator with a source that returns null but stays readable', () => {
+    let iterator, source, map;
+    before(() => {
+      let i = 0;
+      source = new AsyncIterator();
+      source._readable = true;
+      source.read = () => {
+        if (i % 2 === 0) {
+          i += 1;
+          return i;
+        }
+        return null;
+      };
+      map = sinon.spy(item => item);
+      iterator = new MappingIterator(source, map);
+    });
+
+    describe('when reading items', () => {
+      it('should be readable to start', () => {
+        iterator.readable.should.be.true;
+      });
+
+      it('should return 0 for the first item', () => {
+        iterator.read().should.equal(1);
+      });
+
+      it('should return null for the next item', () => {
+        expect(iterator.read()).to.equal(null);
+      });
+
+      it('source should should not be readable after null item', () => {
+        source.readable.should.be.true;
+      });
+
+      it('iterator should should not be readable after null item', () => {
+        iterator.readable.should.be.false;
+      });
+    });
+  });
+
+  describe('A MappingIterator with a map function that closes', () => {
+    let iterator, source, map;
+    before(() => {
+      source = new AsyncIterator();
+      source.read = sinon.spy(() => 1);
+      source._readable = true;
+      let read = false;
+      map = function (item) {
+        iterator.close();
+        if (read)
+          return null;
+
+        read = true;
+        return item;
+      };
+      iterator = new MappingIterator(source, map);
+      captureEvents(iterator, 'readable', 'end');
+    });
+
+    describe('before reading an item', () => {
+      it('should not have called `read` on the source', () => {
+        source.read.should.not.have.been.called;
+      });
+
+      it('should have emitted the `readable` event', () => {
+        iterator._eventCounts.readable.should.equal(1);
+      });
+
+      it('should not have emitted the `end` event', () => {
+        iterator._eventCounts.end.should.equal(0);
+      });
+
+      it('should be readable', () => {
+        iterator.readable.should.be.true;
+      });
+
+      it('should not have ended', () => {
+        iterator.ended.should.be.false;
+      });
+    });
+
+    describe('after reading a first item', () => {
+      let item;
+      before(() => {
+        item = iterator.read();
+      });
+
+      it('should read the correct item', () => {
+        item.should.equal(1);
+      });
+
+      it('should have called `read` on the source only once', () => {
+        source.read.should.have.been.calledOnce;
+      });
+
+      it('should have emitted the `end` event', () => {
+        iterator._eventCounts.end.should.equal(1);
+      });
+
+      it('should not be readable', () => {
+        iterator.readable.should.be.false;
+      });
+
+      it('should have ended', () => {
+        iterator.ended.should.be.true;
+      });
+    });
+
+    describe('after attempting to read again', () => {
+      before(() => {
+        iterator.read();
+      });
+
+      it('should have called `read` on the source only once', () => {
+        source.read.should.have.been.calledOnce;
+      });
+
+      it('should have emitted the `end` event', () => {
+        iterator._eventCounts.end.should.equal(1);
+      });
+
+      it('should not be readable', () => {
+        iterator.readable.should.be.false;
+      });
+
+      it('should have ended', () => {
+        iterator.ended.should.be.true;
+      });
+    });
+  });
+
+  describe('A MappingIterator with a map function that closes', () => {
+    let iterator, source;
+    before(() => {
+      source = new AsyncIterator();
+      source.read = sinon.spy(() => 1);
+      source._readable = true;
+      iterator = new MappingIterator(source, x => x);
+      captureEvents(iterator, 'readable', 'end');
+    });
+
+    describe('before reading an item', () => {
+      it('should not have called `read` on the source', () => {
+        source.read.should.not.have.been.called;
+      });
+
+      it('should have emitted the `readable` event', () => {
+        iterator._eventCounts.readable.should.equal(1);
+      });
+
+      it('should not have emitted the `end` event', () => {
+        iterator._eventCounts.end.should.equal(0);
+      });
+
+      it('should be readable', () => {
+        iterator.readable.should.be.true;
+      });
+
+      it('should not have ended', () => {
+        iterator.ended.should.be.false;
+      });
+
+      it('should return 1 when read', () => {
+        iterator.read().should.equal(1);
+      });
+
+      it('should return 1 when read', () => {
+        iterator.read().should.equal(1);
+      });
+
+      it('should return null when read immediately after source is not readable', () => {
+        source.readable = false;
+        expect(iterator.read()).to.equal(null);
+      });
+
+      it('should return 1 when read on readable source', () => {
+        source.readable = true;
+        expect(iterator.read()).to.equal(1);
+      });
+
+      it('should return null and close if source is not readable and closed', () => {
+        source.readable = false;
+        source._changeState(ENDED);
+        expect(iterator.read()).to.equal(null);
+      });
+
+      it('should have be closed after the next tick', () => {
+        // TODO: Whilst it is correct to wait a tick to do this, it is actually more performant,
+        // and also correct to end *immediately* as this enables a long chain of iterators to be
+        // closed on the same tick rather than over the course of multiple ticks. This should be
+        // changed and tests added for this when merging into
+        // https://github.com/RubenVerborgh/AsyncIterator/pull/45
+        expect(iterator.done).to.equal(true);
+      });
+    });
+  });
+
+  describe('A TransformIterator with a source that errors', () => {
+    let iterator, source, errorHandler;
+
+    before(() => {
+      source = new AsyncIterator();
+      iterator = new MappingIterator(source);
+      iterator.on('error', errorHandler = sinon.stub());
+    });
+
+    describe('before an error occurs', () => {
+      it('should not have emitted any error', () => {
+        errorHandler.should.not.have.been.called;
+      });
+    });
+
+    describe('after a first error occurs', () => {
+      let error1;
+      before(() => {
+        errorHandler.reset();
+        source.emit('error', error1 = new Error('error1'));
+      });
+
+      it('should re-emit the error', () => {
+        errorHandler.should.have.been.calledOnce;
+        errorHandler.should.have.been.calledWith(error1);
+      });
+    });
+
+    describe('after a second error occurs', () => {
+      let error2;
+
+      before(() => {
+        errorHandler.reset();
+        source.emit('error', error2 = new Error('error2'));
+      });
+
+      it('should re-emit the error', () => {
+        errorHandler.should.have.been.calledOnce;
+        errorHandler.should.have.been.calledWith(error2);
+      });
+    });
+
+    describe('after the source has ended and errors again', () => {
+      before(done => {
+        errorHandler.reset();
+        source.close();
+        iterator.on('end', () => {
+          function noop() { /* */ }
+          source.on('error', noop); // avoid triggering the default error handler
+          source.emit('error', new Error('error3'));
+          source.removeListener('error', noop);
+          done();
+        });
+      });
+
+      it('should not re-emit the error', () => {
+        errorHandler.should.not.have.been.called;
+      });
+
+      it('should not leave any error handlers attached', () => {
+        source.listenerCount('error').should.equal(0);
+      });
+    });
+  });
+
+  describe('A chain of maps and filters', () => {
+    for (const iteratorGen of [() => range(0, 2), () => fromArray([0, 1, 2]), () => wrap(range(0, 2))]) {
+      // eslint-disable-next-line no-loop-func
+      describe(`with ${iteratorGen()}`, () => {
+        let iterator;
+
+        beforeEach(() => {
+          iterator = iteratorGen();
+        });
+
+        it('should handle no transforms arrayified', async () => {
+          (await iterator.toArray()).should.deep.equal([0, 1, 2]);
+        });
+
+        it('should apply maps that doubles correctly', async () => {
+          (await iterator.map(x => x * 2).toArray()).should.deep.equal([0, 2, 4]);
+        });
+
+        it('should apply maps that doubles correctly and then maybemaps', async () => {
+          (await iterator.map(x => x * 2).map(x => x === 2 ? null : x * 3).toArray()).should.deep.equal([0, 12]);
+        });
+
+        it('should apply maps that maybemaps correctly', async () => {
+          (await iterator.map(x => x === 2 ? null : x * 3).toArray()).should.deep.equal([0, 3]);
+        });
+
+        it('should apply maps that maybemaps twice', async () => {
+          (await iterator.map(x => x === 2 ? null : x * 3).map(x => x === 0 ? null : x * 3).toArray()).should.deep.equal([9]);
+        });
+
+        it('should apply maps that converts to string', async () => {
+          (await iterator.map(x => `x${x}`).toArray()).should.deep.equal(['x0', 'x1', 'x2']);
+        });
+
+        it('should apply filter correctly', async () => {
+          (await iterator.filter(x => x % 2 === 0).toArray()).should.deep.equal([0, 2]);
+        });
+
+        it('should apply filter then map correctly', async () => {
+          (await iterator.filter(x => x % 2 === 0).map(x => `x${x}`).toArray()).should.deep.equal(['x0', 'x2']);
+        });
+
+        it('should apply map then filter correctly (1)', async () => {
+          (await iterator.map(x => x).filter(x => x % 2 === 0).toArray()).should.deep.equal([0, 2]);
+        });
+
+        it('should apply map then filter to false correctly', async () => {
+          (await iterator.map(x => `x${x}`).filter(x => true).toArray()).should.deep.equal(['x0', 'x1', 'x2']);
+        });
+
+        it('should apply map then filter to true correctly', async () => {
+          (await iterator.map(x => `x${x}`).filter(x => false).toArray()).should.deep.equal([]);
+        });
+
+        it('should apply filter to false then map correctly', async () => {
+          (await iterator.filter(x => true).map(x => `x${x}`).toArray()).should.deep.equal(['x0', 'x1', 'x2']);
+        });
+
+        it('should apply filter to true then map correctly', async () => {
+          (await iterator.filter(x => false).map(x => `x${x}`).filter(x => false).toArray()).should.deep.equal([]);
+        });
+
+        it('should apply filter one then double', async () => {
+          (await iterator.filter(x => x !== 1).map(x => x * 2).toArray()).should.deep.equal([0, 4]);
+        });
+
+        it('should apply double then filter one', async () => {
+          (await iterator.map(x => x * 2).filter(x => x !== 1).toArray()).should.deep.equal([0, 2, 4]);
+        });
+
+        it('should apply map then filter correctly', async () => {
+          (await iterator.map(x => `x${x}`).filter(x => (x[1] === '0')).toArray()).should.deep.equal(['x0']);
+        });
+
+        it('should correctly apply 3 filters', async () => {
+          (await range(0, 5).filter(x => x !== 1).filter(x => x !== 2).filter(x => x !== 2).toArray()).should.deep.equal([0, 3, 4, 5]);
+        });
+
+        it('should correctly apply 3 maps', async () => {
+          (await range(0, 1).map(x => x * 2).map(x => `z${x}`).map(x => `y${x}`).toArray()).should.deep.equal(['yz0', 'yz2']);
+        });
+
+        it('should correctly apply a map, followed by a filter, followed by another map', async () => {
+          (await range(0, 1).map(x => x * 2).filter(x => x !== 2).map(x => `y${x}`).toArray()).should.deep.equal(['y0']);
+        });
+
+        it('should correctly apply a filter-map-filter', async () => {
+          (await range(0, 2).filter(x => x !== 1).map(x => x * 3).filter(x => x !== 6).toArray()).should.deep.equal([0]);
+        });
+
+        describe('chaining with .map', () => {
+          let iter;
+
+          beforeEach(() => {
+            iter = iterator.map(x => x);
+          });
+
+          describe('when iter is closed', () => {
+            beforeEach(done => {
+              iter.on('end', done);
+              iter.close();
+            });
+
+            it('should have the primary iterator destroyed', () => {
+              iter.closed.should.be.true;
+              iterator.destroyed.should.be.true;
+            });
+          });
+
+          describe('nested chaining with .map', () => {
+            let nestedIter;
+
+            beforeEach(done => {
+              nestedIter = iter.map(x => x);
+              nestedIter.on('end', done);
+              nestedIter.close();
+            });
+
+            it('should have the primary and first level iterator destroyed with the last one closed', () => {
+              iterator.destroyed.should.be.true;
+              iter.destroyed.should.be.true;
+              nestedIter.closed.should.be.true;
+            });
+          });
+        });
+
+        describe('when called on an iterator with a `this` argument', () => {
+          const self = {};
+          let map, result;
+
+          before(() => {
+            let i = 0;
+            iterator = new ArrayIterator(['a', 'b', 'c']);
+            map = sinon.spy(item => item + (++i));
+            result = iterator.map(map, self);
+          });
+
+          describe('the return value', () => {
+            const items = [];
+
+            before(done => {
+              result.on('data', item => { items.push(item); });
+              result.on('end', done);
+            });
+
+            it('should call the map function once for each item', () => {
+              map.should.have.been.calledThrice;
+            });
+
+            it('should call the map function with the passed argument as `this`', () => {
+              map.alwaysCalledOn(self).should.be.true;
+            });
+          });
+        });
+
+        describe('when called on an iterator with a `this` argument with nested map', () => {
+          const self = {};
+          let map, result;
+
+          before(() => {
+            let i = 0;
+            iterator = new ArrayIterator(['a', 'b', 'c']);
+            map = sinon.spy(item => item + (++i));
+            result = iterator.map(x => x).map(map, self);
+          });
+
+          describe('the return value', () => {
+            const items = [];
+
+            before(done => {
+              result.on('data', item => { items.push(item); });
+              result.on('end', done);
+            });
+
+            it('should call the map function once for each item', () => {
+              map.should.have.been.calledThrice;
+            });
+
+            it('should call the map function with the passed argument as `this`', () => {
+              map.alwaysCalledOn(self).should.be.true;
+            });
+          });
+        });
+      });
+    }
+  });
+
+  describe('The AsyncIterator#skip function', () => {
+    it('should be a function', () => {
+      expect(AsyncIterator.prototype.skip).to.be.a('function');
+    });
+
+    describe('when called on an iterator', () => {
+      let iterator, result;
+      before(() => {
+        iterator = new ArrayIterator(['a', 'b', 'c', 'd', 'e']);
+        result = iterator.skip(2);
+      });
+
+      describe('the return value', () => {
+        const items = [];
+        before(done => {
+          result.on('data', item => { items.push(item); });
+          result.on('end', done);
+        });
+
+        it('should skip the given number of items', () => {
+          items.should.deep.equal(['c', 'd', 'e']);
+        });
+      });
+    });
+  });
+
+  describe('The AsyncIterator#take function', () => {
+    it('should be a function', () => {
+      expect(AsyncIterator.prototype.take).to.be.a('function');
+    });
+
+    describe('when called on an iterator', () => {
+      let iterator, result;
+      before(() => {
+        iterator = new ArrayIterator(['a', 'b', 'c', 'd', 'e']);
+        result = iterator.take(3);
+      });
+
+      describe('the return value', () => {
+        const items = [];
+        before(done => {
+          result.on('data', item => { items.push(item); });
+          result.on('end', done);
+        });
+
+        it('should take the given number of items', () => {
+          items.should.deep.equal(['a', 'b', 'c']);
+        });
+      });
+    });
+  });
+
+  describe('The AsyncIterator#range function', () => {
+    it('should be a function', () => {
+      expect(AsyncIterator.prototype.range).to.be.a('function');
+    });
+
+    describe('when called on an iterator', () => {
+      let iterator, result;
+      before(() => {
+        iterator = new IntegerIterator();
+        result = iterator.range(20, 29);
+      });
+
+      describe('the return value', () => {
+        const items = [];
+        before(done => {
+          result.on('data', item => { items.push(item); });
+          result.on('end', done);
+        });
+
+        it('should contain the indicated range', () => {
+          items.should.have.length(10);
+          items[0].should.equal(20);
+          items[9].should.equal(29);
+        });
+      });
+    });
+
+    describe('when called on an iterator with an inverse range', () => {
+      let iterator, result;
+      before(() => {
+        iterator = new IntegerIterator();
+        sinon.spy(iterator, 'read');
+      });
+
+      describe('the return value', () => {
+        const items = [];
+        before(done => {
+          result = iterator.range(30, 20);
+          result.on('data', item => { items.push(item); });
+          result.on('end', done);
+        });
+
+        it('should be empty', () => {
+          items.should.be.empty;
+        });
+      });
+    });
+  });
+
+  describe('Skipping', () => {
+    describe('The .skip function', () => {
+      describe('the result', () => {
+        let instance;
+
+        before(() => {
+          instance = new ArrayIterator([]).skip(10);
+        });
+
+        it('should be an AsyncIterator object', () => {
+          instance.should.be.an.instanceof(AsyncIterator);
+        });
+
+        it('should be an EventEmitter object', () => {
+          instance.should.be.an.instanceof(EventEmitter);
+        });
+      });
+    });
+
+    describe('A .skip on an array', () => {
+      let iterator, source;
+
+      before(() => {
+        source = new ArrayIterator([0, 1, 2, 3, 4, 5, 6]);
+        iterator = source.skip(4);
+      });
+
+      describe('when reading items', () => {
+        const items = [];
+
+        before(done => {
+          iterator.on('data', item => { items.push(item); });
+          iterator.on('end', done);
+        });
+
+        it('should return items skipping the specified amount', () => {
+          items.should.deep.equal([4, 5, 6]);
+        });
+      });
+    });
+
+    describe('A .skip on a range', () => {
+      let iterator, source;
+
+      before(() => {
+        source = range(0, 6);
+        iterator = source.skip(4);
+      });
+
+      describe('when reading items', () => {
+        const items = [];
+
+        before(done => {
+          iterator.on('data', item => { items.push(item); });
+          iterator.on('end', done);
+        });
+
+        it('should return items skipping the specified amount', () => {
+          items.should.deep.equal([4, 5, 6]);
+        });
+      });
+    });
+
+    describe('A .skip with a source that emits 0 items', () => {
+      it('should not return any items', done => {
+        const items = [];
+        const iterator = new ArrayIterator([]).skip(10);
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', () => {
+          items.should.deep.equal([]);
+          done();
+        });
+      });
+    });
+
+    describe('A .skip with a limit of 0 items', () => {
+      it('should emit all items', done => {
+        const items = [];
+        const iterator = new ArrayIterator([0, 1, 2, 3, 4, 5, 6]).skip(0);
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', () => {
+          items.should.deep.equal([0, 1, 2, 3, 4, 5, 6]);
+          done();
+        });
+      });
+    });
+
+    describe('A .skip with a limit of Infinity items', () => {
+      it('should skip all items', done => {
+        const items = [];
+        const iterator = new ArrayIterator([0, 1, 2, 3, 4, 5, 6]).skip(Infinity);
+        iterator.on('data', item => { items.push(item); });
+        iterator.on('end', () => {
+          items.should.deep.equal([]);
+          done();
+        });
+      });
+    });
+
+    describe('.take', () => {
+      describe('A .take', () => {
+        let iterator, source;
+        before(() => {
+          source = new ArrayIterator([0, 1, 2, 3, 4, 5, 6]);
+          iterator = source.take(4);
+        });
+
+        describe('when reading items', () => {
+          const items = [];
+          before(done => {
+            iterator.on('data', item => { items.push(item); });
+            iterator.on('end', done);
+          });
+
+          it('should return items to the specified take', () => {
+            items.should.deep.equal([0, 1, 2, 3]);
+          });
+        });
+      });
+
+      describe('A .take with a source that emits 0 items', () => {
+        it('should not return any items', done => {
+          const items = [];
+          const iterator = new ArrayIterator([]).take(10);
+          iterator.on('data', item => { items.push(item); });
+          iterator.on('end', () => {
+            items.should.deep.equal([]);
+            done();
+          });
+        });
+      });
+
+      describe('A .take with a take of 0 items', () => {
+        it('should not emit any items', done => {
+          const items = [];
+          const iterator = new ArrayIterator([0, 1, 2]).take(0);
+          iterator.on('data', item => { items.push(item); });
+          iterator.on('end', () => {
+            items.should.deep.equal([]);
+            done();
+          });
+        });
+      });
+
+      describe('A .take with a take of Infinity items', () => {
+        it('should emit all items', done => {
+          const items = [];
+          const iterator = new ArrayIterator([0, 1, 2, 3, 4, 5, 6]).take(Infinity);
+          iterator.on('data', item => { items.push(item); });
+          iterator.on('end', () => {
+            items.should.deep.equal([0, 1, 2, 3, 4, 5, 6]);
+            done();
+          });
+        });
+      });
+    });
+  });
+});
diff --git a/test/SimpleTransformIterator-test.js b/test/SimpleTransformIterator-test.js
index e033b8e6..039cae08 100644
--- a/test/SimpleTransformIterator-test.js
+++ b/test/SimpleTransformIterator-test.js
@@ -1110,10 +1110,6 @@ describe('SimpleTransformIterator', () => {
           result.on('end', done);
         });
 
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
         it('should execute the map function on all items in order', () => {
           items.should.deep.equal(['a1', 'b2', 'c3']);
         });
@@ -1121,10 +1117,6 @@ describe('SimpleTransformIterator', () => {
         it('should call the map function once for each item', () => {
           map.should.have.been.calledThrice;
         });
-
-        it('should call the map function with the returned iterator as `this`', () => {
-          map.alwaysCalledOn(result).should.be.true;
-        });
       });
     });
 
@@ -1145,14 +1137,6 @@ describe('SimpleTransformIterator', () => {
           result.on('end', done);
         });
 
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
-        it('should execute the map function on all items in order', () => {
-          items.should.deep.equal(['a1', 'b2', 'c3']);
-        });
-
         it('should call the map function once for each item', () => {
           map.should.have.been.calledThrice;
         });
@@ -1184,10 +1168,6 @@ describe('SimpleTransformIterator', () => {
           result.on('end', done);
         });
 
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
         it('should execute the filter function on all items in order', () => {
           items.should.deep.equal(['a', 'c']);
         });
@@ -1195,10 +1175,6 @@ describe('SimpleTransformIterator', () => {
         it('should call the filter function once for each item', () => {
           filter.should.have.been.calledThrice;
         });
-
-        it('should call the filter function with the returned iterator as `this`', () => {
-          filter.alwaysCalledOn(result).should.be.true;
-        });
       });
     });
 
@@ -1218,10 +1194,6 @@ describe('SimpleTransformIterator', () => {
           result.on('end', done);
         });
 
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
         it('should execute the filter function on all items in order', () => {
           items.should.deep.equal(['a', 'c']);
         });
@@ -1327,127 +1299,6 @@ describe('SimpleTransformIterator', () => {
     });
   });
 
-  describe('The AsyncIterator#skip function', () => {
-    it('should be a function', () => {
-      expect(AsyncIterator.prototype.skip).to.be.a('function');
-    });
-
-    describe('when called on an iterator', () => {
-      let iterator, result;
-      before(() => {
-        iterator = new ArrayIterator(['a', 'b', 'c', 'd', 'e']);
-        result = iterator.skip(2);
-      });
-
-      describe('the return value', () => {
-        const items = [];
-        before(done => {
-          result.on('data', item => { items.push(item); });
-          result.on('end', done);
-        });
-
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
-        it('should skip the given number of items', () => {
-          items.should.deep.equal(['c', 'd', 'e']);
-        });
-      });
-    });
-  });
-
-  describe('The AsyncIterator#take function', () => {
-    it('should be a function', () => {
-      expect(AsyncIterator.prototype.take).to.be.a('function');
-    });
-
-    describe('when called on an iterator', () => {
-      let iterator, result;
-      before(() => {
-        iterator = new ArrayIterator(['a', 'b', 'c', 'd', 'e']);
-        result = iterator.take(3);
-      });
-
-      describe('the return value', () => {
-        const items = [];
-        before(done => {
-          result.on('data', item => { items.push(item); });
-          result.on('end', done);
-        });
-
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
-        it('should take the given number of items', () => {
-          items.should.deep.equal(['a', 'b', 'c']);
-        });
-      });
-    });
-  });
-
-  describe('The AsyncIterator#range function', () => {
-    it('should be a function', () => {
-      expect(AsyncIterator.prototype.range).to.be.a('function');
-    });
-
-    describe('when called on an iterator', () => {
-      let iterator, result;
-      before(() => {
-        iterator = new IntegerIterator();
-        result = iterator.range(20, 29);
-      });
-
-      describe('the return value', () => {
-        const items = [];
-        before(done => {
-          result.on('data', item => { items.push(item); });
-          result.on('end', done);
-        });
-
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
-        it('should contain the indicated range', () => {
-          items.should.have.length(10);
-          items[0].should.equal(20);
-          items[9].should.equal(29);
-        });
-      });
-    });
-
-    describe('when called on an iterator with an inverse range', () => {
-      let iterator, result;
-      before(() => {
-        iterator = new IntegerIterator();
-        sinon.spy(iterator, 'read');
-      });
-
-      describe('the return value', () => {
-        const items = [];
-        before(done => {
-          result = iterator.range(30, 20);
-          result.on('data', item => { items.push(item); });
-          result.on('end', done);
-        });
-
-        it('should be a SimpleTransformIterator', () => {
-          result.should.be.an.instanceof(SimpleTransformIterator);
-        });
-
-        it('should be empty', () => {
-          items.should.be.empty;
-        });
-
-        it('should not have called `read` on the iterator', () => {
-          iterator.read.should.not.have.been.called;
-        });
-      });
-    });
-  });
-
   describe('The AsyncIterator#transform function', () => {
     it('should be a function', () => {
       expect(AsyncIterator.prototype.transform).to.be.a('function');