Pr 2135

package.json

@@ -89,7 +89,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",
+ "debug_mocha": "node --inspect --debug-brk ./node_modules/.bin/_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/operators/timeout-spec.ts

@@ -139,4 +139,27 @@ describe('Observable.prototype.timeout', () => {
  expectObservable(result).toBe(expected, values, value);
  expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
+
+ it('should unsubscribe from the scheduled timeout action when timeout is unsubscribed early', () => {
+ const e1 = hot('--a--b--c---d--e--|');
+ const e1subs = '^ ! ';
+ const expected = '--a--b--c-- ';
+ const unsub = ' ! ';
+
+ const result = e1
+ .lift(function(source) {
+ const timeoutSubscriber = this;
+ const { action } = timeoutSubscriber; // get a ref to the action here
+ timeoutSubscriber.add(() => { // because it'll be null by the
+ if (!action.closed) { // time we get into this function.
+ throw new Error('TimeoutSubscriber scheduled action wasn\'t canceled');
+ }
+ });
+ return source._subscribe(timeoutSubscriber);
+ })
+ .timeout(50, null, rxTestScheduler);
+
+ expectObservable(result, unsub).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
 });

spec/operators/timeoutWith-spec.ts

@@ -260,4 +260,30 @@ describe('Observable.prototype.timeoutWith', () => {
  expectSubscriptions(e1.subscriptions).toBe(e1subs);
  expectSubscriptions(e2.subscriptions).toBe(e2subs);
  });
+
+ it('should unsubscribe from the scheduled timeout action when timeout is unsubscribed early', () => {
+ const e1 = hot('---a---b-----c----|');
+ const e1subs = '^ ! ';
+ const e2 = cold( '-x---y| ');
+ const e2subs = ' ^ ! ';
+ const expected = '---a---b----x-- ';
+ const unsub = ' ! ';
+
+ const result = e1
+ .lift(function(source) {
+ const timeoutSubscriber = this;
+ const { action } = timeoutSubscriber; // get a ref to the action here
+ timeoutSubscriber.add(() => { // because it'll be null by the
+ if (!action.closed) { // time we get into this function.
+ throw new Error('TimeoutSubscriber scheduled action wasn\'t canceled');
+ }
+ });
+ return source._subscribe(timeoutSubscriber);
+ })
+ .timeoutWith(40, e2, rxTestScheduler);
+
+ expectObservable(result, unsub).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ expectSubscriptions(e2.subscriptions).toBe(e2subs);
+ });
 });

spec/schedulers/VirtualTimeScheduler-spec.ts

@@ -80,4 +80,15 @@ describe('VirtualTimeScheduler', () => {
  v.flush();
  expect(count).to.equal(3);
  });
-});
+
+ it('should not execute virtual actions that have been rescheduled before flush', () => {
+ const v = new VirtualTimeScheduler();
+ let messages = [];
+ let action: VirtualAction<string> = <VirtualAction<string>> v.schedule(function(state: string) {
+ messages.push(state);
+ }, 10, 'first message');
+ action = <VirtualAction<string>> action.schedule('second message' , 10);
+ v.flush();
+ expect(messages).to.deep.equal(['second message']);
+ });
+});

src/operator/timeout.ts

@@ -1,3 +1,4 @@
+import { Action } from '../scheduler/Action';
 import { async } from '../scheduler/async';
 import { isDate } from '../util/isDate';
 import { Operator } from '../Operator';
@@ -44,15 +45,8 @@ class TimeoutOperator<T> implements Operator<T, T> {
  * @extends {Ignored}
  */
 class TimeoutSubscriber<T> extends Subscriber<T> {
- private index: number = 0;
- private _previousIndex: number = 0;
- get previousIndex(): number {
- return this._previousIndex;
- }
- private _hasCompleted: boolean = false;
- get hasCompleted(): boolean {
- return this._hasCompleted;
- }
+
+ private action: Action<TimeoutSubscriber<T>> = null;
 
  constructor(destination: Subscriber<T>,
  private absoluteTimeout: boolean,
@@ -63,40 +57,36 @@ class TimeoutSubscriber<T> extends Subscriber<T> {
  this.scheduleTimeout();
  }
 
- private static dispatchTimeout(state: any): void {
- const source = state.subscriber;
- const currentIndex = state.index;
- if (!source.hasCompleted && source.previousIndex === currentIndex) {
- source.notifyTimeout();
- }
+ private static dispatchTimeout<T>(subscriber: TimeoutSubscriber<T>): void {
+ subscriber.error(subscriber.errorToSend);
  }
 
  private scheduleTimeout(): void {
- let currentIndex = this.index;
- this.scheduler.schedule(TimeoutSubscriber.dispatchTimeout, this.waitFor, { subscriber: this, index: currentIndex });
- this.index++;
- this._previousIndex = currentIndex;
+ const { action } = this;
+ if (action) {
+ // Recycle the action if we've already scheduled one. All the production
+ // Scheduler Actions mutate their state/delay time and return themeselves.
+ // VirtualActions are immutable, so they create and return a clone. In this
+ // case, we need to set the action reference to the most recent VirtualAction,
+ // to ensure that's the one we clone from next time.
+ this.action = (<Action<TimeoutSubscriber<T>>> action.schedule(this, this.waitFor));
+ } else {
+ this.add(this.action = (<Action<TimeoutSubscriber<T>>> this.scheduler.schedule(
+ TimeoutSubscriber.dispatchTimeout, this.waitFor, this
+ )));
+ }
  }
 
  protected _next(value: T): void {
- this.destination.next(value);
-
  if (!this.absoluteTimeout) {
  this.scheduleTimeout();
  }
+ super._next(value);
  }
 
- protected _error(err: any): void {
- this.destination.error(err);
- this._hasCompleted = true;
- }
-
- protected _complete(): void {
- this.destination.complete();
- this._hasCompleted = true;
- }
-
- notifyTimeout(): void {
- this.error(this.errorToSend);
+ protected _unsubscribe() {
+ this.action = null;
+ this.scheduler = null;
+ this.errorToSend = null;
  }
 }

src/operator/timeoutWith.ts

@@ -1,8 +1,9 @@
+import { Action } from '../scheduler/Action';
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { Scheduler } from '../Scheduler';
 import { async } from '../scheduler/async';
-import { Subscription, TeardownLogic } from '../Subscription';
+import { TeardownLogic } from '../Subscription';
 import { Observable, ObservableInput } from '../Observable';
 import { isDate } from '../util/isDate';
 import { OuterSubscriber } from '../OuterSubscriber';
@@ -49,65 +50,52 @@ class TimeoutWithOperator<T> implements Operator<T, T> {
  * @extends {Ignored}
  */
 class TimeoutWithSubscriber<T, R> extends OuterSubscriber<T, R> {
- private timeoutSubscription: Subscription = undefined;
- private index: number = 0;
- private _previousIndex: number = 0;
- get previousIndex(): number {
- return this._previousIndex;
- }
- private _hasCompleted: boolean = false;
- get hasCompleted(): boolean {
- return this._hasCompleted;
- }
 
- constructor(public destination: Subscriber<T>,
+ private action: Action<TimeoutWithSubscriber<T, R>> = null;
+
+ constructor(destination: Subscriber<T>,
  private absoluteTimeout: boolean,
  private waitFor: number,
  private withObservable: ObservableInput<any>,
  private scheduler: Scheduler) {
- super();
- destination.add(this);
+ super(destination);
  this.scheduleTimeout();
  }
 
- private static dispatchTimeout(state: any): void {
- const source = state.subscriber;
- const currentIndex = state.index;
- if (!source.hasCompleted && source.previousIndex === currentIndex) {
-  source.handleTimeout();
- }
+ private static dispatchTimeout<T, R>(subscriber: TimeoutWithSubscriber<T, R>): void {
+ const { withObservable } = subscriber;
+ subscriber.unsubscribe();
+ subscriber.closed = false;
+ subscriber.isStopped = false;
+ subscriber.add(subscribeToResult(subscriber, withObservable));
  }
 
  private scheduleTimeout(): void {
- let currentIndex = this.index;
- const timeoutState = { subscriber: this, index: currentIndex };
- this.scheduler.schedule(TimeoutWithSubscriber.dispatchTimeout, this.waitFor, timeoutState);
- this.index++;
- this._previousIndex = currentIndex;
+ const { action } = this;
+ if (action) {
+ // Recycle the action if we've already scheduled one. All the production
+ // Scheduler Actions mutate their state/delay time and return themeselves.
+ // VirtualActions are immutable, so they create and return a clone. In this
+ // case, we need to set the action reference to the most recent VirtualAction,
+ // to ensure that's the one we clone from next time.
+ this.action = (<Action<TimeoutWithSubscriber<T, R>>> action.schedule(this, this.waitFor));
+ } else {
+ this.add(this.action = (<Action<TimeoutWithSubscriber<T, R>>> this.scheduler.schedule(
+ TimeoutWithSubscriber.dispatchTimeout, this.waitFor, this
+ )));
+ }
  }
 
- protected _next(value: T) {
- this.destination.next(value);
+ protected _next(value: T): void {
  if (!this.absoluteTimeout) {
  this.scheduleTimeout();
  }
+ super._next(value);
  }
 
- protected _error(err: any) {
- this.destination.error(err);
- this._hasCompleted = true;
- }
-
- protected _complete() {
- this.destination.complete();
- this._hasCompleted = true;
- }
-
- handleTimeout(): void {
- if (!this.closed) {
- const withObservable = this.withObservable;
- this.unsubscribe();
- this.destination.add(this.timeoutSubscription = subscribeToResult(this, withObservable));
- }
+ protected _unsubscribe() {
+ this.action = null;
+ this.scheduler = null;
+ this.withObservable = null;
  }
 }

src/scheduler/VirtualTimeScheduler.ts

@@ -46,6 +46,8 @@ export class VirtualTimeScheduler extends AsyncScheduler {
  */
 export class VirtualAction<T> extends AsyncAction<T> {
 
+ protected active: boolean = true;
+
  constructor(protected scheduler: VirtualTimeScheduler,
  protected work: (this: VirtualAction<T>, state?: T) => void,
  protected index: number = scheduler.index += 1) {
@@ -54,8 +56,11 @@ export class VirtualAction<T> extends AsyncAction<T> {
  }
 
  public schedule(state?: T, delay: number = 0): Subscription {
- return !this.id ?
- super.schedule(state, delay) : (
+ if (!this.id) {
+ return super.schedule(state, delay);
+ }
+ this.active = false;
+ return (
  // If an action is rescheduled, we save allocations by mutating its state,
  // pushing it to the end of the scheduler queue, and recycling the action.
  // But since the VirtualTimeScheduler is used for testing, VirtualActions
@@ -77,6 +82,12 @@ export class VirtualAction<T> extends AsyncAction<T> {
  return undefined;
  }
 
+ protected _execute(state: T, delay: number): any {
+ if (this.active === true) {
+ return super._execute(state, delay);
+ }
+ }
+
  public static sortActions<T>(a: VirtualAction<T>, b: VirtualAction<T>) {
  if (a.delay === b.delay) {
  if (a.index === b.index) {