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

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