fix(find): will now properly unsubscribe when result completes or err…

…ors (#2671)

Force unsubscribe when resulting Observable completes or errors,
even when following operator does not unsubscribe reliably,
so that source Observable is not being subscribed unnecessarily.

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

spec/operators/find-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;
@@ -160,6 +161,33 @@ describe('Observable.prototype.find', () => {
     expectSubscriptions(source.subscriptions).toBe(subs);
   });
 
+  it('should unsubscribe from source when complete, even if following operator does not unsubscribe', () => {
+    const values = {a: 3, b: 9, c: 15, d: 20};
+    const source = hot('---a--b--c--d---|', values);
+    const subs =       '^        !       ';
+    const expected =   '---------(c|)    ';
+
+    const predicate = function (x) { return x % 5 === 0; };
+
+    expectObservable((<any>source).find(predicate).let(doNotUnsubscribe)).toBe(expected, values);
+    expectSubscriptions(source.subscriptions).toBe(subs);
+  });
+
+  it('should unsubscribe from source when predicate function errors,' +
+    ' even if following operator does not unsubscribe', () => {
+
+    const source = hot('--a--b--c--|');
+    const subs =       '^ !';
+    const expected =   '--#';
+
+    const predicate = function (value) {
+      throw 'error';
+    };
+
+    expectObservable((<any>source).find(predicate).let(doNotUnsubscribe)).toBe(expected);
+    expectSubscriptions(source.subscriptions).toBe(subs);
+  });
+
   it('should support type guards without breaking previous behavior', () => {
     // tslint:disable no-unused-variable
 

src/operator/find.ts

@@ -85,6 +85,7 @@ export class FindValueSubscriber<T> extends Subscriber<T> {
 
     destination.next(value);
     destination.complete();
+    this.unsubscribe();
   }
 
   protected _next(value: T): void {
@@ -97,6 +98,7 @@ export class FindValueSubscriber<T> extends Subscriber<T> {
       }
     } catch (err) {
       this.destination.error(err);
+      this.unsubscribe();
     }
   }