fix(schedulers): fix asap and animationFrame schedulers to execute ac…

…ross async boundaries. (#1820)

The AsapScheduler and AnimationFrameSchedulers were totally busted. My bad. Now they execute their scheduled actions in batches. If actions reschedule while executing a batch, a new frame is requested for the rescheduled action to execute in.

This PR also simplifies the public `Scheduler` and `Action` APIs. Implementation details like the `actions` queue and `active` boolean are now on the concrete implementations, so it's easier for people to implement the Scheduler API. This PR also renames `FutureAction` -> `AsyncAction` to conform to the same naming convention as the rest of the Action types.

Fixes #1814

package.json

@@ -94,6 +94,7 @@
  "prepublish": "shx rm -rf ./typings && typings install && npm run build_all",
  "publish_docs": "./publish_docs.sh",
  "test_mocha": "mocha --opts spec/support/default.opts spec-js",
+ "debug_mocha": "node-debug _mocha --opts spec/support/debug.opts spec-js",
  "test_browser": "npm-run-all build_spec_browser && opn spec/support/mocha-browser-runner.html",
  "test": "npm-run-all clean_spec build_spec test_mocha clean_spec",
  "tests2png": "npm run build_spec && mkdirp tmp/docs/img && mkdirp spec-js/support && shx cp spec/support/*.opts spec-js/support/ && mocha --opts spec/support/tests2png.opts spec-js",

spec/Scheduler-spec.ts

@@ -19,6 +19,21 @@ describe('Scheduler.queue', () => {
  expect(call2).to.be.true;
  });
 
+ it('should schedule things recursively via this.schedule', () => {
+ let call1 = false;
+ let call2 = false;
+ Scheduler.queue.active = false;
+ Scheduler.queue.schedule(function (state) {
+ call1 = state.call1;
+ call2 = state.call2;
+ if (!call2) {
+ this.schedule({ call1: true, call2: true });
+ }
+ }, 0, { call1: true, call2: false });
+ expect(call1).to.be.true;
+ expect(call2).to.be.true;
+ });
+
  it('should schedule things in the future too', (done: MochaDone) => {
  let called = false;
  Scheduler.queue.schedule(() => {
@@ -55,4 +70,4 @@ describe('Scheduler.queue', () => {
  });
  }, 0);
  });
-});
+});

spec/schedulers/AnimationFrameScheduler-spec.ts

@@ -0,0 +1,88 @@
+import {expect} from 'chai';
+import * as Rx from '../../dist/cjs/Rx';
+
+const animationFrame = Rx.Scheduler.animationFrame;
+
+/** @test {Scheduler} */
+describe('Scheduler.animationFrame', () => {
+ it('should exist', () => {
+ expect(animationFrame).exist;
+ });
+
+ it('should schedule an action to happen later', (done: MochaDone) => {
+ let actionHappened = false;
+ animationFrame.schedule(() => {
+ actionHappened = true;
+ done();
+ });
+ if (actionHappened) {
+ done(new Error('Scheduled action happened synchronously'));
+ }
+ });
+
+ it('should execute recursively scheduled actions in separate asynchronous contexts', (done: MochaDone) => {
+ let syncExec1 = true;
+ let syncExec2 = true;
+ animationFrame.schedule(function (index) {
+ if (index === 0) {
+ this.schedule(1);
+ animationFrame.schedule(() => { syncExec1 = false; });
+ } else if (index === 1) {
+ this.schedule(2);
+ animationFrame.schedule(() => { syncExec2 = false; });
+ } else if (index === 2) {
+ this.schedule(3);
+ } else if (index === 3) {
+ if (!syncExec1 && !syncExec2) {
+ done();
+ } else {
+ done(new Error('Execution happened synchronously.'));
+ }
+ }
+ }, 0, 0);
+ });
+
+ it('should cancel the animation frame if all scheduled actions unsubscribe before it executes', (done: MochaDone) => {
+ let animationFrameExec1 = false;
+ let animationFrameExec2 = false;
+ const action1 = animationFrame.schedule(() => { animationFrameExec1 = true; });
+ const action2 = animationFrame.schedule(() => { animationFrameExec2 = true; });
+ expect(animationFrame.scheduled).to.exist;
+ expect(animationFrame.actions.length).to.equal(2);
+ action1.unsubscribe();
+ action2.unsubscribe();
+ expect(animationFrame.actions.length).to.equal(0);
+ expect(animationFrame.scheduled).to.equal(undefined);
+ animationFrame.schedule(() => {
+ expect(animationFrameExec1).to.equal(false);
+ expect(animationFrameExec2).to.equal(false);
+ done();
+ });
+ });
+
+ it('should execute the rest of the scheduled actions if the first action is canceled', (done: MochaDone) => {
+ let actionHappened = false;
+ let firstSubscription = null;
+ let secondSubscription = null;
+
+ firstSubscription = animationFrame.schedule(() => {
+ actionHappened = true;
+ if (secondSubscription) {
+ secondSubscription.unsubscribe();
+ }
+ done(new Error('The first action should not have executed.'));
+ });
+
+ secondSubscription = animationFrame.schedule(() => {
+ if (!actionHappened) {
+ done();
+ }
+ });
+
+ if (actionHappened) {
+ done(new Error('Scheduled action happened synchronously'));
+ } else {
+ firstSubscription.unsubscribe();
+ }
+ });
+});

spec/schedulers/AsapScheduler-spec.ts

@@ -20,6 +20,46 @@ describe('Scheduler.asap', () => {
  }
  });
 
+ it('should execute recursively scheduled actions in separate asynchronous contexts', (done: MochaDone) => {
+ let syncExec1 = true;
+ let syncExec2 = true;
+ asap.schedule(function (index) {
+ if (index === 0) {
+ this.schedule(1);
+ asap.schedule(() => { syncExec1 = false; });
+ } else if (index === 1) {
+ this.schedule(2);
+ asap.schedule(() => { syncExec2 = false; });
+ } else if (index === 2) {
+ this.schedule(3);
+ } else if (index === 3) {
+ if (!syncExec1 && !syncExec2) {
+ done();
+ } else {
+ done(new Error('Execution happened synchronously.'));
+ }
+ }
+ }, 0, 0);
+ });
+
+ it('should cancel the setImmediate if all scheduled actions unsubscribe before it executes', (done: MochaDone) => {
+ let asapExec1 = false;
+ let asapExec2 = false;
+ const action1 = asap.schedule(() => { asapExec1 = true; });
+ const action2 = asap.schedule(() => { asapExec2 = true; });
+ expect(asap.scheduled).to.exist;
+ expect(asap.actions.length).to.equal(2);
+ action1.unsubscribe();
+ action2.unsubscribe();
+ expect(asap.actions.length).to.equal(0);
+ expect(asap.scheduled).to.equal(undefined);
+ asap.schedule(() => {
+ expect(asapExec1).to.equal(false);
+ expect(asapExec2).to.equal(false);
+ done();
+ });
+ });
+
  it('should execute the rest of the scheduled actions if the first action is canceled', (done: MochaDone) => {
  let actionHappened = false;
  let firstSubscription = null;

spec/schedulers/QueueScheduler-spec.ts

@@ -0,0 +1,56 @@
+import {expect} from 'chai';
+import * as Rx from '../../dist/cjs/Rx';
+
+const Scheduler = Rx.Scheduler;
+const queue = Scheduler.queue;
+
+/** @test {Scheduler} */
+describe('Scheduler.queue', () => {
+ it('should switch from synchronous to asynchronous at will', (done: MochaDone) => {
+ let lastExecTime = 0;
+ let asyncExec = false;
+ queue.schedule(function (index) {
+ if (index === 0) {
+ lastExecTime = queue.now();
+ this.schedule(1, 100);
+ } else if (index === 1) {
+ if (queue.now() - lastExecTime < 100) {
+ done(new Error('Execution happened synchronously.'));
+ } else {
+ asyncExec = true;
+ lastExecTime = queue.now();
+ this.schedule(2, 0);
+ }
+ } else if (index === 2) {
+ if (asyncExec === false) {
+ done(new Error('Execution happened synchronously.'));
+ } else {
+ done();
+ }
+ }
+ }, 0, 0);
+ asyncExec = false;
+ });
+ it('should unsubscribe the rest of the scheduled actions if an action throws an error', () => {
+ const actions = [];
+ let action2Exec = false;
+ let action3Exec = false;
+ let errorValue = undefined;
+ try {
+ queue.schedule(() => {
+ actions.push(
+ queue.schedule(() => { throw new Error('oops'); }),
+ queue.schedule(() => { action2Exec = true; }),
+ queue.schedule(() => { action3Exec = true; })
+ );
+ });
+ } catch (e) {
+ errorValue = e;
+ }
+ expect(actions.every((action) => action.isUnsubscribed)).to.be.true;
+ expect(action2Exec).to.be.false;
+ expect(action3Exec).to.be.false;
+ expect(errorValue).exist;
+ expect(errorValue.message).to.equal('oops');
+ });
+});

spec/support/debug.opts

@@ -8,7 +8,7 @@
 --bail
 --full-trace
 --check-leaks
---globals WebSocket,FormData
+--globals WebSocket,FormData,XDomainRequest,ActiveXObject
 
 --recursive
 --timeout 100000

src/MiscJSDoc.ts

@@ -8,8 +8,6 @@
 import {Subscriber} from './Subscriber';
 import {TeardownLogic} from './Subscription';
 import {Observable} from './Observable';
-import {Subscription} from './Subscription';
-import {Action} from './scheduler/Action';
 import './scheduler/MiscJSDoc';
 import './observable/dom/MiscJSDoc';
 
@@ -130,90 +128,3 @@ export class ObserverDoc<T> {
  return void 0;
  }
 }
-
-/**
- * An execution context and a data structure to order tasks and schedule their
- * execution. Provides a notion of (potentially virtual) time, through the
- * `now()` getter method.
- *
- * Each unit of work in a Scheduler is called an {@link Action}.
- *
- * ```ts
- * interface Scheduler {
- * now(): number;
- * schedule(work, delay?, state?): Subscription;
- * flush(): void;
- * active: boolean;
- * actions: Action[];
- * scheduledId: number;
- * }
- * ```
- *
- * @interface
- * @name Scheduler
- * @noimport true
- */
-export class SchedulerDoc {
- /**
- * A getter method that returns a number representing the current time
- * (at the time this function was called) according to the scheduler's own
- * internal clock.
- * @return {number} A number that represents the current time. May or may not
- * have a relation to wall-clock time. May or may not refer to a time unit
- * (e.g. milliseconds).
- */
- now(): number {
- return 0;
- }
-
- /**
- * Schedules a function, `work`, for execution. May happen at some point in
- * the future, according to the `delay` parameter, if specified. May be passed
- * some context object, `state`, which will be passed to the `work` function.
- *
- * The given arguments will be processed an stored as an Action object in a
- * queue of actions.
- *
- * @param {function(state: ?T): ?Subscription} work A function representing a
- * task, or some unit of work to be executed by the Scheduler.
- * @param {number} [delay] Time to wait before executing the work, where the
- * time unit is implicit and defined by the Scheduler itself.
- * @param {T} [state] Some contextual data that the `work` function uses when
- * called by the Scheduler.
- * @return {Subscription} A subscription in order to be able to unsubscribe
- * the scheduled work.
- */
- schedule<T>(work: (state?: T) => Subscription | void, delay?: number, state?: T): Subscription {
- return void 0;
- }
-
- /**
- * Prompt the Scheduler to execute all of its queued actions, therefore
- * clearing its queue.
- * @return {void}
- */
- flush(): void {
- return void 0;
- }
-
- /**
- * A flag to indicate whether the Scheduler is currently executing a batch of
- * queued actions.
- * @type {boolean}
- */
- active: boolean = false;
-
- /**
- * The queue of scheduled actions as an array.
- * @type {Action[]}
- */
- actions: Action<any>[] = [];
-
- /**
- * An internal ID used to track the latest asynchronous task such as those
- * coming from `setTimeout`, `setInterval`, `requestAnimationFrame`, and
- * others.
- * @type {number}
- */
- scheduledId: number = 0;
-}

src/Rx.ts

@@ -184,6 +184,8 @@ import observable from 'symbol-observable';
  * asynchronous conversions.
  * @property {Scheduler} async Schedules work with `setInterval`. Use this for
  * time-based operations.
+ * @property {Scheduler} animationFrame Schedules work with `requestAnimationFrame`.
+ * Use this for synchronizing with the platform's painting
  */
 let Scheduler = {
  asap,