feat(testing): Add testScheduler.run() helper

spec/schedulers/TestScheduler-spec.ts

@@ -1,7 +1,8 @@
 import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions, time } from '../helpers/marble-testing';
 import { TestScheduler } from 'rxjs/testing';
-import { Observable, NEVER, EMPTY, Subject, of, Notification } from 'rxjs';
+import { Observable, NEVER, EMPTY, Subject, of, merge, Notification } from 'rxjs';
+import { delay, debounceTime } from 'rxjs/operators';
 
 declare const rxTestScheduler: TestScheduler;
 
@@ -255,4 +256,90 @@ describe('TestScheduler', () => {
       });
     });
   });
+
+  describe('TestScheduler.run()', () => {
+    const assertDeepEquals = (actual: any, expected: any) => {
+      expect(actual).deep.equal(expected);
+    };
+
+    it('should provide the correct helpers', () => {
+      const testScheduler = new TestScheduler(assertDeepEquals);
+
+      testScheduler.run(({ cold, hot, flush, expectObservable, expectSubscriptions }) => {
+        expect(cold).to.be.a('function');
+        expect(hot).to.be.a('function');
+        expect(flush).to.be.a('function');
+        expect(expectObservable).to.be.a('function');
+        expect(expectSubscriptions).to.be.a('function');
+
+        const obs1 = cold('-a-c-e|');
+        const obs2 = hot('-^-b-d-f|');
+        const output = merge(obs1, obs2);
+        const expected =  '-abcdef|';
+
+        expectObservable(output).toBe(expected);
+        expectSubscriptions(obs1.subscriptions).toBe('^-----!');
+        expectSubscriptions(obs2.subscriptions).toBe('^------!');
+      });
+    });
+
+    it('should make operators that use AsyncScheduler automatically use TestScheduler for actual scheduling', () => {
+      const testScheduler = new TestScheduler(assertDeepEquals);
+
+      testScheduler.run(({ cold, expectObservable }) => {
+        const output = cold('-a-b-(c|)').pipe(debounceTime(20), delay(10));
+        const expected =    '------(c|)';
+        expectObservable(output).toBe(expected);
+      });
+    });
+
+    it('should flush automatically', () => {
+      const testScheduler = new TestScheduler((actual, expected) => {
+        expect(actual).deep.equal(expected);
+      });
+      testScheduler.run(({ cold, expectObservable }) => {
+        const output = cold('-a-b-(c|)').pipe(debounceTime(20), delay(10));
+        const expected = '------(c|)';
+        expectObservable(output).toBe(expected);
+
+        expect(testScheduler['flushTests'].length).to.equal(1);
+        expect(testScheduler['actions'].length).to.equal(1);
+      });
+
+      expect(testScheduler['flushTests'].length).to.equal(0);
+      expect(testScheduler['actions'].length).to.equal(0);
+    });
+
+    it('should support explicit flushing', () => {
+      const testScheduler = new TestScheduler(assertDeepEquals);
+
+      testScheduler.run(({ cold, expectObservable, flush }) => {
+        const output = cold('-a-b-(c|)').pipe(debounceTime(20), delay(10));
+        const expected = '------(c|)';
+        expectObservable(output).toBe(expected);
+
+        expect(testScheduler['flushTests'].length).to.equal(1);
+        expect(testScheduler['actions'].length).to.equal(1);
+
+        flush();
+
+        expect(testScheduler['flushTests'].length).to.equal(0);
+        expect(testScheduler['actions'].length).to.equal(0);
+      });
+
+      expect(testScheduler['flushTests'].length).to.equal(0);
+      expect(testScheduler['actions'].length).to.equal(0);
+    });
+
+    it('should pass-through return values, e.g. Promises', (done) => {
+      const testScheduler = new TestScheduler(assertDeepEquals);
+
+      testScheduler.run(() => {
+        return Promise.resolve('foo');
+      }).then(value => {
+        expect(value).to.equal('foo');
+        done();
+      });
+    });
+  });
 });

src/internal/scheduler/AsyncScheduler.ts

@@ -1,7 +1,11 @@
 import { Scheduler } from '../Scheduler';
+import { Action } from './Action';
 import { AsyncAction } from './AsyncAction';
+import { SchedulerAction } from '../types';
+import { Subscription } from '../Subscription';
 
 export class AsyncScheduler extends Scheduler {
+  public static delegate?: Scheduler;
   public actions: Array<AsyncAction<any>> = [];
   /**
    * A flag to indicate whether the Scheduler is currently executing a batch of
@@ -17,6 +21,25 @@ export class AsyncScheduler extends Scheduler {
    */
   public scheduled: any = undefined;
 
+  constructor(SchedulerAction: typeof Action,
+              now: () => number = Scheduler.now) {
+    super(SchedulerAction, () => {
+      if (AsyncScheduler.delegate && AsyncScheduler.delegate !== this) {
+        return AsyncScheduler.delegate.now();
+      } else {
+        return now();
+      }
+    });
+  }
+
+  public schedule<T>(work: (this: SchedulerAction<T>, state?: T) => void, delay: number = 0, state?: T): Subscription {
+    if (AsyncScheduler.delegate && AsyncScheduler.delegate !== this) {
+      return AsyncScheduler.delegate.schedule(work, delay, state);
+    } else {
+      return super.schedule(work, delay, state);
+    }
+  }
+
   public flush(action: AsyncAction<any>): void {
 
     const {actions} = this;

src/internal/scheduler/VirtualTimeScheduler.ts

@@ -38,6 +38,10 @@ export class VirtualTimeScheduler extends AsyncScheduler {
       throw error;
     }
   }
+
+  public schedule<T>(work: (this: SchedulerAction<T>, state?: T) => void, delay: number = 0, state?: T): Subscription {
+    return new VirtualAction<T>(this, work).schedule(state, delay);
+  }
 }
 
 /**

src/internal/testing/TestScheduler.ts

@@ -6,9 +6,18 @@ import { TestMessage } from './TestMessage';
 import { SubscriptionLog } from './SubscriptionLog';
 import { Subscription } from '../Subscription';
 import { VirtualTimeScheduler, VirtualAction } from '../scheduler/VirtualTimeScheduler';
+import { AsyncScheduler } from '../scheduler/AsyncScheduler';
 
 const defaultMaxFrame: number = 750;
 
+interface RunHelpers {
+  cold: typeof TestScheduler.prototype.createColdObservable;
+  hot: typeof TestScheduler.prototype.createHotObservable;
+  flush: typeof TestScheduler.prototype.flush;
+  expectObservable: typeof TestScheduler.prototype.expectObservable;
+  expectSubscriptions: typeof TestScheduler.prototype.expectSubscriptions;
+}
+
 interface FlushableTest {
   ready: boolean;
   actual?: any[];
@@ -129,10 +138,16 @@ export class TestScheduler extends VirtualTimeScheduler {
     }
 
     super.flush();
-    const readyFlushTests = this.flushTests.filter(test => test.ready);
-    while (readyFlushTests.length > 0) {
-      const test = readyFlushTests.shift();
-      this.assertDeepEqual(test.actual, test.expected);
+    const { flushTests } = this;
+    const flushTestsCopy = flushTests.slice();
+
+    for (let i = 0, l = flushTests.length; i < l; i++) {
+      const test = flushTestsCopy[i];
+      if (test.ready) {
+        // remove it from the original array, not our copy
+        flushTests.splice(i, 1);
+        this.assertDeepEqual(test.actual, test.expected);
+      }
     }
   }
 
@@ -243,4 +258,20 @@ export class TestScheduler extends VirtualTimeScheduler {
     }
     return testMessages;
   }
+
+  run<T>(callback: (helpers: RunHelpers) => T): T {
+    AsyncScheduler.delegate = this;
+    const helpers = {
+      cold: this.createColdObservable.bind(this),
+      hot: this.createHotObservable.bind(this),
+      flush: this.flush.bind(this),
+      expectObservable: this.expectObservable.bind(this),
+      expectSubscriptions: this.expectSubscriptions.bind(this),
+    };
+    const ret = callback(helpers);
+    this.flush();
+    AsyncScheduler.delegate = undefined;
+
+    return ret;
+  }
 }