fix(Schedulers): ensure schedulers can be reused after error in execu…

…tion

spec/Scheduler-spec.ts

@@ -34,4 +34,25 @@ describe('Scheduler.queue', () => {
       done();
     }, 70);
   });
+
+  it('should be reusable after an error is thrown during execution', (done: DoneSignature) => {
+    const results = [];
+
+    expect(() => {
+      Scheduler.queue.schedule(() => {
+        results.push(1);
+      });
+
+      Scheduler.queue.schedule(() => {
+        throw new Error('bad');
+      });
+    }).toThrow(new Error('bad'));
+
+    setTimeout(() => {
+      Scheduler.queue.schedule(() => {
+        results.push(2);
+        done();
+      });
+    }, 0);
+  });
 });

src/scheduler/Action.ts

@@ -8,4 +8,5 @@ export interface Action extends Subscription {
   schedule(state?: any, delay?: number): void;
   execute(): void;
   scheduler: Scheduler;
+  error: any;
 }

src/scheduler/FutureAction.ts

@@ -8,6 +8,8 @@ export class FutureAction<T> extends Subscription implements Action {
   public id: number;
   public state: T;
   public delay: number;
+  public error: any;
+
   private pending: boolean = false;
 
   constructor(public scheduler: Scheduler,
@@ -17,13 +19,13 @@ export class FutureAction<T> extends Subscription implements Action {
 
   execute() {
     if (this.isUnsubscribed) {
-      throw new Error('How did did we execute a canceled Action?');
+      this.error = new Error('executing a cancelled action');
     } else {
       try {
         this.work(this.state);
       } catch (e) {
         this.unsubscribe();
-        throw e;
+        this.error = e;
       }
     }
   }

src/scheduler/QueueScheduler.ts

@@ -21,6 +21,10 @@ export class QueueScheduler implements Scheduler {
     const actions = this.actions;
     for (let action: QueueAction<any>; action = actions.shift(); ) {
       action.execute();
+      if (action.error) {
+        this.active = false;
+        throw action.error;
+      }
     }
     this.active = false;
   }

src/scheduler/VirtualTimeScheduler.ts

@@ -25,6 +25,11 @@ export class VirtualTimeScheduler implements Scheduler {
       this.frame = action.delay;
       if (this.frame <= maxFrames) {
         action.execute();
+        if (action.error) {
+          actions.length = 0;
+          this.frame = 0;
+          throw action.error;
+        }
       } else {
         break;
       }
@@ -65,6 +70,7 @@ class VirtualAction<T> extends Subscription implements Action {
   state: T;
   delay: number;
   calls = 0;
+  error: any;
 
   constructor(public scheduler: VirtualTimeScheduler,
               public work: (x?: T) => Subscription | void,