Merge refactor

spec/helpers/test-helper.js

@@ -42,9 +42,6 @@ global.it = function (description, cb, timeout) {
       global.rxTestScheduler.flush();
     });
   } else {
-    if (description === 'should work with never and empty') {
-      console.log("TEST");
-    }
     glit.apply(this, arguments);
   }
 };

spec/observables/merge-spec.js

@@ -0,0 +1,102 @@
+/* globals expect, it, describe, hot, cold, expectObservable */
+
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.merge(...observables)', function() {
+  it('should merge cold and cold', function() {
+    var e1 =  cold('---a-----b-----c----|');
+    var e2 =  cold('------x-----y-----z----|');
+    var expected = '---a--x--b--y--c--z----|';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected);
+  });
+
+  it('should merge hot and hot', function() {
+    var e1 =  hot('---a---^-b-----c----|');
+    var e2 =  hot('-----x-^----y-----z----|');
+    var expected =       '--b--y--c--z----|';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected);
+  });
+
+  it('should merge hot and cold', function(){
+    var e1 =  hot('---a-^---b-----c----|');
+    var e2 =  cold(    '--x-----y-----z----|');
+    var expected =     '--x-b---y-c---z----|';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected);
+  });
+
+  it('should merge parallel emissions', function(){
+    var e1 =   hot('---a----b----c----|');
+    var e2 =   hot('---x----y----z----|');
+    var expected = '---(ax)-(by)-(cz)-|';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected);
+  });
+
+  it('should merge empty and empty', function(){
+    var e1 = Observable.empty();
+    var e2 = Observable.empty();
+    expectObservable(Observable.merge(e1, e2)).toBe('|');
+  });
+
+  it('should merge never and empty', function(){
+    var e1 = Observable.never();
+    var e2 = Observable.empty();
+    expectObservable(Observable.merge(e1, e2)).toBe('-');
+  }); 
+
+  it('should merge never and never', function(){
+    var e1 = Observable.never();
+    var e2 = Observable.never();
+    expectObservable(Observable.merge(e1, e2)).toBe('-');
+  });
+
+  it('should merge empty and throw', function(){
+    var e1 = Observable.empty();
+    var e2 = Observable.throw(new Error('blah'));
+    expectObservable(Observable.merge(e1, e2)).toBe('#', undefined, new Error('blah'));
+  });
+
+  it('should merge hot and throw', function(){
+    var e1 = hot('--a--b--c--|');
+    var e2 = Observable.throw(new Error('blah'));
+    expectObservable(Observable.merge(e1, e2)).toBe('#', undefined, new Error('blah'));
+  });
+
+  it('should merge never and throw', function(){
+    var e1 = Observable.never();
+    var e2 = Observable.throw(new Error('blah'));
+    expectObservable(Observable.merge(e1, e2)).toBe('#', undefined, new Error('blah'));
+  });
+
+
+  it('should merge empty and error', function(){
+    var e1 = Observable.empty();
+    var e2 =    hot('-------#', undefined, new Error('blah'));
+    var expected =  '-------#';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected, undefined, new Error('blah'));
+  });
+
+  it('should merge hot and error', function(){
+    var e1 =   hot('--a--b--c--|');
+    var e2 =   hot('-------#', undefined, new Error('blah'));
+    var expected = '--a--b-#';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected, undefined, new Error('blah'));
+  });
+
+  it('should merge never and error', function(){
+    var e1 = Observable.never();
+    var e2 =    hot('-------#', undefined, new Error('blah'));
+    var expected =  '-------#';
+    expectObservable(Observable.merge(e1, e2)).toBe(expected, undefined, new Error('blah'));
+  });
+});
+
+describe('Observable.merge(number, ...observables)', function(){
+  it('should handle concurrency limits', function () {
+    var e1 =  cold('---a---b---c---|');
+    var e2 =  cold('-d---e---f--|');
+    var e3 =  cold(            '---x---y---z---|');
+    var expected = '-d-a-e-b-f-c---x---y---z---|';
+    expectObservable(Observable.merge(e1, e2, e3, 2)).toBe(expected);
+  });
+});

spec/operators/expand-spec.js

@@ -1,29 +1,57 @@
+/* globals describe, it, expect, expectObservable, hot, cold */
 var Rx = require('../../dist/cjs/Rx');
 var Observable = Rx.Observable;
 
 describe('Observable.prototype.expand()', function () {
-  it('should map and recursively flatten', function (done) {
-    var expected = [1, 2, 3, 4, 5];
-    Observable.of(0).expand(function (x) {
-      if (x > 4) {
+  it('should map and recursively flatten', function() {
+    var values = {
+      a: 1,
+      b: 1 + 1, // a + a,
+      c: 2 + 2, // b + b,
+      d: 4 + 4, // c + c,
+      e: 8 + 8, // d + d
+    }
+    var e1 = hot('a', values);
+    /*
+      expectation explanation: (conjunction junction?) ...
+
+      since `cold('---(z|)')` emits `x + x` and completes on frame 30
+      but the next "expanded" return value is synchronously subscribed to in
+      that same frame, it stacks like so:
+
+      a
+      ---(b|)
+         ---(c|)
+            ---(d|)
+               ---(e|)      (...which flattens into:)
+      a--b--c--d--(e|)
+    */
+    var expected = 'a--b--c--d--(e|)';
+
+    expectObservable(e1.expand(function(x) {
+      if(x === 16) {
         return Observable.empty();
       }
-      return Observable.of(x + 1);
-    })
-    .subscribe(function (x) {
-      expect(x).toBe(expected.shift());
-    }, null, done);
+      return cold('---(z|)', { z: x + x });
+    })).toBe(expected, values);
   });
-  it('should map and recursively flatten with ScalarObservables', function (done) {
-    var expected = [1, 2, 3, 4, 5];
-    Observable.of(0).expand(function (x) {
-      if (x > 4) {
+
+  it('should map and recursively flatten with scalars', function() {
+    var values = {
+      a: 1,
+      b: 1 + 1, // a + a,
+      c: 2 + 2, // b + b,
+      d: 4 + 4, // c + c,
+      e: 8 + 8, // d + d
+    }
+    var e1 = hot('a', values);
+    var expected = '(abcde|)';
+
+    expectObservable(e1.expand(function(x) {
+      if(x === 16) {
         return Observable.empty();
       }
-      return Observable.of(x + 1);
-    })
-    .subscribe(function (x) {
-      expect(x).toBe(expected.shift());
-    }, null, done);
+      return Observable.of(x + x); // scalar
+    })).toBe(expected, values);
   });
 });

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;
 
@@ -33,4 +33,12 @@ describe('mergeAll', function () {
       done();
     });
   });
+
+  it('should handle merging 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--db--ec--f---|';
+    expectObservable(e1.mergeAll()).toBe(expected);
+  });
 });

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 () {

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/concat-static.ts

@@ -3,12 +3,13 @@ import Observable from '../Observable';
 import Scheduler from '../Scheduler';
 import immediate from '../schedulers/immediate';
 
-export default function concat<R>(...observables: any[]) : Observable<R> {
-  let scheduler = immediate;
-  const len = observables.length;
-  if(typeof observables[observables.length - 1].schedule === 'function') {
-    scheduler = observables.pop();
-    observables.push(1, scheduler);
+export default function concat<R>(...observables: (Observable<any>|Scheduler)[]) : Observable<R> {
+  let scheduler:Scheduler = immediate;
+  let args = <any[]>observables;
+  const len = args.length;
+  if(typeof (args[observables.length - 1]).schedule === 'function') {
+    scheduler = args.pop();
+    args.push(1, scheduler);
   }
   return merge.apply(this, observables);
 }

src/operators/concat.ts

@@ -1,8 +1,12 @@
 import merge from './merge-static';
 import Observable from '../Observable';
+import Scheduler from '../Scheduler';
 
-export default function concatProto<R>(...observables:any[]) : Observable<R> {
-  observables.unshift(this);
-  observables.push(1);
-  return merge.apply(this, observables);
+export default function concatProto<R>(...observables:(Observable<any>|Scheduler)[]) : Observable<R> {
+  var args = <any[]>observables;
+  args.unshift(this);
+  if(args.length > 1 && typeof args[args.length - 1].schedule === 'function') {
+    args.splice(args.length - 2, 0, 1);
+  }
+  return merge.apply(this, args);
 }

src/operators/concatAll.ts

@@ -1,5 +1,5 @@
-import { MergeOperator } from './merge-support';
+import { MergeAllOperator } from './mergeAll-support';
 
 export default function concatAll() {
-  return this.lift(new MergeOperator(1));
+  return this.lift(new MergeAllOperator(1));
 }

src/operators/concatMapTo.ts

@@ -1,7 +1,7 @@
 import Observable from '../Observable';
 import { FlatMapToOperator } from './flatMapTo-support';
 
-export default function concatMapTo<T, R>(observable: Observable<any>,
-                                          projectResult?: (x: T, y: any, ix: number, iy: number) => R) : Observable<R> {
+export default function concatMapTo<T, R, R2>(observable: Observable<R>,
+                                          projectResult?: (innerValue: R, outerValue: T, innerIndex: number, outerIndex: number) => R2) : Observable<R2> {
   return this.lift(new FlatMapToOperator(observable, projectResult, 1));
-}
+}