fix(first): unsubscription logic properly called on complete and error (

#2463)

Force unsubscription logic when operator completes or errors, so that source Observable is only subscribed as long as it has to, even when combined with operators that do not immediately unsubscribe from 'first' when it completes or errors.

Closes #2455

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

spec/helpers/doNotUnsubscribe.ts

@@ -13,4 +13,8 @@ class DoNotUnsubscribeOperator<T, R> implements Rx.Operator<T, R> {
 
 class DoNotUnsubscribeSubscriber<T> extends Rx.Subscriber<T> {
   unsubscribe() {} // tslint:disable-line no-empty
-}
+<<<<<<< HEAD
+}
+=======
+}
+>>>>>>> af747f06... fix(first): unsubscription logic properly called on complete and error (#2463)

spec/operators/first-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;
@@ -220,6 +221,41 @@ describe('Observable.prototype.first', () => {
     expectSubscriptions(e1.subscriptions).toBe(sub);
   });
 
+  it('should unsubscribe from the source after first value, even if destination doesn\'t unsubscribe', () => {
+    const e1 = hot('--^---a---|');
+    const sub =      '^---!';
+    const expected = '----(a|)';
+
+    const result = e1.first().let(doNotUnsubscribe);
+
+    expectObservable(result).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should unsubscribe from the source after completion without value,' +
+    ' even if destination doesn\'t unsubscribe', () => {
+    const e1 = hot('--^---|');
+    const sub =      '^---!';
+    const expected = '----(a|)';
+
+    const result = e1.first(undefined, undefined, 'a').let(doNotUnsubscribe);
+
+    expectObservable(result).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
+  it('should unsubscribe from the source after erroring without value,' +
+    ' even if destination doesn\'t unsubscribe', () => {
+    const e1 = hot('--^---|');
+    const sub =      '^---!';
+    const expected = '----#';
+
+    const result = e1.first().let(doNotUnsubscribe);
+
+    expectObservable(result).toBe(expected, undefined, new Rx.EmptyError());
+    expectSubscriptions(e1.subscriptions).toBe(sub);
+  });
+
   it('should support type guards without breaking previous behavior', () => {
     // tslint:disable no-unused-variable
 

src/operator/first.ts

@@ -156,6 +156,7 @@ class FirstSubscriber<T, R> extends Subscriber<T> {
       this._emitted = true;
       destination.next(value);
       destination.complete();
+      this.unsubscribe();
       this.hasCompleted = true;
     }
   }
@@ -165,8 +166,10 @@ class FirstSubscriber<T, R> extends Subscriber<T> {
     if (!this.hasCompleted && typeof this.defaultValue !== 'undefined') {
       destination.next(this.defaultValue);
       destination.complete();
+      this.unsubscribe();
     } else if (!this.hasCompleted) {
       destination.error(new EmptyError);
+      this.unsubscribe();
     }
   }
 }