fix(schedulers): Queue, Asap, and AnimationFrame Schedulers should be…

… Async if delay > 0

spec/observables/interval-spec.ts

@@ -5,6 +5,9 @@ import * as Rx from '../../dist/cjs/Rx';
 declare const {hot, asDiagram, expectObservable, expectSubscriptions};
 declare const rxTestScheduler: Rx.TestScheduler;
 const Observable = Rx.Observable;
+const asap = Rx.Scheduler.asap;
+const queue = Rx.Scheduler.queue;
+const animationFrame = Rx.Scheduler.animationFrame;
 
 /** @test {interval} */
 describe('Observable.interval', () => {
@@ -66,4 +69,85 @@ describe('Observable.interval', () => {
       done(new Error('should not be called'));
     });
   });
+
+  it('should create an observable emitting periodically with the AsapScheduler', (done: MochaDone) => {
+    const sandbox = sinon.sandbox.create();
+    const fakeTimer = sandbox.useFakeTimers();
+    const interval = 10;
+    const events = [0, 1, 2, 3, 4, 5];
+    const source = Observable.interval(interval, asap).take(6);
+    source.subscribe({
+      next(x) {
+        expect(x).to.equal(events.shift());
+      },
+      error(e) {
+        sandbox.restore();
+        done(e);
+      },
+      complete() {
+        expect(asap.actions.length).to.equal(0);
+        expect(asap.scheduled).to.equal(undefined);
+        sandbox.restore();
+        done();
+      }
+    });
+    let i = -1, n = events.length;
+    while (++i < n) {
+      fakeTimer.tick(interval);
+    }
+  });
+
+  it('should create an observable emitting periodically with the QueueScheduler', (done: MochaDone) => {
+    const sandbox = sinon.sandbox.create();
+    const fakeTimer = sandbox.useFakeTimers();
+    const interval = 10;
+    const events = [0, 1, 2, 3, 4, 5];
+    const source = Observable.interval(interval, queue).take(6);
+    source.subscribe({
+      next(x) {
+        expect(x).to.equal(events.shift());
+      },
+      error(e) {
+        sandbox.restore();
+        done(e);
+      },
+      complete() {
+        expect(queue.actions.length).to.equal(0);
+        expect(queue.scheduled).to.equal(undefined);
+        sandbox.restore();
+        done();
+      }
+    });
+    let i = -1, n = events.length;
+    while (++i < n) {
+      fakeTimer.tick(interval);
+    }
+  });
+
+  it('should create an observable emitting periodically with the AnimationFrameScheduler', (done: MochaDone) => {
+    const sandbox = sinon.sandbox.create();
+    const fakeTimer = sandbox.useFakeTimers();
+    const interval = 10;
+    const events = [0, 1, 2, 3, 4, 5];
+    const source = Observable.interval(interval, animationFrame).take(6);
+    source.subscribe({
+      next(x) {
+        expect(x).to.equal(events.shift());
+      },
+      error(e) {
+        sandbox.restore();
+        done(e);
+      },
+      complete() {
+        expect(animationFrame.actions.length).to.equal(0);
+        expect(animationFrame.scheduled).to.equal(undefined);
+        sandbox.restore();
+        done();
+      }
+    });
+    let i = -1, n = events.length;
+    while (++i < n) {
+      fakeTimer.tick(interval);
+    }
+  });
 });

spec/schedulers/AnimationFrameScheduler-spec.ts

@@ -1,4 +1,5 @@
 import {expect} from 'chai';
+import * as sinon from 'sinon';
 import * as Rx from '../../dist/cjs/Rx';
 
 const animationFrame = Rx.Scheduler.animationFrame;
@@ -9,6 +10,21 @@ describe('Scheduler.animationFrame', () => {
     expect(animationFrame).exist;
   });
 
+  it('should act like the async scheduler if delay > 0', () => {
+    let actionHappened = false;
+    const sandbox = sinon.sandbox.create();
+    const fakeTimer = sandbox.useFakeTimers();
+    animationFrame.schedule(() => {
+      actionHappened = true;
+    }, 50);
+    expect(actionHappened).to.be.false;
+    fakeTimer.tick(25);
+    expect(actionHappened).to.be.false;
+    fakeTimer.tick(25);
+    expect(actionHappened).to.be.true;
+    sandbox.restore();
+  });
+
   it('should schedule an action to happen later', (done: MochaDone) => {
     let actionHappened = false;
     animationFrame.schedule(() => {

spec/schedulers/AsapScheduler-spec.ts

@@ -1,4 +1,5 @@
 import {expect} from 'chai';
+import * as sinon from 'sinon';
 import * as Rx from '../../dist/cjs/Rx';
 
 const asap = Rx.Scheduler.asap;
@@ -9,6 +10,21 @@ describe('Scheduler.asap', () => {
     expect(asap).exist;
   });
 
+  it('should act like the async scheduler if delay > 0', () => {
+    let actionHappened = false;
+    const sandbox = sinon.sandbox.create();
+    const fakeTimer = sandbox.useFakeTimers();
+    asap.schedule(() => {
+      actionHappened = true;
+    }, 50);
+    expect(actionHappened).to.be.false;
+    fakeTimer.tick(25);
+    expect(actionHappened).to.be.false;
+    fakeTimer.tick(25);
+    expect(actionHappened).to.be.true;
+    sandbox.restore();
+  });
+
   it('should schedule an action to happen later', (done: MochaDone) => {
     let actionHappened = false;
     asap.schedule(() => {

spec/schedulers/QueueScheduler-spec.ts

@@ -7,6 +7,21 @@ const queue = Scheduler.queue;
 
 /** @test {Scheduler} */
 describe('Scheduler.queue', () => {
+  it('should act like the async scheduler if delay > 0', () => {
+    let actionHappened = false;
+    const sandbox = sinon.sandbox.create();
+    const fakeTimer = sandbox.useFakeTimers();
+    queue.schedule(() => {
+      actionHappened = true;
+    }, 50);
+    expect(actionHappened).to.be.false;
+    fakeTimer.tick(25);
+    expect(actionHappened).to.be.false;
+    fakeTimer.tick(25);
+    expect(actionHappened).to.be.true;
+    sandbox.restore();
+  });
+
   it('should switch from synchronous to asynchronous at will', () => {
     const sandbox = sinon.sandbox.create();
     const fakeTimer = sandbox.useFakeTimers();

src/scheduler/AnimationFrameAction.ts

@@ -29,8 +29,10 @@ export class AnimationFrameAction<T> extends AsyncAction<T> {
     ));
   }
   protected recycleAsyncId(scheduler: AnimationFrameScheduler, id?: any, delay: number = 0): any {
-    // If delay exists and is greater than 0, recycle as an async action.
-    if (delay !== null && delay > 0) {
+    // If delay exists and is greater than 0, or if the delay is null (the
+    // action wasn't rescheduled) but was originally scheduled as an async
+    // action, then recycle as an async action.
+    if ((delay !== null && delay > 0) || (delay === null && this.delay > 0)) {
       return super.recycleAsyncId(scheduler, id, delay);
     }
     // If the scheduler queue is empty, cancel the requested animation frame and

src/scheduler/AnimationFrameScheduler.ts

@@ -2,7 +2,7 @@ import { AsyncAction } from './AsyncAction';
 import { AsyncScheduler } from './AsyncScheduler';
 
 export class AnimationFrameScheduler extends AsyncScheduler {
-  public flush(): void {
+  public flush(action?: AsyncAction<any>): void {
 
     this.active = true;
     this.scheduled = undefined;
@@ -11,7 +11,7 @@ export class AnimationFrameScheduler extends AsyncScheduler {
     let error: any;
     let index: number = -1;
     let count: number = actions.length;
-    let action: AsyncAction<any> = actions.shift();
+    action = action || actions.shift();
 
     do {
       if (error = action.execute(action.state, action.delay)) {

src/scheduler/AsapAction.ts

@@ -29,8 +29,10 @@ export class AsapAction<T> extends AsyncAction<T> {
     ));
   }
   protected recycleAsyncId(scheduler: AsapScheduler, id?: any, delay: number = 0): any {
-    // If delay exists and is greater than 0, recycle as an async action.
-    if (delay !== null && delay > 0) {
+    // If delay exists and is greater than 0, or if the delay is null (the
+    // action wasn't rescheduled) but was originally scheduled as an async
+    // action, then recycle as an async action.
+    if ((delay !== null && delay > 0) || (delay === null && this.delay > 0)) {
       return super.recycleAsyncId(scheduler, id, delay);
     }
     // If the scheduler queue is empty, cancel the requested microtask and

src/scheduler/AsapScheduler.ts

@@ -2,7 +2,7 @@ import { AsyncAction } from './AsyncAction';
 import { AsyncScheduler } from './AsyncScheduler';
 
 export class AsapScheduler extends AsyncScheduler {
-  public flush(): void {
+  public flush(action?: AsyncAction<any>): void {
 
     this.active = true;
     this.scheduled = undefined;
@@ -11,7 +11,7 @@ export class AsapScheduler extends AsyncScheduler {
     let error: any;
     let index: number = -1;
     let count: number = actions.length;
-    let action: AsyncAction<any> = actions.shift();
+    action = action || actions.shift();
 
     do {
       if (error = action.execute(action.state, action.delay)) {

src/scheduler/QueueAction.ts

@@ -31,8 +31,10 @@ export class QueueAction<T> extends AsyncAction<T> {
   }
 
   protected requestAsyncId(scheduler: QueueScheduler, id?: any, delay: number = 0): any {
-    // If delay is greater than 0, enqueue as an async action.
-    if (delay !== null && delay > 0) {
+    // If delay exists and is greater than 0, or if the delay is null (the
+    // action wasn't rescheduled) but was originally scheduled as an async
+    // action, then recycle as an async action.
+    if ((delay !== null && delay > 0) || (delay === null && this.delay > 0)) {
       return super.requestAsyncId(scheduler, id, delay);
     }
     // Otherwise flush the scheduler starting with this action.