fix(devkit): properly emit failures when converting nx executor (#4866)

nrwl · Feb 23, 2021 · f206925 · f206925
1 parent ff5048e
commit f206925
Showing 1 changed file with 36 additions and 21 deletions.
diff --git a/packages/devkit/src/utils/convert-nx-executor.ts b/packages/devkit/src/utils/convert-nx-executor.ts
@@ -1,4 +1,4 @@
-import { from } from 'rxjs';
+import { from, Observable } from 'rxjs';
 import { map } from 'rxjs/operators';
 
 import {
@@ -36,29 +36,44 @@ export function convertNxExecutor(executor: Executor) {
           builderContext.target.target
         ];
     }
-    return from(toPromise(executor(options, context))).pipe(
-      map((output) => {
-        if (!output) {
-          return {
-            success: true,
-          };
-        } else {
-          return {
-            ...output,
-            success: true,
-          };
-        }
-      })
-    );
+    return toObservable(executor(options, context));
   };
   return require('@angular-devkit/architect').createBuilder(builderFunction);
 }
 
-async function toPromise(promiseOrAsyncIterator: any): Promise<any> {
-  if (typeof promiseOrAsyncIterator.then === 'function')
-    return promiseOrAsyncIterator;
-  let q;
-  for await (q of promiseOrAsyncIterator) {
+function toObservable<T extends { success: boolean }>(
+  promiseOrAsyncIterator: Promise<T> | AsyncIterableIterator<T>
+): Observable<T> {
+  if (typeof (promiseOrAsyncIterator as any).then === 'function') {
+    return from(promiseOrAsyncIterator as Promise<T>);
+  } else {
+    return new Observable((subscriber) => {
+      let asyncIterator = promiseOrAsyncIterator as AsyncIterableIterator<T>;
+
+      function recurse(iterator: AsyncIterableIterator<T>) {
+        iterator
+          .next()
+          .then((result) => {
+            if (!result.done) {
+              subscriber.next(result.value);
+              recurse(iterator);
+            } else {
+              if (result.value) {
+                subscriber.next(result.value);
+              }
+              subscriber.complete();
+            }
+          })
+          .catch((e) => {
+            subscriber.error(e);
+          });
+      }
+
+      recurse(asyncIterator);
+
+      return () => {
+        asyncIterator.return();
+      };
+    });
   }
-  return q;
 }