refactor(@angular-devkit/schematics): reduce RxJS usage in callRule/callSource functions

The amount of additional RxJS operators and chaining has been reduced within the `callRule`
and `callSource` functions. This also reduces the complexity of the code and removes
several repetitive code segments.

Author: clydin

packages/angular_devkit/schematics/src/engine/schematic.ts

@@ -7,7 +7,7 @@
  */
 
 import { BaseException } from '@angular-devkit/core';
-import { Observable, of as observableOf } from 'rxjs';
+import { Observable } from 'rxjs';
 import { concatMap, first, map } from 'rxjs/operators';
 import { callRule } from '../rules/call';
 import { Tree } from '../tree/interface';
@@ -29,7 +29,8 @@ export class InvalidSchematicsNameException extends BaseException {
 }
 
 export class SchematicImpl<CollectionT extends object, SchematicT extends object>
-  implements Schematic<CollectionT, SchematicT> {
+  implements Schematic<CollectionT, SchematicT>
+{
   constructor(
     private _description: SchematicDescription<CollectionT, SchematicT>,
     private _factory: RuleFactory<{}>,
@@ -73,7 +74,7 @@ export class SchematicImpl<CollectionT extends object, SchematicT extends object
           input = tree;
         }
 
-        return callRule(this._factory(transformedOptions), observableOf(input), context).pipe(
+        return callRule(this._factory(transformedOptions), input, context).pipe(
           map((output) => {
             if (output === input) {
               return tree;

packages/angular_devkit/schematics/src/rules/call.ts

@@ -6,9 +6,9 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-import { BaseException, isPromise } from '@angular-devkit/core';
-import { Observable, from, isObservable, of as observableOf, throwError } from 'rxjs';
-import { defaultIfEmpty, last, mergeMap, tap } from 'rxjs/operators';
+import { BaseException } from '@angular-devkit/core';
+import { Observable, defer, isObservable } from 'rxjs';
+import { defaultIfEmpty, mergeMap } from 'rxjs/operators';
 import { Rule, SchematicContext, Source } from '../engine/interface';
 import { Tree, TreeSymbol } from '../tree/interface';
 
@@ -48,67 +48,52 @@ export class InvalidSourceResultException extends BaseException {
 }
 
 export function callSource(source: Source, context: SchematicContext): Observable<Tree> {
-  const result = source(context);
+  return defer(async () => {
+    let result = source(context);
 
-  if (isObservable(result)) {
-    // Only return the last Tree, and make sure it's a Tree.
-    return result.pipe(
-      defaultIfEmpty(),
-      last(),
-      tap((inner) => {
-        if (!inner || !(TreeSymbol in inner)) {
-          throw new InvalidSourceResultException(inner);
-        }
-      }),
-    );
-  } else if (result && TreeSymbol in result) {
-    return observableOf(result);
-  } else {
-    return throwError(new InvalidSourceResultException(result));
-  }
+    if (isObservable(result)) {
+      result = await result.pipe(defaultIfEmpty()).toPromise();
+    }
+
+    if (result && TreeSymbol in result) {
+      return result as Tree;
+    }
+
+    throw new InvalidSourceResultException(result);
+  });
 }
 
 export function callRule(
   rule: Rule,
   input: Tree | Observable<Tree>,
   context: SchematicContext,
 ): Observable<Tree> {
-  return (isObservable(input) ? input : observableOf(input)).pipe(
-    mergeMap((inputTree) => {
-      const result = rule(inputTree, context);
+  if (isObservable(input)) {
+    return input.pipe(mergeMap((inputTree) => callRuleAsync(rule, inputTree, context)));
+  } else {
+    return defer(() => callRuleAsync(rule, input, context));
+  }
+}
+
+async function callRuleAsync(rule: Rule, tree: Tree, context: SchematicContext): Promise<Tree> {
+  let result = await rule(tree, context);
+
+  while (typeof result === 'function') {
+    // This is considered a Rule, chain the rule and return its output.
+    result = await result(tree, context);
+  }
+
+  if (typeof result === 'undefined') {
+    return tree;
+  }
+
+  if (isObservable(result)) {
+    result = await result.pipe(defaultIfEmpty(tree)).toPromise();
+  }
+
+  if (TreeSymbol in result) {
+    return result as Tree;
+  }
 
-      if (!result) {
-        return observableOf(inputTree);
-      } else if (typeof result == 'function') {
-        // This is considered a Rule, chain the rule and return its output.
-        return callRule(result, inputTree, context);
-      } else if (isObservable(result)) {
-        // Only return the last Tree, and make sure it's a Tree.
-        return result.pipe(
-          defaultIfEmpty(),
-          last(),
-          tap((inner) => {
-            if (!inner || !(TreeSymbol in inner)) {
-              throw new InvalidRuleResultException(inner);
-            }
-          }),
-        );
-      } else if (isPromise(result)) {
-        return from(result).pipe(
-          mergeMap((inner) => {
-            if (typeof inner === 'function') {
-              // This is considered a Rule, chain the rule and return its output.
-              return callRule(inner, inputTree, context);
-            } else {
-              return observableOf(inputTree);
-            }
-          }),
-        );
-      } else if (TreeSymbol in result) {
-        return observableOf(result);
-      } else {
-        return throwError(new InvalidRuleResultException(result));
-      }
-    }),
-  );
+  throw new InvalidRuleResultException(result);
 }