refactor(Action): move delay setting to Action to support Virtual Tim…

…e Scheduling

spec/schedulers/VirtualTimeScheduler-spec.js

@@ -24,8 +24,6 @@ describe('VirtualTimeScheduler', function() {
     expect(invoked).toEqual([1, 2, 3, 4, 5]);
   });
 
-
-
   it('should schedule things in order when flushed if each this is scheduled at random', function () {
     var v = new VirtualTimeScheduler();
     var invoked = [];
@@ -43,4 +41,19 @@ describe('VirtualTimeScheduler', function() {
 
     expect(invoked).toEqual([1, 3, 5, 2, 6, 4]);
   });
+
+  it('should support recursive scheduling', function () {
+    var v = new VirtualTimeScheduler();
+    var count = 0;
+    var expected = [100, 200, 300];
+
+    v.schedule(function (state) {
+      if (++count === 3) return;
+      expect(this.delay).toBe(expected.shift());
+      this.schedule(state, this.delay)
+    }, 100, 'test');
+
+    v.flush();
+    expect(count).toBe(3);
+  });
 });

src/observables/IntervalObservable.ts

@@ -11,17 +11,17 @@ export default class IntervalObservable<T> extends Observable<T> {
 
   static dispatch(state) {
 
-    const { index, subscriber } = state;
+    const { index, subscriber, period } = state;
 
     subscriber.next(index);
 
     if (subscriber.isUnsubscribed) {
       return;
     }
 
-    state.index = index + 1;
+    state.index += 1;
 
-    (<any> this).schedule(state);
+    (<any> this).schedule(state, period);
   }
 
   constructor(private period: number = 0, private scheduler: Scheduler = nextTick) {
@@ -41,7 +41,7 @@ export default class IntervalObservable<T> extends Observable<T> {
     const scheduler = this.scheduler;
 
     subscriber.add(scheduler.schedule(IntervalObservable.dispatch, period, {
-      index, subscriber
+      index, subscriber, period
     }));
   }
 }

src/observables/TimerObservable.ts

@@ -29,7 +29,7 @@ export default class TimerObservable<T> extends Observable<T> {
       }));
     } else {
       state.index = index + 1;
-      action.schedule(state);
+      action.schedule(state, period);
     }
   }
 

src/operators/bufferTime.ts

@@ -35,7 +35,7 @@ class BufferTimeSubscriber<T> extends Subscriber<T> {
       this.add(scheduler.schedule(dispatchBufferClose, bufferTimeSpan, { subscriber: this, buffer }));
       this.add(scheduler.schedule(dispatchBufferCreation, bufferCreationInterval, { bufferTimeSpan, bufferCreationInterval, subscriber: this, scheduler }));
     } else {
-      this.add(scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, { subscriber: this, buffer }));
+      this.add(scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, { subscriber: this, buffer, bufferTimeSpan }));
     }
   }
 
@@ -82,15 +82,15 @@ function dispatchBufferTimeSpanOnly(state) {
   }
 
   state.buffer = subscriber.openBuffer();
-  (<any>this).schedule(state);
+  (<any>this).schedule(state, state.bufferTimeSpan);
 }
 
 function dispatchBufferCreation(state) {
-  let { bufferTimeSpan, subscriber, scheduler } = state;
+  let {  bufferCreationInterval, bufferTimeSpan, subscriber, scheduler } = state;
   let buffer = subscriber.openBuffer();
   var action = <Action>this;
   action.add(scheduler.schedule(dispatchBufferClose, bufferTimeSpan, { subscriber, buffer }));
-  action.schedule(state);
+  action.schedule(state, bufferCreationInterval);
 }
 
 function dispatchBufferClose({ subscriber, buffer }) {

src/operators/delay.ts

@@ -41,8 +41,8 @@ class DelaySubscriber<T> extends Subscriber<T> {
       queue.shift().notification.observe(destination);
     }
     if (queue.length > 0) {
-      (<any> this).delay = Math.max(0, queue[0].time - scheduler.now());
-      (<any> this).schedule(state);
+      let delay = Math.max(0, queue[0].time - scheduler.now());
+      (<any> this).schedule(state, delay);
     } else {
       source.active = false;
     }

src/operators/sampleTime.ts

@@ -25,7 +25,7 @@ class SampleTimeSubscriber<T> extends Subscriber<T> {
 
   constructor(destination: Observer<T>, private delay: number, private scheduler: Scheduler) {
     super(destination);
-    this.add(scheduler.schedule(dispatchNotification, delay, { subscriber: this }));
+    this.add(scheduler.schedule(dispatchNotification, delay, { subscriber: this, delay }));
   }
 
   _next(value: T) {
@@ -40,7 +40,8 @@ class SampleTimeSubscriber<T> extends Subscriber<T> {
   }
 }
 
-function dispatchNotification<T>(state: { subscriber: SampleTimeSubscriber<T> }) {
-  state.subscriber.notifyNext();
-  (<any>this).schedule(state);
+function dispatchNotification<T>(state) {
+  let { subscriber, delay } = state;
+  subscriber.notifyNext();
+  (<any>this).schedule(state, delay);
 }

src/operators/windowTime.ts

@@ -37,7 +37,7 @@ class WindowTimeSubscriber<T> extends Subscriber<T> {
       this.add(scheduler.schedule(dispatchWindowCreation, windowCreationInterval, { windowTimeSpan, windowCreationInterval, subscriber: this, scheduler }))
     } else {
       let window = this.openWindow();
-      this.add(scheduler.schedule(dispatchWindowTimeSpanOnly, windowTimeSpan, { subscriber: this, window }));
+      this.add(scheduler.schedule(dispatchWindowTimeSpanOnly, windowTimeSpan, { subscriber: this, window, windowTimeSpan }));
     }
   }
 
@@ -79,25 +79,22 @@ class WindowTimeSubscriber<T> extends Subscriber<T> {
   }
 }
 
-function dispatchWindowTimeSpanOnly(state) {
-  const subscriber: WindowTimeSubscriber<any> = state.subscriber;
-
-  const prevWindow: Subject<any> = state.window;
-  if (prevWindow) {
-    prevWindow.complete();
+function dispatchWindowTimeSpanOnly<T>(state: { window: Subject<any>, windowTimeSpan: number, subscriber: WindowTimeSubscriber<T>}) {
+  const { subscriber, windowTimeSpan, window } = state;
+  if (window) {
+    window.complete();
   }
-
-  let window = subscriber.openWindow();
-  (<any>this).schedule({ subscriber, window });
+  state.window = subscriber.openWindow();
+  (<any>this).schedule(state, windowTimeSpan);
 }
 
 function dispatchWindowCreation(state) {
-  let { windowTimeSpan, subscriber, scheduler } = state;
+  let { windowTimeSpan, subscriber, scheduler, windowCreationInterval } = state;
   let window = subscriber.openWindow();
   let action = <Action>this;
   let context = { action, subscription: null };
   action.add(context.subscription = scheduler.schedule(dispatchWindowClose, windowTimeSpan, { subscriber, window, context }));
-  action.schedule(state);
+  action.schedule(state, windowCreationInterval);
 }
 
 function dispatchWindowClose({ subscriber, window, context }) {

src/schedulers/Action.ts

@@ -1,11 +1,13 @@
 import Subscription from '../Subscription';
+import Scheduler from '../Scheduler';
 
 interface Action extends Subscription<any> {
   work: (state?: any) => void|Subscription<any>
-  state: any;
+  state?: any;
   delay?: number;
-  schedule(state: any);
+  schedule(state?: any, delay?: number);
   execute(): void;
+  scheduler: Scheduler;
 }
 
 export default Action;

src/schedulers/FutureAction.ts

@@ -6,20 +6,20 @@ import ImmediateAction from './ImmediateAction';
 export default class FutureAction<T> extends ImmediateAction<T> {
 
   id: any;
-
+  delay: number;
+
   constructor(public scheduler: ImmediateScheduler,
-              public work: (x?: any) => Subscription<T> | void,
-              public delay: number) {
+              public work: (x?: any) => Subscription<T> | void) {
     super(scheduler, work);
   }
 
-  schedule(state?:any): Action {
+  schedule(state?:any, delay:number = 0): Action {
     if (this.isUnsubscribed) {
       return this;
     }
 
+    this.delay = delay;
     this.state = state;
-
     const id = this.id;
 
     if (id != null) {

src/schedulers/ImmediateScheduler.ts

@@ -37,6 +37,6 @@ export default class ImmediateScheduler implements Scheduler {
   }
 
   scheduleLater<T>(work: (x?: any) => Subscription<T> | void, delay: number, state?: any): Action {
-    return new FutureAction(this, work, delay).schedule(state);
+    return new FutureAction(this, work).schedule(state, delay);
   }
 }

src/schedulers/VirtualTimeScheduler.ts

@@ -8,7 +8,7 @@ export default class VirtualTimeScheduler implements Scheduler {
   scheduled: boolean = false;
   index: number = 0;
   sorted: boolean = false;
-  frame: number = -1;
+  frame: number = 0;
 
   now() {
     return 0;
@@ -25,38 +25,41 @@ export default class VirtualTimeScheduler implements Scheduler {
 
   flush() {
     this.sortActions();
-    this.actions.forEach((action, frame) => {
-      this.frame = frame;
+    const actions = this.actions;
+    while (actions.length > 0) {
+      let action = actions.shift();
+      this.frame = action.delay;
       action.execute();
-    });
-    this.actions.length = 0;
-    this.frame = -1;
+    }
+    this.frame = 0;
   }
 
   schedule<T>(work: (x?: any) => Subscription<T> | void, delay: number = 0, state?: any): Subscription<T> {
     this.sorted = false;
-    return new VirtualAction(this, work, delay, this.index++).schedule(state);
+    return new VirtualAction(this, work, this.index++).schedule(state, delay);
   }
 }
 
 class VirtualAction<T> extends Subscription<T> implements Action {
   state: any;
-
+  delay: number;
+
   constructor(public scheduler: VirtualTimeScheduler,
     public work: (x?: any) => Subscription<T> | void,
-    public delay: number,
     public index: number) {
     super();
   }
 
-  schedule(state?: any): VirtualAction<T> {
+  schedule(state?: any, delay: number = 0): VirtualAction<T> {
     if (this.isUnsubscribed) {
       return this;
     }
-
-    this.state = state;
     const scheduler = this.scheduler;
-    scheduler.actions.push(this);
+    var action = scheduler.frame === this.delay ? this :
+      new VirtualAction(scheduler, this.work, scheduler.index += 1);
+    action.state = state;
+    action.delay = scheduler.frame + delay;
+    scheduler.actions.push(action);
     return this;
   }
 

src/util/isScheduler.ts

@@ -0,0 +1,3 @@
+export default function isScheduler(value: any): boolean {
+  return value && typeof value.schedule === 'function';
+}