refactor(@angular-devkit/build-angular): convert karma builder to AsyncIterable

Refactor the Karma builder's `execute` function to return an `AsyncIterable`
using a `ReadableStream`. This removes the RxJS dependency and aligns the
builder with modern Angular CLI implementation patterns.

Additionally, this change fixes a race condition where the Karma server
could start even if the builder was cancelled during asynchronous
initialization. An `isCancelled` flag is now used to ensure execution
stops if a cancellation occurs before the server starts.

Author: clydin

packages/angular_devkit/build_angular/src/builders/karma/browser_builder.ts

@@ -7,17 +7,16 @@
  */
 
 import { purgeStaleBuildCache } from '@angular/build/private';
-import { BuilderContext, BuilderOutput } from '@angular-devkit/architect';
-import type { Config, ConfigOptions } from 'karma';
+import type { BuilderContext, BuilderOutput } from '@angular-devkit/architect';
+import type { Config, ConfigOptions, Server } from 'karma';
 import * as path from 'node:path';
-import { Observable, defaultIfEmpty, from, switchMap } from 'rxjs';
-import { Configuration } from 'webpack';
+import type { Configuration } from 'webpack';
 import { getCommonConfig, getStylesConfig } from '../../tools/webpack/configs';
-import { ExecutionTransformer } from '../../transforms';
+import type { ExecutionTransformer } from '../../transforms';
 import { generateBrowserWebpackConfigFromContext } from '../../utils/webpack-browser-config';
-import { Schema as BrowserBuilderOptions, OutputHashing } from '../browser/schema';
+import { type Schema as BrowserBuilderOptions, OutputHashing } from '../browser/schema';
 import { FindTestsPlugin } from './find-tests-plugin';
-import { Schema as KarmaBuilderOptions } from './schema';
+import type { Schema as KarmaBuilderOptions } from './schema';
 
 export type KarmaConfigOptions = ConfigOptions & {
   buildWebpack?: unknown;
@@ -33,9 +32,22 @@ export function execute(
     // The karma options transform cannot be async without a refactor of the builder implementation
     karmaOptions?: (options: KarmaConfigOptions) => KarmaConfigOptions;
   } = {},
-): Observable<BuilderOutput> {
-  return from(initializeBrowser(options, context, transforms.webpackConfiguration)).pipe(
-    switchMap(async ([karma, webpackConfig]) => {
+): AsyncIterable<BuilderOutput> {
+  let karmaServer: Server;
+  let isCancelled = false;
+
+  return new ReadableStream({
+    async start(controller) {
+      const [karma, webpackConfig] = await initializeBrowser(
+        options,
+        context,
+        transforms.webpackConfiguration,
+      );
+
+      if (isCancelled) {
+        return;
+      }
+
       const projectName = context.target?.project;
       if (!projectName) {
         throw new Error(`The 'karma' builder requires a target to be specified.`);
@@ -71,44 +83,54 @@ export function execute(
         logger: context.logger,
       };
 
-      const parsedKarmaConfig = await karma.config.parseConfig(
+      const parsedKarmaConfig = (await karma.config.parseConfig(
         options.karmaConfig && path.resolve(context.workspaceRoot, options.karmaConfig),
         transforms.karmaOptions ? transforms.karmaOptions(karmaOptions) : karmaOptions,
         { promiseConfig: true, throwErrors: true },
-      );
+      )) as KarmaConfigOptions;
 
-      return [karma, parsedKarmaConfig] as [typeof karma, KarmaConfigOptions];
-    }),
-    switchMap(
-      ([karma, karmaConfig]) =>
-        new Observable<BuilderOutput>((subscriber) => {
-          // Pass onto Karma to emit BuildEvents.
-          karmaConfig.buildWebpack ??= {};
-          if (typeof karmaConfig.buildWebpack === 'object') {
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any
-            (karmaConfig.buildWebpack as any).failureCb ??= () =>
-              subscriber.next({ success: false });
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any
-            (karmaConfig.buildWebpack as any).successCb ??= () =>
-              subscriber.next({ success: true });
-          }
+      if (isCancelled) {
+        return;
+      }
 
-          // Complete the observable once the Karma server returns.
-          const karmaServer = new karma.Server(karmaConfig as Config, (exitCode) => {
-            subscriber.next({ success: exitCode === 0 });
-            subscriber.complete();
-          });
+      const enqueue = (value: BuilderOutput) => {
+        try {
+          controller.enqueue(value);
+        } catch {
+          // Controller is already closed
+        }
+      };
 
-          const karmaStart = karmaServer.start();
+      const close = () => {
+        try {
+          controller.close();
+        } catch {
+          // Controller is already closed
+        }
+      };
 
-          // Cleanup, signal Karma to exit.
-          return () => {
-            void karmaStart.then(() => karmaServer.stop());
-          };
-        }),
-    ),
-    defaultIfEmpty({ success: false }),
-  );
+      // Pass onto Karma to emit BuildEvents.
+      parsedKarmaConfig.buildWebpack ??= {};
+      if (typeof parsedKarmaConfig.buildWebpack === 'object') {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        (parsedKarmaConfig.buildWebpack as any).failureCb ??= () => enqueue({ success: false });
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        (parsedKarmaConfig.buildWebpack as any).successCb ??= () => enqueue({ success: true });
+      }
+
+      // Close the stream once the Karma server returns.
+      karmaServer = new karma.Server(parsedKarmaConfig as Config, (exitCode) => {
+        enqueue({ success: exitCode === 0 });
+        close();
+      });
+
+      await karmaServer.start();
+    },
+    async cancel() {
+      isCancelled = true;
+      await karmaServer?.stop();
+    },
+  });
 }
 
 async function initializeBrowser(