fix(elementAt): will now properly unsubscribe when it completes or er…

…rors (#2501)

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.

spec/helpers/doNotUnsubscribe.ts

@@ -0,0 +1,16 @@
+///<reference path='../../typings/index.d.ts'/>
+import * as Rx from '../../dist/cjs/Rx';
+
+export function doNotUnsubscribe<T>(ob: Rx.Observable<T>): Rx.Observable<T> {
+  return ob.lift(new DoNotUnsubscribeOperator());
+}
+
+class DoNotUnsubscribeOperator<T, R> implements Rx.Operator<T, R> {
+  call(subscriber: Rx.Subscriber<R>, source: any): any {
+    return source.subscribe(new DoNotUnsubscribeSubscriber(subscriber));
+  }
+}
+
+class DoNotUnsubscribeSubscriber<T> extends Rx.Subscriber<T> {
+  unsubscribe() {} // tslint:disable-line no-empty
+}

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();
   }
 }