From 0bfe6bb6e325469ffcbd1726e6b6877c571eab44 Mon Sep 17 00:00:00 2001
From: Mateusz Podlasin <Podlas29@gmail.com>
Date: Fri, 16 Jun 2017 19:54:09 +0200
Subject: [PATCH] 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 | 30 ++++++++++++++++++++++++++++++
 src/operator/elementAt.ts        |  4 ++++
 2 files changed, 34 insertions(+)

diff --git a/spec/operators/elementAt-spec.ts b/spec/operators/elementAt-spec.ts
index ac27b5cc85..5d0c7a8a74 100644
--- a/spec/operators/elementAt-spec.ts
+++ b/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);
+  });
 });
diff --git a/src/operator/elementAt.ts b/src/operator/elementAt.ts
index 65561e0890..a64fb12fde 100644
--- a/src/operator/elementAt.ts
+++ b/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();
   }
 }