fix(switchAll): switch all will properly handle async observables

spec/operators/switchAll-spec.js

@@ -1,4 +1,4 @@
-/* expect, it, describe */
+/* expect, it, describe, expectObserable, hot, cold */
 var Rx = require('../../dist/cjs/Rx');
 
 var Observable = Rx.Observable;
@@ -26,4 +26,12 @@ describe('Observable.prototype.switchAll()', function(){
       expect(x).toBe(r[i++]);
     }, null, done);
   });
+
+  it('should handle a hot observable of observables', function() {
+    var x = cold(        '--a---b---c--|');
+    var y = cold(                '---d--e---f---|');
+    var e1 = hot(  '------x-------y------|', { x: x, y: y });
+    var expected = '--------a---b----d--e---f---|';
+    expectObservable(e1.switchAll()).toBe(expected);
+  });
 });

src/operators/switchAll.ts

@@ -19,31 +19,55 @@ class SwitchOperator<T, R> implements Operator<T, R> {
   }
 }
 
-class SwitchSubscriber<T, R> extends MergeSubscriber<T, R> {
-
+class SwitchSubscriber<T> extends Subscriber<T> {
+  private active: number = 0;
+  private hasCompleted: boolean = false;
   innerSubscription: Subscription<T>;
 
-  constructor(destination: Observer<R>) {
-    super(destination, 1);
+  constructor(destination: Observer<T>) {
+    super(destination);
   }
-
-  _buffer(value) {
-    const active = this.active;
-    if(active > 0) {
-      this.active = active - 1;
-      const inner = this.innerSubscription;
-      if(inner) {
-        inner.unsubscribe()
-        this.innerSubscription = null;
-      }
+
+  _next(value: any) {
+    this.active++;
+    this.unsubscribeInner();
+    this.add(this.innerSubscription = value.subscribe(new InnerSwitchSubscriber(this.destination, this))); 
+  }
+
+  _complete() {
+    this.hasCompleted = true;
+    if(this.active === 0) {
+      this.destination.complete();
+    }
+  }
+
+  unsubscribeInner() {
+    const innerSubscription = this.innerSubscription;
+    if(innerSubscription) {
+      this.active--;
+      innerSubscription.unsubscribe();
+      this.remove(innerSubscription);
     }
-    this._next(value);
   }
+
+  notifyComplete() {
+    this.unsubscribeInner();
+    if(this.hasCompleted && this.active === 0) {
+      this.destination.complete();
+    }
+  }
+}
 
-  _subscribeInner(observable, value, index) {
-    this.innerSubscription = new Subscription();
-    this.innerSubscription.add(super._subscribeInner(observable, value, index));
-    return this.innerSubscription;
+class InnerSwitchSubscriber<T> extends Subscriber<T> {
+  constructor(destination: Observer<any>, private parent: SwitchSubscriber<T>) {
+    super(destination);
+  }
+
+  _next(value: T) {
+    super._next(value);
+  }
+  _complete() {
+    this.parent.notifyComplete();
   }
 }