Consistent source setting on closed iterators.

asynciterator.ts

@@ -1206,16 +1206,17 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
   }
 
   set source(value: AsyncIterator<S> | undefined) {
-    // Do not change sources if the iterator is already done
-    if (this.done)
-      return;
-
     // Validate and set source
     const source = this._source = this._validateSource(value);
     source[DESTINATION] = this;
 
-    // Close this iterator if the source has already ended
-    if (source.done) {
+    // Do not read the source if this iterator already ended
+    if (this.done) {
+      if (this._destroySource)
+        source.destroy();
+    }
+    // Close this iterator if the source already ended
+    else if (source.done) {
       this.close();
     }
     // Otherwise, react to source events
@@ -1791,18 +1792,19 @@ export class ClonedIterator<T> extends TransformIterator<T> {
   }
 
   set source(value: AsyncIterator<T> | undefined) {
-    // Do not change sources if the iterator is already done
-    if (this.done)
-      return;
-
     // Validate and set the source
     const source = this._source = this._validateSource(value);
     // Create a history reader for the source if none already existed
     const history = (source && (source as any)[DESTINATION]) ||
       (source[DESTINATION] = new HistoryReader<T>(source) as any);
 
+    // Do not read the source if this iterator already ended
+    if (this.done) {
+      if (this._destroySource)
+        source.destroy();
+    }
     // Close this clone if history is empty and the source has ended
-    if (history.endsAt(0)) {
+    else if (history.endsAt(0)) {
       this.close();
     }
     else {

test/ClonedIterator-test.js

@@ -84,10 +84,6 @@ describe('ClonedIterator', () => {
         clone.close();
       });
 
-      it('should not do anything when a source is set', () => {
-        clone.source = {};
-      });
-
       it('should have undefined as `source` property', () => {
         expect(clone.source).to.be.undefined;
       });
@@ -115,6 +111,13 @@ describe('ClonedIterator', () => {
       it('should return an empty property set', () => {
         clone.getProperties().should.deep.equal({});
       });
+
+      it('should destroy a newly added source', () => {
+        const source = new AsyncIterator();
+        sinon.spy(source, 'destroy');
+        clone.source = source;
+        source.destroy.should.have.been.calledOnce;
+      });
     });
   });
 

test/TransformIterator-test.js

@@ -1039,6 +1039,66 @@ describe('TransformIterator', () => {
     });
   });
 
+  describe('A TransformIterator with a source creation function returning a slow promise', () => {
+    let iterator, source, createSource, sourcePromise, resolvePromise;
+    before(() => {
+      source = new ArrayIterator(['a']);
+      sinon.spy(source, 'read');
+      sinon.spy(source, 'destroy');
+      sourcePromise = new Promise(resolve => {
+        resolvePromise = resolve;
+      });
+      createSource = sinon.spy(() => sourcePromise);
+      iterator = new TransformIterator(createSource);
+      captureEvents(iterator, 'readable', 'end');
+    });
+
+    describe('before the source is created', () => {
+      it('should allow destruction', () => {
+        iterator.destroy();
+        iterator.done.should.equal(true);
+      });
+    });
+
+    describe('after the promise resolves', () => {
+      before(() => resolvePromise(source));
+
+      it('should destroy the source', () => {
+        source.destroy.should.have.been.calledOnce;
+      });
+    });
+  });
+
+  describe('A TransformIterator with a source creation function returning a slow promise without destroy source', () => {
+    let iterator, source, createSource, sourcePromise, resolvePromise;
+    before(() => {
+      source = new ArrayIterator(['a']);
+      sinon.spy(source, 'read');
+      sinon.spy(source, 'destroy');
+      sourcePromise = new Promise(resolve => {
+        resolvePromise = resolve;
+      });
+      createSource = sinon.spy(() => sourcePromise);
+      iterator = new TransformIterator(createSource, { destroySource: false });
+      captureEvents(iterator, 'readable', 'end');
+    });
+
+    describe('before the source is created', () => {
+      it('should allow destruction', () => {
+        iterator.destroy();
+        iterator.done.should.equal(true);
+      });
+    });
+
+    describe('after the promise resolves', () => {
+      before(() => resolvePromise(source));
+
+      it('should not destroy the source', () => {
+        source.destroy.should.not.have.been.called;
+      });
+    });
+  });
+
   describe('A TransformIterator with a source creation function and without autoStart', () => {
     let iterator, source, createSource, sourcePromise, resolvePromise;
     before(() => {