fix(forms): ensure observable validators are properly canceled (#15132)

Observable subscriptions from previous validation runs should be canceled
before a new subscription is created for the next validation run.
Currently the subscription that sets the errors is canceled properly,
but the source observable created by the validator is not. While this
does not affect validation status or error setting, the source
observables will incorrectly continue through the pipeline until they
complete. This change ensures that the whole stream is canceled.

AsyncValidatorFn previously had an "any" return type, but now it more
explicitly requires a Promise or Observable return type. We don't
anticipate this causing problems given that any other return type
would have caused a runtime error already.

packages/forms/src/directives/validators.ts

@@ -166,7 +166,7 @@ export interface ValidatorFn { (c: AbstractControl): {[key: string]: any}; }
  * @stable
  */
 export interface AsyncValidatorFn {
-  (c: AbstractControl): any /*Promise<{[key: string]: any}>|Observable<{[key: string]: any}>*/;
+  (c: AbstractControl): Promise<{[key: string]: any}>|Observable<{[key: string]: any}>;
 }
 
 /**

packages/forms/src/model.ts

@@ -6,12 +6,12 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-import {EventEmitter, ɵisObservable as isObservable, ɵisPromise as isPromise} from '@angular/core';
+import {EventEmitter} from '@angular/core';
 import {Observable} from 'rxjs/Observable';
-import {fromPromise} from 'rxjs/observable/fromPromise';
 
 import {composeAsyncValidators, composeValidators} from './directives/shared';
 import {AsyncValidatorFn, ValidatorFn} from './directives/validators';
+import {toObservable} from './validators';
 
 
 
@@ -57,11 +57,6 @@ function _find(control: AbstractControl, path: Array<string|number>| string, del
     return null;
   }, control);
 }
-
-function toObservable(r: any): Observable<any> {
-  return isPromise(r) ? fromPromise(r) : r;
-}
-
 function coerceToValidator(validator: ValidatorFn | ValidatorFn[]): ValidatorFn {
   return Array.isArray(validator) ? composeValidators(validator) : validator;
 }
@@ -420,12 +415,8 @@ export abstract class AbstractControl {
     if (this.asyncValidator) {
       this._status = PENDING;
       const obs = toObservable(this.asyncValidator(this));
-      if (!(isObservable(obs))) {
-        throw new Error(
-            `expected the following validator to return Promise or Observable: ${this.asyncValidator}. If you are using FormBuilder; did you forget to brace your validators in an array?`);
-      }
       this._asyncValidationSubscription =
-          obs.subscribe({next: (res: {[key: string]: any}) => this.setErrors(res, {emitEvent})});
+          obs.subscribe((res: {[key: string]: any}) => this.setErrors(res, {emitEvent}));
     }
   }
 

packages/forms/src/validators.ts

@@ -6,8 +6,12 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-import {InjectionToken, ɵisPromise as isPromise, ɵmerge as merge} from '@angular/core';
-import {toPromise} from 'rxjs/operator/toPromise';
+import {InjectionToken, ɵisObservable as isObservable, ɵisPromise as isPromise, ɵmerge as merge} from '@angular/core';
+import {Observable} from 'rxjs/Observable';
+import {forkJoin} from 'rxjs/observable/forkJoin';
+import {fromPromise} from 'rxjs/observable/fromPromise';
+import {map} from 'rxjs/operator/map';
+
 import {AsyncValidatorFn, Validator, ValidatorFn} from './directives/validators';
 import {AbstractControl, FormControl, FormGroup} from './model';
 
@@ -156,8 +160,8 @@ export class Validators {
     if (presentValidators.length == 0) return null;
 
     return function(control: AbstractControl) {
-      const promises = _executeAsyncValidators(control, presentValidators).map(_convertToPromise);
-      return Promise.all(promises).then(_mergeErrors);
+      const observables = _executeAsyncValidators(control, presentValidators).map(toObservable);
+      return map.call(forkJoin(observables), _mergeErrors);
     };
   }
 }
@@ -166,8 +170,12 @@ function isPresent(o: any): boolean {
   return o != null;
 }
 
-function _convertToPromise(obj: any): Promise<any> {
-  return isPromise(obj) ? obj : toPromise.call(obj);
+export function toObservable(r: any): Observable<any> {
+  const obs = isPromise(r) ? fromPromise(r) : r;
+  if (!(isObservable(obs))) {
+    throw new Error(`Expected validator to return Promise or Observable.`);
+  }
+  return obs;
 }
 
 function _executeValidators(control: AbstractControl, validators: ValidatorFn[]): any[] {