Merge pull request #13 from ReactiveX/master

fix(find): force unsubscribe when it completes or errors (ReactiveX#2550)

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/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;
@@ -162,6 +163,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 followring 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();
     }
   }