fix(core): teardown and stop producer before complete/error

To fix bugs such as #91 and others where adding .debug() would magically solve the bug, we need to
change the semantics around producer stop and stream completion/error. Previously, a stream would
complete first, and stop the producer after. Now, the stream stops its producer just before
completing. This was important to make the test 'Stream.prototype.flatten with map should not run
multiple executions of the inner' pass.

BREAKING CHANGE: in this version, when a stream completes or errors, its producer has already been
stopped. In previous versions, the stream first completes, propagates the complete to other
listeners and operators, and then its producer is stopped. You may barely notice this breaking
change when updating your code. Most existing code will still work like before.

src/core.ts

@@ -1085,21 +1085,21 @@ export class Stream<T> implements InternalListener<T> {
     this._err = err;
     const a = this._ils;
     const L = a.length;
+    this._x();
     if (L == 1) a[0]._e(err); else {
       const b = copy(a);
       for (let i = 0; i < L; i++) b[i]._e(err);
     }
-    this._x();
   }
 
   _c(): void {
     const a = this._ils;
     const L = a.length;
+    this._x();
     if (L == 1) a[0]._c(); else {
       const b = copy(a);
       for (let i = 0; i < L; i++) b[i]._c();
     }
-    this._x();
   }
 
   _x(): void { // tear down logic, after error or complete
@@ -1122,11 +1122,11 @@ export class Stream<T> implements InternalListener<T> {
     if (ta !== NO) return ta._add(il);
     const a = this._ils;
     a.push(il);
-    if (a.length === 1) {
-      if (this._stopID !== NO) {
-        clearTimeout(this._stopID);
-        this._stopID = NO;
-      }
+    if (a.length > 1) return;
+    if (this._stopID !== NO) {
+      clearTimeout(this._stopID);
+      this._stopID = NO;
+    } else {
       const p = this._prod;
       if (p !== NO) p._start(this);
     }

tests/factory/merge.ts

@@ -41,8 +41,7 @@ describe('xs.merge', () => {
 
   it('should complete properly when stopped asynchronously and restarted synchronously', (done) => {
     const initial = xs.of('foo');
-    // debug here needed to stop MergeProducer asynchronously via _remove
-    const stream = xs.merge(initial).debug(() => {});
+    const stream = xs.merge(initial);
 
     const noop = () => {};
     stream.addListener({next: noop, error: noop, complete: noop});

tests/memoryStream.ts

@@ -24,7 +24,7 @@ describe('MemoryStream', () => {
 
   it('should be createable giving a custom producer object', (done) => {
     const expected = [10, 20, 30];
-    let listenerGotEnd: boolean = false;
+    let producerStopped: boolean = false;
 
     const stream = xs.createWithMemory({
       start(listener: Listener<number>) {
@@ -37,7 +37,8 @@ describe('MemoryStream', () => {
       stop() {
         done();
         assert.equal(expected.length, 0);
-        assert.equal(listenerGotEnd, true);
+        assert.equal(producerStopped, false);
+        producerStopped = true;
       },
     });
 
@@ -47,7 +48,7 @@ describe('MemoryStream', () => {
       },
       error: (err: any) => done(err),
       complete: () => {
-        listenerGotEnd = true;
+        assert.equal(producerStopped, true);
       },
     });
   });

tests/operator/flatten.ts

@@ -263,6 +263,24 @@ describe('Stream.prototype.flatten', () => {
         }
       });
     });
+
+    it('should not run multiple executions of the inner', (done) => {
+      const inner = xs.periodic(350).take(2).map(i => i * 100);
+      const outer = xs.periodic(200).take(4);
+      const stream = outer.map(() => inner).flatten();
+
+      const expected = [0, 100];
+      stream.addListener({
+        next: (x: number) => {
+          assert.equal(x, expected.shift());
+        },
+        error: (err: any) => done(err),
+        complete: () => {
+          assert.equal(expected.length, 0);
+          done();
+        }
+      });
+    });
   });
 
   describe('with filter+map fusion', () => {

tests/stream.ts

@@ -40,7 +40,7 @@ describe('Stream', () => {
 
   it('should be createable giving a custom producer object', (done) => {
     const expected = [10, 20, 30];
-    let listenerGotEnd: boolean = false;
+    let producerStopped: boolean = false;
 
     const producer: Producer<number> = {
       start(listener: Listener<number>) {
@@ -53,7 +53,8 @@ describe('Stream', () => {
       stop() {
         done();
         assert.equal(expected.length, 0);
-        assert.equal(listenerGotEnd, true);
+        assert.equal(producerStopped, false);
+        producerStopped = true;
       },
     };
 
@@ -64,7 +65,7 @@ describe('Stream', () => {
       },
       error: (err: any) => done(err),
       complete: () => {
-        listenerGotEnd = true;
+        assert.equal(producerStopped, true);
       },
     });
   });
@@ -218,7 +219,7 @@ describe('Stream', () => {
       },
       error: (err: any) => done(err),
       complete: () => {
-        assert.equal(on, true);
+        assert.equal(on, false);
         assert.equal(expected1.length, 0);
       },
     });
@@ -232,7 +233,7 @@ describe('Stream', () => {
       },
       error: (err: any) => done(err),
       complete: () => {
-        assert.equal(on, true);
+        assert.equal(on, false);
         assert.equal(expected2.length, 0);
       },
     });
@@ -265,7 +266,7 @@ describe('Stream', () => {
       },
       error: (err: any) => {
         assert.equal(err, 'oops');
-        assert.equal(on, true);
+        assert.equal(on, false);
         assert.equal(expected1.length, 0);
       },
       complete: () => {
@@ -282,7 +283,7 @@ describe('Stream', () => {
       },
       error: (err: any) => {
         assert.equal(err, 'oops');
-        assert.equal(on, true);
+        assert.equal(on, false);
         assert.equal(expected2.length, 0);
       },
       complete: () => {