fix(find): unsubscribe from source when found (#3968)

* test(find): add failing unsubscribe test
* test(findIndex): add failing unsubscribe test
* test(first): add unsubscription test
* fix(find): unsubscribe from source when found
* chore(test): remove any

Author: cartant

spec/operators/find-spec.ts

@@ -1,10 +1,13 @@
 import{expect}from'chai';
-import{find,mergeMap}from'rxjs/operators';
+import{find,mergeMap,delay}from'rxjs/operators';
+import{TestScheduler}from'rxjs/testing';
 import{hot,cold,expectObservable,expectSubscriptions}from'../helpers/marble-testing';
 import{of,Observable,from}from'rxjs';
 
 declarefunctionasDiagram(arg: string): Function;
 
+declareconstrxTestScheduler: TestScheduler;
+
 /** @test {find} */
 describe('find operator',()=>{
 functiontruePredicate(x: any){
@@ -19,13 +22,13 @@ describe('find operator', () => {
 
 constpredicate=function(x: number){returnx%5===0;};
 
-expectObservable((<any>source).pipe(find(predicate))).toBe(expected,values);
+expectObservable(source.pipe(find(predicate))).toBe(expected,values);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
 it('should throw if not provided a function',()=>{
 expect(()=>{
-(<any>of('yut','yee','sam')).pipe(find('yee'asany));
+of('yut','yee','sam').pipe(find('yee'asany));
 }).to.throw(TypeError,'predicate is not a function');
 });
 
@@ -34,7 +37,7 @@ describe('find operator', () => {
 constsubs='^';
 constexpected='-';
 
-expectObservable((<any>source).pipe(find(truePredicate))).toBe(expected);
+expectObservable(source.pipe(find(truePredicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -43,7 +46,7 @@ describe('find operator', () => {
 constsubs='(^!)';
 constexpected='(x|)';
 
-constresult=(<any>source).pipe(find(truePredicate));
+constresult=source.pipe(find(truePredicate));
 
 expectObservable(result).toBe(expected,{x: undefined});
 expectSubscriptions(source.subscriptions).toBe(subs);
@@ -58,7 +61,7 @@ describe('find operator', () => {
 returnvalue==='a';
 };
 
-expectObservable((<any>source).pipe(find(predicate))).toBe(expected);
+expectObservable(source.pipe(find(predicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -71,7 +74,7 @@ describe('find operator', () => {
 returnvalue==='b';
 };
 
-expectObservable((<any>source).pipe(find(predicate))).toBe(expected);
+expectObservable(source.pipe(find(predicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -87,7 +90,7 @@ describe('find operator', () => {
 returnvalue===this.target;
 };
 
-expectObservable((<any>source).pipe(find(predicate,finder))).toBe(expected);
+expectObservable(source.pipe(find(predicate,finder))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -100,7 +103,7 @@ describe('find operator', () => {
 returnvalue==='z';
 };
 
-expectObservable((<any>source).pipe(find(predicate))).toBe(expected,{x: undefined});
+expectObservable(source.pipe(find(predicate))).toBe(expected,{x: undefined});
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -110,7 +113,7 @@ describe('find operator', () => {
 constexpected='------- ';
 constunsub=' ! ';
 
-constresult=(<any>source).pipe(find((value: string)=>value==='z'));
+constresult=source.pipe(find((value: string)=>value==='z'));
 
 expectObservable(result,unsub).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
@@ -122,7 +125,7 @@ describe('find operator', () => {
 constexpected='------- ';
 constunsub=' ! ';
 
-constresult=(<any>source).pipe(
+constresult=source.pipe(
 mergeMap((x: string)=>of(x)),
 find((value: string)=>value==='z'),
 mergeMap((x: string)=>of(x))
@@ -132,6 +135,20 @@ describe('find operator', () => {
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
+it('should unsubscribe when the predicate is matched',()=>{
+constsource=hot('--a--b---c-|');
+constsubs='^ !';
+constexpected='-------(b|)';
+
+constduration=rxTestScheduler.createTime('--|');
+
+expectObservable(source.pipe(
+find((value: string)=>value==='b'),
+delay(duration,rxTestScheduler)
+)).toBe(expected);
+expectSubscriptions(source.subscriptions).toBe(subs);
+});
+
 it('should raise if source raise error while element does not match with predicate',()=>{
 constsource=hot('--a--b--#');
 constsubs='^ !';
@@ -141,7 +158,7 @@ describe('find operator', () => {
 returnvalue==='z';
 };
 
-expectObservable((<any>source).pipe(find(predicate))).toBe(expected);
+expectObservable(source.pipe(find(predicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -154,7 +171,7 @@ describe('find operator', () => {
 throw'error';
 };
 
-expectObservable((<any>source).pipe(find(predicate))).toBe(expected);
+expectObservable(source.pipe(find(predicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 

spec/operators/findIndex-spec.ts

@@ -1,9 +1,12 @@
-import{findIndex,mergeMap}from'rxjs/operators';
+import{findIndex,mergeMap,delay}from'rxjs/operators';
+import{TestScheduler}from'rxjs/testing';
 import{hot,cold,expectObservable,expectSubscriptions}from'../helpers/marble-testing';
 import{of}from'rxjs';
 
 declarefunctionasDiagram(arg: string): Function;
 
+declareconstrxTestScheduler: TestScheduler;
+
 /** @test {findIndex} */
 describe('findIndex operator',()=>{
 functiontruePredicate(x: any){
@@ -18,7 +21,7 @@ describe('findIndex operator', () => {
 
 constpredicate=function(x: number){returnx%5===0;};
 
-expectObservable((<any>source).pipe(findIndex(predicate))).toBe(expected,{x: 2});
+expectObservable(source.pipe(findIndex(predicate))).toBe(expected,{x: 2});
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -27,7 +30,7 @@ describe('findIndex operator', () => {
 constsubs='^';
 constexpected='-';
 
-expectObservable((<any>source).pipe(findIndex(truePredicate))).toBe(expected);
+expectObservable(source.pipe(findIndex(truePredicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -36,7 +39,7 @@ describe('findIndex operator', () => {
 constsubs='(^!)';
 constexpected='(x|)';
 
-constresult=(<any>source).pipe(findIndex(truePredicate));
+constresult=source.pipe(findIndex(truePredicate));
 
 expectObservable(result).toBe(expected,{x: -1});
 expectSubscriptions(source.subscriptions).toBe(subs);
@@ -52,7 +55,7 @@ describe('findIndex operator', () => {
 returnvalue===sourceValue;
 };
 
-expectObservable((<any>source).pipe(findIndex(predicate))).toBe(expected,{x: 0});
+expectObservable(source.pipe(findIndex(predicate))).toBe(expected,{x: 0});
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -65,7 +68,7 @@ describe('findIndex operator', () => {
 returnvalue===7;
 };
 
-expectObservable((<any>source).pipe(findIndex(predicate))).toBe(expected,{x: 1});
+expectObservable(source.pipe(findIndex(predicate))).toBe(expected,{x: 1});
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -78,7 +81,7 @@ describe('findIndex operator', () => {
 constpredicate=function(this: typeofsourceValues,value: number){
 returnvalue===this.b;
 };
-constresult=(<any>source).pipe(findIndex(predicate,sourceValues));
+constresult=source.pipe(findIndex(predicate,sourceValues));
 
 expectObservable(result).toBe(expected,{x: 1});
 expectSubscriptions(source.subscriptions).toBe(subs);
@@ -93,7 +96,7 @@ describe('findIndex operator', () => {
 returnvalue==='z';
 };
 
-expectObservable((<any>source).pipe(findIndex(predicate))).toBe(expected,{x: -1});
+expectObservable(source.pipe(findIndex(predicate))).toBe(expected,{x: -1});
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -103,7 +106,7 @@ describe('findIndex operator', () => {
 constexpected='------- ';
 constunsub=' ! ';
 
-constresult=(<any>source).pipe(findIndex((value: string)=>value==='z'));
+constresult=source.pipe(findIndex((value: string)=>value==='z'));
 
 expectObservable(result,unsub).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
@@ -115,16 +118,30 @@ describe('findIndex operator', () => {
 constexpected='------- ';
 constunsub=' ! ';
 
-constresult=(<any>source).pipe(
+constresult=source.pipe(
 mergeMap((x: string)=>of(x)),
 findIndex((value: string)=>value==='z'),
-mergeMap((x: string)=>of(x))
+mergeMap((x: number)=>of(x))
 );
 
 expectObservable(result,unsub).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
+it('should unsubscribe when the predicate is matched',()=>{
+constsource=hot('--a--b---c-|');
+constsubs='^ !';
+constexpected='-------(x|)';
+
+constduration=rxTestScheduler.createTime('--|');
+
+expectObservable(source.pipe(
+findIndex((value: string)=>value==='b'),
+delay(duration,rxTestScheduler)
+)).toBe(expected,{x: 1});
+expectSubscriptions(source.subscriptions).toBe(subs);
+});
+
 it('should raise if source raise error while element does not match with predicate',()=>{
 constsource=hot('--a--b--#');
 constsubs='^ !';
@@ -134,7 +151,7 @@ describe('findIndex operator', () => {
 returnvalue==='z';
 };
 
-expectObservable((<any>source).pipe(findIndex(predicate))).toBe(expected);
+expectObservable(source.pipe(findIndex(predicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 
@@ -147,7 +164,7 @@ describe('findIndex operator', () => {
 throw'error';
 };
 
-expectObservable((<any>source).pipe(findIndex(predicate))).toBe(expected);
+expectObservable(source.pipe(findIndex(predicate))).toBe(expected);
 expectSubscriptions(source.subscriptions).toBe(subs);
 });
 });

spec/operators/first-spec.ts

@@ -1,10 +1,13 @@
 import{expect}from'chai';
 import{hot,expectObservable,expectSubscriptions}from'../helpers/marble-testing';
-import{first,mergeMap}from'rxjs/operators';
+import{first,mergeMap,delay}from'rxjs/operators';
+import{TestScheduler}from'rxjs/testing';
 import{of,from,Observable,Subject,EmptyError}from'rxjs';
 
 declarefunctionasDiagram(arg: string): Function;
 
+declareconstrxTestScheduler: TestScheduler;
+
 /** @test {first} */
 describe('Observable.prototype.first',()=>{
 asDiagram('first')('should take the first value of an observable with many values',()=>{
@@ -101,6 +104,20 @@ describe('Observable.prototype.first', () => {
 expectSubscriptions(e1.subscriptions).toBe(e1subs);
 });
 
+it('should unsubscribe when the first value is receiv',()=>{
+constsource=hot('--a--b---c-|');
+constsubs='^ !';
+constexpected='----(a|)';
+
+constduration=rxTestScheduler.createTime('--|');
+
+expectObservable(source.pipe(
+first(),
+delay(duration,rxTestScheduler)
+)).toBe(expected);
+expectSubscriptions(source.subscriptions).toBe(subs);
+});
+
 it('should return first value that matches a predicate',()=>{
 conste1=hot('--a-^--b--c--a--c--|');
 constexpected='------(c|)';

src/internal/operators/find.ts

@@ -84,6 +84,7 @@ export class FindValueSubscriber<T> extends Subscriber<T> {
 
 destination.next(value);
 destination.complete();
+this.unsubscribe();
 }
 
 protected_next(value: T): void{