fix(delay): unsubscribe scheduled action

spec/operators/delay-spec.ts

@@ -1,3 +1,4 @@
+import { expect } from 'chai';
 import * as Rx from '../../dist/package/Rx';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
 
@@ -147,4 +148,20 @@ describe('Observable.prototype.delay', () => {
 
     expectObservable(result).toBe(expected);
   });
+
+  it('should unsubscribe scheduled action when result is unsubscribed explicitly', () => {
+    const e1 =  cold('---ab---|');
+    const t =   time(    '---|');
+    const expected = '------';
+    const unsub =    '-----!';
+    const subs =     '^    !';
+
+    const result = e1.delay(t, rxTestScheduler);
+
+    expectObservable(result, unsub).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+    rxTestScheduler.maxFrames = 50;
+    rxTestScheduler.flush();
+    expect(rxTestScheduler.actions).to.have.length(0);
+  });
 });

src/operators/delay.ts

@@ -7,7 +7,7 @@ import { Action } from '../scheduler/Action';
 import { Notification } from '../Notification';
 import { Observable } from '../Observable';
 import { PartialObserver } from '../Observer';
-import { TeardownLogic } from '../Subscription';
+import { TeardownLogic, Subscription } from '../Subscription';
 import { MonoTypeOperatorFunction } from '../interfaces';
 
 /**
@@ -79,7 +79,7 @@ interface DelayState<T> {
  */
 class DelaySubscriber<T> extends Subscriber<T> {
   private queue: Array<DelayMessage<T>> = [];
-  private active: boolean = false;
+  private nextAction: Subscription = null;
   private errored: boolean = false;
 
   private static dispatch<T>(this: Action<DelayState<T>>, state: DelayState<T>): void {
@@ -97,7 +97,7 @@ class DelaySubscriber<T> extends Subscriber<T> {
       this.schedule(state, delay);
     } else {
       this.unsubscribe();
-      source.active = false;
+      source.nextAction = null;
     }
   }
 
@@ -108,8 +108,7 @@ class DelaySubscriber<T> extends Subscriber<T> {
   }
 
   private _schedule(scheduler: IScheduler): void {
-    this.active = true;
-    scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch, this.delay, {
+    this.nextAction = scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch, this.delay, {
       source: this, destination: this.destination, scheduler: scheduler
     });
   }
@@ -123,7 +122,7 @@ class DelaySubscriber<T> extends Subscriber<T> {
     const message = new DelayMessage(scheduler.now() + this.delay, notification);
     this.queue.push(message);
 
-    if (this.active === false) {
+    if (this.nextAction === null) {
       this._schedule(scheduler);
     }
   }
@@ -141,6 +140,14 @@ class DelaySubscriber<T> extends Subscriber<T> {
   protected _complete() {
     this.scheduleNotification(Notification.createComplete());
   }
+
+  unsubscribe(): void {
+    if (this.nextAction !== null) {
+      this.nextAction.unsubscribe();
+      this.nextAction = null;
+    }
+    super.unsubscribe();
+  }
 }
 
 class DelayMessage<T> {

src/scheduler/VirtualTimeScheduler.ts

@@ -24,7 +24,15 @@ export class VirtualTimeScheduler extends AsyncScheduler {
     const {actions, maxFrames} = this;
     let error: any, action: AsyncAction<any>;
 
-    while ((action = actions.shift()) && (this.frame = action.delay) <= maxFrames) {
+    while (actions.length > 0) {
+      action = actions[0];
+      if (action.delay > maxFrames) {
+        this.frame = maxFrames;
+        break;
+      }
+
+      actions.shift();
+      this.frame = action.delay;
       if (error = action.execute(action.state, action.delay)) {
         break;
       }