feat(TestScheduler): add TestScheduler

adds a test scheduler that features a marble parser as well as
cold and hot observable creation methods

closes #270

spec/schedulers/TestScheduler-spec.js

@@ -0,0 +1,67 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var TestScheduler = Rx.TestScheduler;
+var Notification = Rx.Notification;
+
+describe('TestScheduler', function() {
+  it('should exist', function () {
+    expect(typeof TestScheduler).toBe('function');
+  });
+
+  describe('parseMarbles()', function () {
+    it('should parse a marble string into a series of notifications and types', function () {
+      var result = TestScheduler.parseMarbles('-------a---b---|', { a: 'A', b: 'B' });
+      expect(result).toDeepEqual([
+        { frame: 70, notification: Notification.createNext('A') },
+        { frame: 110, notification: Notification.createNext('B') },
+        { frame: 150, notification: Notification.createComplete() }
+      ]);
+    });
+
+    it('should parse a marble string with a subscription point', function () {
+      var result = TestScheduler.parseMarbles('---^---a---b---|', { a: 'A', b: 'B' });
+      expect(result).toDeepEqual([
+        { frame: 40, notification: Notification.createNext('A') },
+        { frame: 80, notification: Notification.createNext('B') },
+        { frame: 120, notification: Notification.createComplete() }
+      ]);
+    });
+
+    it('should parse a marble string with an error', function () {
+      var result = TestScheduler.parseMarbles('-------a---b---#', { a: 'A', b: 'B' }, 'omg error!');
+      expect(result).toDeepEqual([
+        { frame: 70, notification: Notification.createNext('A') },
+        { frame: 110, notification: Notification.createNext('B') },
+        { frame: 150, notification: Notification.createError('omg error!') }
+      ]);
+    });
+  }); 
+
+  describe('createColdObservable()', function () {
+    it('should create a cold observable', function () {
+      var expected = ['A', 'B'];
+      var scheduler = new TestScheduler();
+      var source = scheduler.createColdObservable('--a---b--|', { a: 'A', b: 'B' });
+      expect(source instanceof Rx.Observable).toBe(true);
+      source.subscribe(function (x) {
+        expect(x).toBe(expected.shift());
+      });
+      scheduler.flush();
+      expect(expected.length).toBe(0);
+    });
+  });
+
+  describe('createHotObservable()', function () {
+    it('should create a cold observable', function () {
+      var expected = ['A', 'B'];
+      var scheduler = new TestScheduler();
+      var source = scheduler.createHotObservable('--a---b--|', { a: 'A', b: 'B' });
+      expect(source instanceof Rx.Subject).toBe(true);
+      source.subscribe(function (x) {
+        expect(x).toBe(expected.shift());
+      });
+      scheduler.flush();
+      expect(expected.length).toBe(0);
+    });
+  });
+});

src/Rx.ts

@@ -2,6 +2,7 @@ import Subject from './Subject';
 import ImmediateScheduler from './schedulers/ImmediateScheduler';
 import NextTickScheduler from './schedulers/NextTickScheduler';
 import VirtualTimeScheduler from './schedulers/VirtualTimeScheduler';
+import TestScheduler from './schedulers/TestScheduler';
 import immediate from './schedulers/immediate';
 import nextTick from './schedulers/nextTick';
 import Observable from './Observable';
@@ -229,5 +230,6 @@ export {
     BehaviorSubject,
     ConnectableObservable,
     Notification,
-    VirtualTimeScheduler
+    VirtualTimeScheduler,
+    TestScheduler
 };

src/schedulers/TestScheduler.ts

@@ -0,0 +1,66 @@
+import Observable from '../Observable';
+import VirtualTimeScheduler from './VirtualTimeScheduler';
+import Notification from '../Notification';
+import Subject from '../Subject';
+
+export default class TestScheduler extends VirtualTimeScheduler {
+  createColdObservable(marbles: string, values?: any, error?: any) {
+    if (marbles.indexOf('^') !== -1) {
+      throw new Error('cold observable cannot have subscription offset "^"');
+    }
+    let messages = TestScheduler.parseMarbles(marbles, values, error);
+    return Observable.create(subscriber => {
+      messages.forEach(({ notification, frame }) => {
+        this.schedule(() => {
+          notification.observe(subscriber);
+        }, frame);
+      });
+    });
+  }
+
+  createHotObservable(marbles: string, values?: any, error?: any) {
+    let messages = TestScheduler.parseMarbles(marbles, values, error);
+    let subject = new Subject();
+    messages.forEach(({ notification, frame }) => {
+      this.schedule(() => {
+        notification.observe(subject);
+      }, frame);
+    });
+    return subject;
+  }
+
+  static parseMarbles(marbles: string, values?: any, errorValue?: any) : ({ notification: Notification<any>, frame: number })[] {
+    let len = marbles.length;
+    let results: ({ notification: Notification<any>, frame: number })[] = [];
+    let subIndex = marbles.indexOf('^');
+    let frameOffset = subIndex === -1 ? 0 : (subIndex * -10);
+
+    for (let i = 0; i < len; i++) {
+      let frame = i * 10;
+      let notification;
+      let c = marbles[i];
+      switch (c) {
+        case '-':
+          break;
+        case '|':
+          notification = Notification.createComplete();
+          break;
+        case '^':  
+          break;
+        case '#':
+          notification = Notification.createError(errorValue || 'error');
+          break;
+        default:
+          notification = Notification.createNext(values[c]);
+          break;
+      }
+
+      frame += frameOffset;
+
+      if (notification) {
+        results.push({ notification, frame });
+      }
+    }
+    return results;
+  }
+}

src/schedulers/VirtualTimeScheduler.ts

@@ -8,6 +8,7 @@ export default class VirtualTimeScheduler implements Scheduler {
   scheduled: boolean = false;
   index: number = 0;
   sorted: boolean = false;
+  frame: number = -1;
 
   now() {
     return 0;
@@ -24,10 +25,12 @@ export default class VirtualTimeScheduler implements Scheduler {
 
   flush() {
     this.sortActions();
-    this.actions.forEach(action => {
+    this.actions.forEach((action, frame) => {
+      this.frame = frame;
       action.execute();
     });
     this.actions.length = 0;
+    this.frame = -1;
   }
 
   schedule<T>(work: (x?: any) => Subscription<T> | void, delay: number = 0, state?: any): Subscription<T> {