fix(imitate): fix imitate() isomorphism

Adding the first listener (which starts the stream) to the mimic stream should be equivalent to
adding the first listener to the target stream. This commit adds a test and fixes imitate() for that
use case.

src/core.ts

@@ -904,11 +904,13 @@ export class Stream<T> implements InternalListener<T> {
   protected _hil: InternalListener<T>; // 'hil' = Hidden Internal Listener
   protected _stopID: any = empty;
   protected _prod: InternalProducer<T>;
+  protected _target: Stream<T>; // imitation target if this Stream will imitate
 
   constructor(producer?: InternalProducer<T>) {
     this._prod = producer;
     this._ils = [];
     this._hil = null;
+    this._target = null;
   }
 
   _n(t: T): void {
@@ -980,6 +982,10 @@ export class Stream<T> implements InternalListener<T> {
   }
 
   _add(il: InternalListener<T>): void {
+    const ta = this._target;
+    if (ta && ta._ils.length === 0) {
+      return ta._add(il);
+    }
     const a = this._ils;
     a.push(il);
     if (a.length === 1) {
@@ -1001,6 +1007,8 @@ export class Stream<T> implements InternalListener<T> {
       if (p && a.length <= 0) {
         this._stopID = setTimeout(() => p._stop());
       }
+    } else if (this._target) {
+      this._target._remove(il);
     }
   }
 
@@ -1666,16 +1674,17 @@ export class Stream<T> implements InternalListener<T> {
    * represents an event stream, and that would be a candidate for creating a
    * proxy Stream which then imitates the target Stream.
    *
-   * @param {Stream} other The stream to imitate on the current one. Must not be
-   * a MemoryStream.
+   * @param {Stream} target The other stream to imitate on the current one. Must
+   * not be a MemoryStream.
    */
-  imitate(other: Stream<T>): void {
-    if (other instanceof MemoryStream) {
+  imitate(target: Stream<T>): void {
+    if (target instanceof MemoryStream) {
       throw new Error('A MemoryStream was given to imitate(), but it only ' +
       'supports a Stream. Read more about this restriction here: ' +
       'https://github.com/staltz/xstream#faq');
     }
-    other._setHIL(this);
+    this._target = target;
+    target._setHIL(this);
   }
 
   /**

tests/operator/imitate.ts

@@ -6,10 +6,10 @@ import * as assert from 'assert';
 
 describe('Stream.prototype.imitate', () => {
   it('should be able to model a circular dependency in the stream graph', (done) => {
-    const fakeSecond = xs.create<number>();
-    const first = fakeSecond.map(x => x * 10).take(3);
+    const secondMimic = xs.create<number>();
+    const first = secondMimic.map(x => x * 10).take(3);
     const second = first.map(x => x + 1).startWith(1).compose(delay<number>(1));
-    fakeSecond.imitate(second);
+    secondMimic.imitate(second);
     const expected = [1, 11, 111, 1111];
 
     second.addListener({
@@ -24,6 +24,25 @@ describe('Stream.prototype.imitate', () => {
     });
   });
 
+  it('should be able to model a circular dependency, mimic subscribed', (done) => {
+    const secondMimic = xs.create<number>();
+    const first = secondMimic.map(x => x * 10).take(3);
+    const second = first.map(x => x + 1).startWith(1).compose(delay<number>(1));
+    secondMimic.imitate(second);
+    const expected = [1, 11, 111, 1111];
+
+    secondMimic.addListener({
+      next: (x: number) => {
+        assert.equal(x, expected.shift());
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        assert.equal(expected.length, 0);
+        done();
+      },
+    });
+  });
+
   it('should broadcast the source stream to multiple listeners', (done) => {
     const fakeSecond = xs.create<number>();
     const first = fakeSecond.map(x => x * 10).take(3);