feat(@angular-devkit/architect): support async generator based builders

"Watch" type builders (builders with more than one result) can now be implemented as async generator functions. This allows such a builder to be more easily implemented in cases where an upstream API is async iterator/generator based or when async generators are the preferred method of implementation.

Author: clydin

etc/api/angular_devkit/architect/src/index.d.ts

@@ -40,7 +40,7 @@ export declare type BuilderInput = json.JsonObject & RealBuilderInput;
 
 export declare type BuilderOutput = json.JsonObject & RealBuilderOutput;
 
-export declare type BuilderOutputLike = SubscribableOrPromise<BuilderOutput> | BuilderOutput;
+export declare type BuilderOutputLike = AsyncIterable<BuilderOutput> | SubscribableOrPromise<BuilderOutput> | BuilderOutput;
 
 export declare type BuilderProgress = json.JsonObject & RealBuilderProgress & TypedBuilderProgress;
 
@@ -62,6 +62,8 @@ export interface BuilderRun {
 
 export declare function createBuilder<OptT extends json.JsonObject, OutT extends BuilderOutput = BuilderOutput>(fn: BuilderHandlerFn<OptT>): Builder<OptT>;
 
+export declare function fromAsyncIterable<T>(iterable: AsyncIterable<T>): Observable<T>;
+
 export declare function isBuilderOutput(obj: any): obj is BuilderOutput;
 
 export interface ScheduleOptions {

packages/angular_devkit/architect/src/api.ts

@@ -6,7 +6,7 @@
  * found in the LICENSE file at https://angular.io/license
  */
 import { analytics, experimental, json, logging } from '@angular-devkit/core';
-import { Observable, SubscribableOrPromise, from } from 'rxjs';
+import { Observable, SubscribableOrPromise, Subscriber, from } from 'rxjs';
 import { switchMap } from 'rxjs/operators';
 import { Schema as RealBuilderInput, Target as RealTarget } from './input-schema';
 import { Schema as RealBuilderOutput } from './output-schema';
@@ -258,17 +258,50 @@ export interface BuilderContext {
 /**
  * An accepted return value from a builder. Can be either an Observable, a Promise or a vector.
  */
-export type BuilderOutputLike = SubscribableOrPromise<BuilderOutput> | BuilderOutput;
+export type BuilderOutputLike = AsyncIterable<BuilderOutput> | SubscribableOrPromise<BuilderOutput> | BuilderOutput;
 
 // tslint:disable-next-line:no-any
 export function isBuilderOutput(obj: any): obj is BuilderOutput {
   if (!obj || typeof obj.then === 'function' || typeof obj.subscribe === 'function') {
     return false;
   }
 
+  if (typeof obj[Symbol.asyncIterator] === 'function') {
+    return false;
+  }
+
   return typeof obj.success === 'boolean';
 }
 
+export function fromAsyncIterable<T>(iterable: AsyncIterable<T>): Observable<T> {
+  return new Observable((subscriber) => {
+    handleAsyncIterator(subscriber, iterable[Symbol.asyncIterator]()).then(
+      () => subscriber.complete(),
+      (error) => subscriber.error(error),
+    );
+  });
+}
+
+async function handleAsyncIterator<T>(
+  subscriber: Subscriber<T>,
+  iterator: AsyncIterator<T>,
+): Promise<void> {
+  const teardown = new Promise<void>((resolve) => subscriber.add(() => resolve()));
+
+  try {
+    while (!subscriber.closed) {
+      const result = await Promise.race([teardown, iterator.next()]);
+      if (!result || result.done) {
+        break;
+      }
+
+      subscriber.next(result.value);
+    }
+  } finally {
+    await iterator.return?.();
+  }
+}
+
 /**
  * A builder handler function. The function signature passed to `createBuilder()`.
  */

packages/angular_devkit/architect/src/create-builder.ts

@@ -6,7 +6,7 @@
  * found in the LICENSE file at https://angular.io/license
  */
 import { analytics, experimental, json, logging } from '@angular-devkit/core';
-import { Observable, Subscription, from, of, throwError } from 'rxjs';
+import { Observable, Subscription, from, isObservable, of, throwError } from 'rxjs';
 import { tap } from 'rxjs/operators';
 import {
   BuilderContext,
@@ -18,6 +18,7 @@ import {
   ScheduleOptions,
   Target,
   TypedBuilderProgress,
+  fromAsyncIterable,
   isBuilderOutput,
   targetStringFromTarget,
 } from './api';
@@ -200,6 +201,8 @@ export function createBuilder<
           result = fn(i.options as OptT, context);
           if (isBuilderOutput(result)) {
             result = of(result);
+          } else if (!isObservable(result) && isAsyncIterable(result)) {
+            result = fromAsyncIterable(result);
           } else {
             result = from(result);
           }
@@ -234,3 +237,7 @@ export function createBuilder<
     [BuilderVersionSymbol]: require('../package.json').version,
   };
 }
+
+function isAsyncIterable<T>(obj: unknown): obj is AsyncIterable<T> {
+  return !!obj && typeof (obj as AsyncIterable<T>)[Symbol.asyncIterator] === 'function';
+}

packages/angular_devkit/architect/src/index_spec.ts

@@ -70,6 +70,14 @@ describe('architect', () => {
     await run.stop();
   });
 
+  it('supports async generator builders', async () => {
+    testArchitectHost.addBuilder('package:test', createBuilder(async function*() { yield { success: true }; }));
+
+    const run = await architect.scheduleBuilder('package:test', {});
+    expect(await run.result).toEqual(jasmine.objectContaining({ success: true }));
+    await run.stop();
+  });
+
   it('runs builders parallel', async () => {
     const run = await architect.scheduleBuilder('package:test', {});
     const run2 = await architect.scheduleBuilder('package:test', {});
@@ -123,7 +131,7 @@ describe('architect', () => {
     }
   });
 
-  it('works with watching builders', async () => {
+  it('works with watching observable builders', async () => {
     let results = 0;
     testArchitectHost.addBuilder('package:test-watch', createBuilder((_, context) => {
       called++;
@@ -150,6 +158,29 @@ describe('architect', () => {
     expect(all.length).toBe(10);
   });
 
+  it('works with watching async generator builders', async () => {
+    let results = 0;
+    testArchitectHost.addBuilder('package:test-watch-gen', createBuilder(async function*(_, context) {
+      called++;
+
+      for (let x = 0; x < 10; x++) {
+        context.reportRunning();
+        yield { success: true };
+        results++;
+      }
+    }));
+
+    const run = await architect.scheduleBuilder('package:test-watch-gen', {});
+    await run.result;
+    expect(called).toBe(1);
+    expect(results).toBe(1);
+
+    const all = await run.output.pipe(toArray()).toPromise();
+    expect(called).toBe(1);
+    expect(results).toBe(10);
+    expect(all.length).toBe(10);
+  });
+
   it('reports errors in the builder', async () => {
     testArchitectHost.addBuilder('package:error', createBuilder(() => {
       throw new Error('Error in the builder.');