refactor(@angular-devkit/schematics): reduce direct rxjs usage in filter host tree

All direct usage of rxjs has been removed from the FilterHostTree implementation.
This removes the need to import rxjs within the containing file. The recursive
logic has also been replaced with an iterative approach.

(cherry picked from commit d087b5f)

Author: clydin

packages/angular_devkit/schematics/src/tree/host-tree.ts

@@ -18,8 +18,6 @@ import {
   virtualFs,
 } from '@angular-devkit/core';
 import { ParseError, parse as jsoncParse, printParseErrorCode } from 'jsonc-parser';
-import { EMPTY, Observable, concatMap, map, mergeMap } from 'rxjs';
-import { TextDecoder } from 'util';
 import {
   ContentHasMutatedException,
   FileAlreadyExistException,
@@ -496,38 +494,39 @@ export class FilterHostTree extends HostTree {
     // cast to allow access
     const originalBackend = (tree as FilterHostTree)._backend;
 
-    const recurse: (base: Path) => Observable<void> = (base) => {
-      return originalBackend.list(base).pipe(
-        mergeMap((x) => x),
-        map((path) => join(base, path)),
-        concatMap((path) => {
-          let isDirectory = false;
-          originalBackend.isDirectory(path).subscribe((val) => (isDirectory = val));
-          if (isDirectory) {
-            return recurse(path);
-          }
-
-          let isFile = false;
-          originalBackend.isFile(path).subscribe((val) => (isFile = val));
-          if (!isFile || !filter(path)) {
-            return EMPTY;
-          }
+    // Walk the original backend and add files that match the filter to the new backend
+    const pendingPaths: Path[] = ['/' as Path];
+    while (pendingPaths.length > 0) {
+      const currentPath = pendingPaths.pop();
+      if (currentPath === undefined) {
+        break;
+      }
 
-          let content: ArrayBuffer | null = null;
-          originalBackend.read(path).subscribe((val) => (content = val));
-          if (!content) {
-            return EMPTY;
-          }
+      let isDirectory = false;
+      originalBackend.isDirectory(currentPath).subscribe((val) => (isDirectory = val));
+      if (isDirectory) {
+        originalBackend
+          .list(currentPath)
+          .subscribe((val) => pendingPaths.push(...val.map((p) => join(currentPath, p))));
+        continue;
+      }
 
-          return newBackend.write(path, content as {} as virtualFs.FileBuffer);
-        }),
-      );
-    };
+      let isFile = false;
+      originalBackend.isFile(currentPath).subscribe((val) => (isFile = val));
+      if (!isFile || !filter(currentPath)) {
+        continue;
+      }
 
-    recurse(normalize('/')).subscribe();
+      let content = null;
+      originalBackend.read(currentPath).subscribe((val) => (content = val));
+      if (content !== null) {
+        newBackend.write(currentPath, content).subscribe();
+      }
+    }
 
     super(newBackend);
 
+    // Add actions that match the filter to new tree
     for (const action of tree.actions) {
       if (!filter(action.path)) {
         continue;