fix(elementAt): force unsubscribe when it completes or errors

Force unsubscription logic in elementAt, when resulting Observable
completes or errors, so that all resources are disposed as fast
as possible, even if next operator in chain does not unsubscribe
reliably when given complete or error notification.

BREAKING CHANGE: unsubscription cadence has changed for `elementAt`,
this means side-effects related to unsubscription may occur at a
different time than in previous versions.

spec/operators/elementAt-spec.ts

@@ -1,6 +1,7 @@
 import {expect} from 'chai';
 import * as Rx from '../../dist/cjs/Rx';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
+import { doNotUnsubscribe } from '../helpers/doNotUnsubscribe';
 
 declare const { asDiagram };
 declare const hot: typeof marbleTestingSignature.hot;
@@ -126,4 +127,33 @@ describe('Observable.prototype.elementAt', () => {
  expectObservable((<any>source).elementAt(3, defaultValue)).toBe(expected, { x: defaultValue });
  expectSubscriptions(source.subscriptions).toBe(subs);
  });
+
+ it('should unsubscribe from source Observable, even if destination does not unsubscribe', () => {
+ const source = hot('--a--b--c-d---|');
+ const subs = '^ ! ';
+ const expected = '--------(c|) ';
+
+ expectObservable((<any>source).elementAt(2).let(doNotUnsubscribe)).toBe(expected);
+ expectSubscriptions(source.subscriptions).toBe(subs);
+ });
+
+ it('should unsubscribe from source if index is out of range, even if destination does not unsubscribe', () => {
+ const source = hot('--a--|');
+ const subs = '^ !';
+ const expected = '-----#';
+
+ expectObservable((<any>source).elementAt(3).let(doNotUnsubscribe))
+ .toBe(expected, null, new Rx.ArgumentOutOfRangeError());
+ expectSubscriptions(source.subscriptions).toBe(subs);
+ });
+
+ it('should unsubscribe when returning default value, even if destination does not unsubscribe', () => {
+ const source = hot('--a--|');
+ const subs = '^ !';
+ const expected = '-----(x|)';
+ const defaultValue = '42';
+
+ expectObservable((<any>source).elementAt(3, defaultValue).let(doNotUnsubscribe)).toBe(expected, { x: defaultValue });
+ expectSubscriptions(source.subscriptions).toBe(subs);
+ });
 });

src/operator/elementAt.ts

@@ -78,6 +78,7 @@ class ElementAtSubscriber<T> extends Subscriber<T> {
  if (this.index-- === 0) {
  this.destination.next(x);
  this.destination.complete();
+ this.unsubscribe();
  }
  }
 
@@ -88,8 +89,11 @@ class ElementAtSubscriber<T> extends Subscriber<T> {
  destination.next(this.defaultValue);
  } else {
  destination.error(new ArgumentOutOfRangeError);
+ this.unsubscribe();
+ return;
  }
  }
  destination.complete();
+ this.unsubscribe();
  }
 }