fix(take): fix take() behavior when stopping

Bug was: take synchronously next'ed then complete'd and then would remove the TakeOperator from the
list of listeners of its input stream, which would be a concurrent change to the array, while it was
being iterated over in Stream._n, causing bad state in that iteration.

src/core.ts

@@ -952,13 +952,14 @@ export class TakeOperator<T> implements Operator<T, T> {
   }
 
   _stop(): void {
-    this.ins._remove(this);
+    setTimeout(() => { this.ins._remove(this); }, 0);
     this.out = null;
     this.taken = 0;
   }
 
   _n(t: T) {
     const out = this.out;
+    if (!out) return;
     if (this.taken++ < this.max - 1) {
       out._n(t);
     } else {
@@ -969,11 +970,15 @@ export class TakeOperator<T> implements Operator<T, T> {
   }
 
   _e(err: any) {
-    this.out._e(err);
+    const out = this.out;
+    if (!out) return;
+    out._e(err);
   }
 
   _c() {
-    this.out._c();
+    const out = this.out;
+    if (!out) return;
+    out._c();
   }
 }
 

tests/extra/flattenSequentially.ts

@@ -41,14 +41,14 @@ describe('flattenSequentially (extra)', () => {
     });
 
     it('should expand 3 sync events as a periodic each', (done) => {
-      const stream = xs.of(0, 1, 2)
+      const stream = xs.of(1, 2, 3)
         .map(i => xs.periodic(100 * i).take(2).map(x => `${i}${x}`))
         .compose(flattenSequentially());
       // ---x---x---x---x---x---x
-      // ---00--01
-      //         -------10------11
-      //                         -----------20----------21
-      const expected = ['00', '01', '10', '11', '20', '21'];
+      // ---10--11
+      //         -------20------21
+      //                         -----------30----------31
+      const expected = ['10', '11', '20', '21', '30', '31'];
       const listener = {
         next: (x: number) => {
           assert.equal(x, expected.shift());

tests/extra/fromEvent.ts

@@ -64,18 +64,16 @@ describe('fromEvent (extra)', () => {
 
     let expected = [1, 2, 3];
 
-    const listener = {
-      next(x: any) {
+    stream.addListener({
+      next: (x: any) => {
         assert.strictEqual(x, expected.shift());
       },
-      error: done,
+      error: (err: any) => done(err),
       complete: () => {
         assert.strictEqual(expected.length, 0);
         done();
       }
-    };
-
-    stream.addListener(listener);
+    });
 
     target.emit(1);
     target.emit(2);
@@ -88,8 +86,8 @@ describe('fromEvent (extra)', () => {
     const stream = fromEvent(target, 'test', true);
 
     stream.take(1).addListener({
-      next(x) {},
-      error: done,
+      next: (x) => {},
+      error: (err: any) => done(err),
       complete() {
         setTimeout(() => {
           assert.strictEqual('test', target.removedEvent);

tests/operator/endWhen.ts

@@ -11,11 +11,11 @@ describe('Stream.prototype.endWhen', () => {
 
     stream.addListener({
       next: (x: number) => {
-        assert.equal(x, expected.shift());
+        assert.strictEqual(x, expected.shift());
       },
       error: (err: any) => done(err),
       complete: () => {
-        assert.equal(expected.length, 0);
+        assert.strictEqual(expected.length, 0);
         done();
       },
     });
@@ -29,11 +29,11 @@ describe('Stream.prototype.endWhen', () => {
 
     stream.addListener({
       next: (x: number) => {
-        assert.equal(x, expected.shift());
+        assert.strictEqual(x, expected.shift());
       },
       error: (err: any) => done(err),
       complete: () => {
-        assert.equal(expected.length, 0);
+        assert.strictEqual(expected.length, 0);
         done();
       },
     });

tests/operator/flatten.ts

@@ -1,4 +1,4 @@
-import xs from '../../src/index';
+import xs, {Stream} from '../../src/index';
 import * as assert from 'assert';
 
 describe('Stream.prototype.flatten', () => {
@@ -22,14 +22,14 @@ describe('Stream.prototype.flatten', () => {
     });
 
     it('should expand 3 sync events as a periodic, only last one passes', (done) => {
-      const stream = xs.fromArray([0, 1, 2])
+      const stream = xs.fromArray([1, 2, 3])
         .map(i => xs.periodic(100 * i).take(2).map(x => `${i}${x}`))
         .flatten();
       // ---x---x---x---x---x---x
-      // ---00--01
-      // -------10------11
-      // -----------20----------21
-      const expected = ['20', '21'];
+      // ---10--11
+      // -------20------21
+      // -----------30----------31
+      const expected = ['30', '31'];
       const listener = {
         next: (x: number) => {
           assert.equal(x, expected.shift());

tests/operator/take.ts

@@ -3,13 +3,13 @@ import * as assert from 'assert';
 
 describe('Stream.prototype.take', () => {
   it('should allow specifying max amount to take from input stream', (done) => {
-    const stream = xs.periodic(50).take(4)
+    const stream = xs.periodic(50).take(4);
     const expected = [0, 1, 2, 3];
     let listener = {
       next: (x: number) => {
         assert.equal(x, expected.shift());
       },
-      error: done,
+      error: (err: any) => done(err),
       complete: () => {
         assert.equal(expected.length, 0);
         stream.removeListener(listener);
@@ -18,4 +18,33 @@ describe('Stream.prototype.take', () => {
     };
     stream.addListener(listener);
   });
+
+  it('should not break sibling listeners when TakeOperator tears down', (done) => {
+    const source = xs.periodic(50);
+    const streamA = source.take(3);
+    const streamB = source.take(6);
+    const expectedA = [0, 1, 2];
+    const expectedB = [0, 1, 2, 3, 4, 5];
+
+    streamA.addListener({
+      next: (x: number) => {
+        assert.equal(x, expectedA.shift());
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        assert.equal(expectedA.length, 0);
+      },
+    });
+
+    streamB.addListener({
+      next: (x: number) => {
+        assert.equal(x, expectedB.shift());
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        assert.equal(expectedB.length, 0);
+        done();
+      },
+    });
+  });
 });