feat(@angular-devkit/architect): add scheduleTargetAndForget function

It came up with Nrwl that this is a common pattern; someone wants to schedule a
target but does not want to manage the run himself. This function cancels the
run when the Observable is unsubscribed from (which is not the case for a
traditional run).

Because stop logic can be asynchronous, we need to add a teardown logic
handler to the context, which turns out to be useful for other cases as
well.

packages/angular_devkit/architect/src/api.ts

@@ -6,7 +6,8 @@
  * found in the LICENSE file at https://angular.io/license
  */
 import { experimental, json, logging } from '@angular-devkit/core';
-import { Observable } from 'rxjs';
+import { Observable, 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';
 import { Schema as RealBuilderProgress, State as BuilderProgressState } from './progress-schema';
@@ -211,6 +212,11 @@ export interface BuilderContext {
    * @param status Update the status string. If omitted the status string is not modified.
    */
   reportProgress(current: number, total?: number, status?: string): void;
+
+  /**
+   * Add teardown logic to this Context, so that when it's being stopped it will execute teardown.
+   */
+  addTeardown(teardown: () => (Promise<void> | void)): void;
 }
 
 
@@ -268,3 +274,41 @@ export function targetFromTargetString(str: string): Target {
     ...(tuple[2] !== undefined) && { configuration: tuple[2] },
   };
 }
+
+/**
+ * Schedule a target, and forget about its run. This will return an observable of outputs, that
+ * as a a teardown will stop the target from running. This means that the Run object this returns
+ * should not be shared.
+ *
+ * The reason this is not part of the Context interface is to keep the Context as normal form as
+ * possible. This is really an utility that people would implement in their project.
+ *
+ * @param context The context of your current execution.
+ * @param target The target to schedule.
+ * @param overrides Overrides that are used in the target.
+ * @param scheduleOptions Additional scheduling options.
+ */
+export function scheduleTargetAndForget(
+  context: BuilderContext,
+  target: Target,
+  overrides?: json.JsonObject,
+  scheduleOptions?: ScheduleOptions,
+): Observable<BuilderOutput> {
+  let resolve: (() => void) | null = null;
+  const promise = new Promise<void>(r => resolve = r);
+  context.addTeardown(() => promise);
+
+  return from(context.scheduleTarget(target, overrides, scheduleOptions)).pipe(
+    switchMap(run => new Observable<BuilderOutput>(observer => {
+      const subscription = run.output.subscribe(observer);
+
+      return () => {
+        subscription.unsubscribe();
+        // We can properly ignore the floating promise as it's a "reverse" promise; the teardown
+        // is waiting for the resolve.
+        // tslint:disable-next-line:no-floating-promises
+        run.stop().then(resolve);
+      };
+    })),
+  );
+}

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

@@ -37,6 +37,8 @@ export function createBuilder<
     const progressChannel = context.createChannel('progress');
     const logChannel = context.createChannel('log');
     let currentState: BuilderProgressState = BuilderProgressState.Stopped;
+    const teardownLogics: Array<() => (PromiseLike<void> | void)> = [];
+    let tearingDown = false;
     let current = 0;
     let status = '';
     let total = 1;
@@ -72,10 +74,15 @@ export function createBuilder<
         i => {
           switch (i.kind) {
             case experimental.jobs.JobInboundMessageKind.Stop:
-              observer.complete();
+              // Run teardown logic then complete.
+              tearingDown = true;
+              Promise.all(teardownLogics.map(fn => fn() || Promise.resolve()))
+                .then(() => observer.complete(), err => observer.error(err));
               break;
             case experimental.jobs.JobInboundMessageKind.Input:
-              onInput(i.value);
+              if (!tearingDown) {
+                onInput(i.value);
+              }
               break;
           }
         },
@@ -159,6 +166,9 @@ export function createBuilder<
                 progress({ state: currentState, current, total, status }, context);
             }
           },
+          addTeardown(teardown: () => (Promise<void> | void)): void {
+            teardownLogics.push(teardown);
+          },
         };
 
         context.reportRunning();