fix(merge): fix issues with async in merge

- moves support types from merge-support to mergeAll-support
- fixes strange issues with async
- adds virtual time tests for merge and mergeAll

Author: benlesh

spec/operators/merge-all-spec.js

@@ -1,4 +1,4 @@
-/* globals describe, it, expect */
+/* globals describe, it, expect, expectObservable, hot, cold */
 var Rx = require('../../dist/cjs/Rx');
 var Observable = Rx.Observable;
 

spec/operators/merge-spec.js

@@ -1,4 +1,4 @@
-/* globals describe, it, expect */
+/* globals describe, it, expect, hot, cold, expectObservable, rxTestScheduler */
 var Rx = require('../../dist/cjs/Rx');
 var Observable = Rx.Observable;
 var immediateScheduler = Rx.Scheduler.immediate;
@@ -23,6 +23,14 @@ describe("Observable.prototype.merge", function () {
       expect(val).toBe(r[i++]);
     }, null, done);
   });
+  
+  
+  it('should handle merging two hot observables', function (){
+    var e1 =    hot('--a-----b-----c----|');
+    var e2 =    hot('-----d-----e-----f---|');
+    var expected =  '--a--d--b--e--c--f---|';
+    expectObservable(e1.merge(e2, rxTestScheduler)).toBe(expected);
+  });
 });
 
 describe('Observable.prototype.mergeAll', function () {

src/operators/merge-static.ts

@@ -1,7 +1,7 @@
 import Scheduler from '../Scheduler';
 import Observable from '../Observable';
 import ArrayObservable from '../observables/ArrayObservable';
-import { MergeOperator } from './merge-support';
+import { MergeAllOperator } from './mergeAll-support';
 import immediate from '../schedulers/immediate';
 
 export default function merge<R>(...observables: (Observable<any>|Scheduler|number)[]): Observable<R> {
@@ -20,6 +20,6 @@ export default function merge<R>(...observables: (Observable<any>|Scheduler|numb
   if(observables.length === 1) {
     return <Observable<R>>observables[0];
   }
-
-  return new ArrayObservable(observables, scheduler).lift(new MergeOperator(concurrent));
+  
+  return new ArrayObservable(observables, scheduler).lift(new MergeAllOperator(concurrent));
 }

src/operators/merge-support.ts

@@ -1,7 +1,8 @@
 import Observable from '../Observable';
 import mergeStatic from './merge-static';
+import Scheduler from '../Scheduler';
 
-export default function merge<R>(...observables: (Observable<any>|number)[]): Observable<R> {
+export default function merge<R>(...observables: (Observable<any>|Scheduler|number)[]): Observable<R> {
   observables.unshift(this);
   return mergeStatic.apply(this, observables);
 }

src/operators/merge.ts

@@ -0,0 +1,68 @@
+import Observable from '../Observable';
+import Operator from '../Operator';
+import Subscriber from '../Subscriber';
+import Observer from '../Observer';
+import Subscription from '../Subscription';
+
+export class MergeAllOperator<T, R> implements Operator<T, R> {
+  constructor(private concurrent: number) {
+    
+  }
+  
+  call(observer: Observer<T>) {
+    return new MergeAllSubscriber(observer, this.concurrent);
+  }
+}
+
+export class MergeAllSubscriber<T> extends Subscriber<T> {
+  private hasCompleted: boolean = false;
+  private buffer: Observable<any>[] = [];
+  private active: number = 0;
+  constructor(destination: Observer<T>, private concurrent:number) {
+    super(destination);
+  }
+  
+  _next(observable: any) {
+    if(this.active < this.concurrent) {
+      if(observable._isScalar) {
+        this.destination.next(observable.value);
+      } else {
+        const innerSub = new Subscription();
+        this.add(innerSub);
+        this.active++;
+        innerSub.add(observable.subscribe(new MergeAllInnerSubscriber(this.destination, this, innerSub)));
+      }
+    } else {
+      this.buffer.push(observable);
+    }
+  }
+  
+  _complete() {
+    this.hasCompleted = true;
+    if(this.active === 0 && this.buffer.length === 0) {
+      this.destination.complete();
+    }
+  }
+  
+  notifyComplete(innerSub: Subscription<T>) {
+    const buffer = this.buffer;
+    this.remove(innerSub);
+    this.active--;
+    if(buffer.length > 0) {
+      this._next(buffer.shift());
+    } else if (this.active === 0 && this.hasCompleted) {
+      this.destination.complete();
+    }
+  }
+}
+
+export class MergeAllInnerSubscriber<T> extends Subscriber<T> {
+  constructor(destination: Observer<T>, private parent: MergeAllSubscriber<T>, 
+    private innerSub: Subscription<T> ) {
+    super(destination);
+  }
+  
+  _complete() {
+    this.parent.notifyComplete(this.innerSub);
+  }
+}

src/operators/mergeAll-support.ts

@@ -3,66 +3,8 @@ import Operator from '../Operator';
 import Subscriber from '../Subscriber';
 import Observer from '../Observer';
 import Subscription from '../Subscription';
+import { MergeAllOperator } from './mergeAll-support';
 
 export default function mergeAll<R>(concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
   return this.lift(new MergeAllOperator(concurrent));
-}
-
-class MergeAllOperator<T, R> implements Operator<T, R> {
-  constructor(private concurrent: number) {
-    
-  }
-  
-  call(observer: Observer<T>) {
-    return new MergeAllSubscriber(observer, this.concurrent);
-  }
-}
-
-class MergeAllSubscriber<T> extends Subscriber<T> {
-  private hasCompleted: boolean = false;
-  private buffer: Observable<any>[] = [];
-  private active: number = 0;
-  constructor(destination: Observer<T>, private concurrent:number) {
-    super(destination);
-  }
-  
-  _next(value: any) {
-    if(this.active < this.concurrent) {
-      const innerSub = new Subscription();
-      this.add(innerSub);
-      this.active++;
-      innerSub.add(value.subscribe(new MergeAllInnerSubscriber(this.destination, this, innerSub)));
-    } else {
-      this.buffer.push(value);
-    }
-  }
-  
-  _complete() {
-    this.hasCompleted = true;
-    if(this.active === 0 && this.buffer.length === 0) {
-      this.destination.complete();
-    }
-  }
-  
-  notifyComplete(innerSub: Subscription<T>) {
-    const buffer = this.buffer;
-    this.remove(innerSub);
-    this.active--;
-    if(buffer.length > 0) {
-      this._next(buffer.shift());
-    } else if (this.active === 0 && this.hasCompleted) {
-      this.destination.complete();
-    }
-  }
-}
-
-class MergeAllInnerSubscriber<T> extends Subscriber<T> {
-  constructor(destination: Observer<T>, private parent: MergeAllSubscriber<T>, 
-    private innerSub: Subscription<T> ) {
-    super(destination);
-  }
-  
-  _complete() {
-    this.parent.notifyComplete(this.innerSub);
-  }
 }