config-loader: migrate all sources to async iterators

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

packages/config-loader/package.json

@@ -43,13 +43,13 @@
     "fs-extra": "10.1.0",
     "json-schema": "^0.4.0",
     "json-schema-merge-allof": "^0.8.1",
+    "lodash": "^4.14.151",
     "json-schema-traverse": "^1.0.0",
     "minimist": "^1.2.8",
     "node-fetch": "^2.6.7",
     "typescript-json-schema": "^0.55.0",
     "yaml": "^2.0.0",
-    "yup": "^0.32.9",
-    "zen-observable": "^0.10.0"
+    "yup": "^0.32.9"
   },
   "devDependencies": {
     "@backstage/cli": "workspace:^",

packages/config-loader/src/lib/env.test.ts

@@ -1,146 +0,0 @@
-/*
- * Copyright 2020 The Backstage Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import { readEnvConfig } from './env';
-
-describe('readEnvConfig', () => {
-  it('should return empty config for empty env', () => {
-    expect(readEnvConfig({})).toEqual([]);
-  });
-
-  it('should return empty config for no matching keys', () => {
-    expect(
-      readEnvConfig({
-        NODE_ENV: 'production',
-        NOPE_ENV: 'development',
-        APP_CONFIG: 'foo',
-        APP__CONFIG_derp: 'herp',
-      }),
-    ).toEqual([]);
-  });
-
-  it('should create config from env', () => {
-    expect(
-      readEnvConfig({
-        NODE_ENV: 'production',
-        APP_CONFIG_foo: '"bar"',
-        APP_CONFIG_numbers_a: '1',
-        APP_CONFIG_numbers_b: '2',
-        APP_CONFIG_numbers_c: 'false',
-        APP_CONFIG_numbers_d: 'abc',
-        APP_CONFIG_numbers_e: undefined,
-        APP_CONFIG_very_deep_nested_config_object: '{}',
-      }),
-    ).toEqual([
-      {
-        data: {
-          foo: 'bar',
-          numbers: { a: 1, b: 2, c: false, d: 'abc' },
-          very: { deep: { nested: { config: { object: {} } } } },
-        },
-        context: 'env',
-      },
-    ]);
-  });
-
-  it('should accept string values', () => {
-    expect(
-      readEnvConfig({ APP_CONFIG_foo: '"abc"', APP_CONFIG_bar: 'xyz' }),
-    ).toEqual([
-      {
-        data: {
-          foo: 'abc',
-          bar: 'xyz',
-        },
-        context: 'env',
-      },
-    ]);
-  });
-
-  it('should accept complex objects', () => {
-    expect(
-      readEnvConfig({
-        APP_CONFIG_foo: '{ "a": 123, "b": "123", "c": [] }',
-        APP_CONFIG_bar: '[123, "abc", {}]',
-      }),
-    ).toEqual([
-      {
-        data: {
-          foo: { a: 123, b: '123', c: [] },
-          bar: [123, 'abc', {}],
-        },
-        context: 'env',
-      },
-    ]);
-  });
-
-  it.each([
-    ['APP_CONFIG__foo'],
-    ['APP_CONFIG_foo_'],
-    ['APP_CONFIG_fo_0'],
-    ['APP_CONFIG_fo/o'],
-    ['APP_CONFIG_fo o'],
-    ['APP_CONFIG_foo_(foo)_foo'],
-  ])('should reject invalid key %p', key => {
-    expect(() => readEnvConfig({ [key]: '0' })).toThrow(
-      `Invalid env config key '${key.replace('APP_CONFIG_', '')}'`,
-    );
-  });
-
-  it.each([['hello'], ['"hello'], ['{'], ['}']])(
-    'should fallback to string when invalid json value %p',
-    value => {
-      expect(readEnvConfig({ APP_CONFIG_foo: value })).toEqual([
-        {
-          data: {
-            foo: value,
-          },
-          context: 'env',
-        },
-      ]);
-    },
-  );
-
-  it('should not allow null as a value', () => {
-    expect(() =>
-      readEnvConfig({
-        APP_CONFIG_foo: 'null',
-      }),
-    ).toThrow(
-      "Failed to parse JSON-serialized config value for key 'foo', Error: value may not be null",
-    );
-  });
-
-  it('should not allow duplicate values', () => {
-    expect(() =>
-      readEnvConfig({
-        APP_CONFIG_foo_bar: '1',
-        APP_CONFIG_foo_bar_baz: '2',
-      }),
-    ).toThrow(
-      "Could not nest config for key 'foo_bar_baz' under existing value 'foo_bar'",
-    );
-  });
-
-  it('should not allow mixing of objects and other values', () => {
-    expect(() =>
-      readEnvConfig({
-        APP_CONFIG_nested_foo: '1',
-        APP_CONFIG_nested: '2',
-      }),
-    ).toThrow("Refusing to override existing config at key 'nested'");
-  });
-});

packages/config-loader/src/sources/ConfigSources.ts

@@ -14,14 +14,15 @@
  * limitations under the License.
  */
 
-import ObservableImpl from 'zen-observable';
-import { LoadConfigOptionsRemote } from './loader';
-import parseArgs from 'minimist';
-import { ConfigSource, ConfigSourceData } from './types';
-import { RemoteConfigSource } from './RemoteConfigSource';
-import { FileConfigSource } from './FileConfigSource';
-import { EnvConfigSource } from './EnvConfigSource';
 import { Config, ConfigReader } from '@backstage/config';
+import parseArgs from 'minimist';
+import { EnvConfigSource } from './EnvConfigSource';
+import { FileConfigSource } from './FileConfigSource';
+import { MergeConfigSource } from './MergedConfigSource';
+import { RemoteConfigSource } from './RemoteConfigSource';
+import { ConfigSource } from './types';
+import { ObservableConfigProxy } from './ObservableConfigProxy';
+import { LoadConfigOptionsRemote } from '../loader';
 
 export class ConfigSources {
   static parseArgs(
@@ -62,46 +63,28 @@ export class ConfigSources {
   }
 
   static merge(sources: ConfigSource[]): ConfigSource {
-    return {
-      configData$: new ObservableImpl<ConfigSourceData[]>(observer => {
-        const dataArr = new Array<ConfigSourceData[]>(sources.length);
-        let gotAll = false;
-        const subscriptions = sources.map((source, i) =>
-          source.configData$.subscribe({
-            next({ data }) {
-              dataArr[i] = data;
-              if (gotAll || dataArr.every(Boolean)) {
-                gotAll = true;
-                observer.next(dataArr.flat(1));
-              }
-            },
-            error(error) {
-              observer.error(error);
-            },
-          }),
-        );
-        return () =>
-          subscriptions.forEach(subscription => subscription.unsubscribe());
-      }),
-    };
+    return MergeConfigSource.fromConfigSources(sources);
   }
 
   static toConfig(source: ConfigSource): Promise<LiveConfig> {
-    return new Promise((resolve, reject) => {
-      let config: Config | undefined = undefined;
-      source.configData$.subscribe({
-        next({ data }) {
+    return new Promise(async (resolve, reject) => {
+      let config: ObservableConfigProxy | undefined = undefined;
+      try {
+        const abortController = new AbortController();
+        for await (const { data } of source.readConfigData({
+          signal: abortController.signal,
+        })) {
           if (config) {
             config.setConfig(ConfigReader.fromConfigs(data));
           } else {
-            config = ConfigReader.fromConfigs(data);
+            config = ObservableConfigProxy.create(abortController);
+            config!.setConfig(ConfigReader.fromConfigs(data));
             resolve(config);
           }
-        },
-        error(error) {
-          reject(error);
-        },
-      });
+        }
+      } catch (error) {
+        reject(error);
+      }
     });
   }
 }

packages/config-loader/src/sources/EnvConfigSource.test.ts

@@ -15,15 +15,29 @@
  */
 
 import { EnvConfigSource, readEnvConfig } from './EnvConfigSource';
-import ObservableImpl from 'zen-observable';
+import { ConfigSource, ConfigSourceData } from './types';
+
+async function readAll(source: ConfigSource) {
+  const entries = new Array<{ data: ConfigSourceData[] }>();
+  for await (const item of source.readConfigData()) {
+    entries.push(item);
+  }
+  return entries;
+}
 
 describe('EnvConfigSource', () => {
   it('should return empty config for empty env', async () => {
     const source = EnvConfigSource.create({ env: {} });
-    const spy = jest.fn();
-    await ObservableImpl.from(source.configData$).forEach(d => spy(d));
-    expect(spy).toHaveBeenCalledTimes(1);
-    expect(spy).toHaveBeenCalledWith({ data: [] });
+
+    await expect(readAll(source)).resolves.toEqual([{ data: [] }]);
+  });
+
+  it('should forward config values', async () => {
+    const source = EnvConfigSource.create({ env: { APP_CONFIG_foo: 'bar' } });
+
+    await expect(readAll(source)).resolves.toEqual([
+      { data: [{ context: 'env', data: { foo: 'bar' } }] },
+    ]);
   });
 });
 

packages/config-loader/src/sources/EnvConfigSource.ts

@@ -17,7 +17,7 @@
 import { AppConfig } from '@backstage/config';
 import { assertError } from '@backstage/errors';
 import { JsonObject } from '@backstage/types';
-import { ConfigSource } from './types';
+import { ConfigSource, ConfigSourceData } from './types';
 
 export class EnvConfigSource implements ConfigSource {
   static create(options: {
@@ -32,7 +32,7 @@ export class EnvConfigSource implements ConfigSource {
     private readonly env: { [name: string]: string | undefined },
   ) {}
 
-  async *readConfigData() {
+  async *readConfigData(): AsyncIterableIterator<{ data: ConfigSourceData[] }> {
     const data = readEnvConfig(this.env);
     yield { data };
     return;

packages/config-loader/src/sources/FileConfigSource.ts

@@ -14,6 +14,85 @@
  * limitations under the License.
  */
 
-export class FileConfigSource implements ConfigSource {
-  static create(options: { logger: Logger }): ConfigSource {}
+import chokidar, { FSWatcher } from 'chokidar';
+import fs from 'fs-extra';
+import { basename, isAbsolute } from 'path';
+import yaml from 'yaml';
+import {
+  AsyncConfigSourceIterator,
+  ConfigSource,
+  ConfigSourceData,
+  ReadConfigDataOptions,
+} from './types';
+
+export interface FileConfigSourceOptions {
+  path: string;
+}
+
+export class FileConfigSource implements ConfigSource {
+  static create(options: FileConfigSourceOptions): ConfigSource {
+    if (!isAbsolute(options.path)) {
+      throw new Error(`Config load path is not absolute: '${options.path}'`);
+    }
+    return new FileConfigSource(options);
+  }
+
+  readonly #path: string;
+
+  private constructor(options: FileConfigSourceOptions) {
+    this.#path = options.path;
+  }
+
+  async *readConfigData(
+    options?: ReadConfigDataOptions,
+  ): AsyncConfigSourceIterator {
+    const signal = options?.signal;
+    const configFileName = basename(this.#path);
+
+    const readConfigFile = async (): Promise<ConfigSourceData> => {
+      const content = await fs.readFile(this.#path, 'utf8');
+      const data = yaml.parse(content);
+      return { data, context: configFileName, path: this.#path };
+    };
+
+    const watcher = chokidar.watch(this.#path, {
+      usePolling: process.env.NODE_ENV === 'test',
+    });
+
+    signal?.addEventListener('abort', () => {
+      watcher.close();
+    });
+
+    yield { data: [await readConfigFile()] };
+
+    for (;;) {
+      const event = await this.#waitForEvent(watcher, signal);
+      if (event === 'abort') {
+        return;
+      }
+      yield { data: [await readConfigFile()] };
+    }
+  }
+
+  #waitForEvent(
+    watcher: FSWatcher,
+    signal?: AbortSignal,
+  ): Promise<'change' | 'abort'> {
+    return new Promise(resolve => {
+      function onChange() {
+        resolve('change');
+        onDone();
+      }
+      function onAbort() {
+        resolve('abort');
+        onDone();
+      }
+      function onDone() {
+        watcher.removeListener('change', onChange);
+        signal?.removeEventListener('abort', onAbort);
+      }
+      watcher.addListener('change', onChange);
+      signal?.addEventListener('abort', onAbort);
+    });
+  }
 }

packages/config-loader/src/sources/MergedConfigSource.ts

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2023 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {
+  AsyncConfigSourceIterator,
+  ConfigSource,
+  ConfigSourceData,
+  ReadConfigDataOptions,
+} from './types';
+
+export class MergeConfigSource implements ConfigSource {
+  static fromConfigSources(sources: ConfigSource[]): ConfigSource {
+    return new MergeConfigSource(sources);
+  }
+
+  private constructor(private readonly sources: ConfigSource[]) {}
+
+  async *readConfigData(
+    options?: ReadConfigDataOptions,
+  ): AsyncConfigSourceIterator {
+    const its = this.sources.map(source => source.readConfigData(options));
+    const initialResults = await Promise.all(its.map(it => it.next()));
+    const data = initialResults.map((result, i) => {
+      if (result.done) {
+        throw new Error(
+          `Config source ${String(this.sources[i])} returned no data`,
+        );
+      }
+      return result.value.data;
+    });
+
+    yield { data: data.flat(1) };
+
+    const results: Array<
+      | Promise<
+          readonly [number, IteratorResult<{ data: ConfigSourceData[] }, void>]
+        >
+      | undefined
+    > = its.map((it, i) => nextWithIndex(it, i));
+
+    while (results.some(Boolean)) {
+      try {
+        const [i, result] = (await Promise.race(results))!;
+        if (result.done) {
+          results[i] = undefined;
+        } else {
+          results[i] = nextWithIndex(its[i], i);
+          data[i] = result.value.data;
+          yield { data: data.flat(1) };
+        }
+      } catch (error) {
+        const source = this.sources[error.index];
+        if (source) {
+          throw new Error(`Config source ${String(source)} failed: ${error}`);
+        }
+        throw error;
+      }
+    }
+  }
+}
+
+// Helper to wait for the next value of the iterator, while decorating the value
+// or error with the index of the iterator.
+function nextWithIndex<T>(
+  iterator: AsyncIterator<T, void, void>,
+  index: number,
+): Promise<readonly [index: number, result: IteratorResult<T, void>]> {
+  return iterator.next().then(
+    r => [index, r] as const,
+    e => {
+      throw Object.assign(e, { index });
+    },
+  );
+}

packages/config-loader/src/sources/MutableConfigSource.ts

@@ -14,9 +14,13 @@
  * limitations under the License.
  */
 
-import { JsonObject, Observable } from '@backstage/types';
-import { SimpleBehaviorSubject } from './SimpleBehaviorSubject';
-import { ConfigSource, ConfigSourceData } from './types';
+import { JsonObject } from '@backstage/types';
+import {
+  AsyncConfigSourceIterator,
+  ConfigSource,
+  ReadConfigDataOptions,
+} from './types';
+import { simpleDefer, SimpleDeferred, waitOrAbort } from './utils';
 
 export class MutableConfigSource implements ConfigSource {
   static create(options: { data: JsonObject; context?: string }): ConfigSource {
@@ -26,15 +30,34 @@ export class MutableConfigSource implements ConfigSource {
     );
   }
 
-  private subject: SimpleBehaviorSubject<JsonObject>;
-  readonly data$: Observable<ConfigSourceData[]>;
+  #currentData: JsonObject;
+  #deferred: SimpleDeferred<void>;
+  readonly #context: string;
 
   private constructor(initialData: JsonObject, context: string) {
-    this.subject = new SimpleBehaviorSubject(initialData);
-    this.data$ = this.subject.observable.map(data => [{ context, data }]);
+    this.#currentData = initialData;
+    this.#context = context;
+    this.#deferred = simpleDefer();
+  }
+
+  async *readConfigData(
+    options?: ReadConfigDataOptions | undefined,
+  ): AsyncConfigSourceIterator {
+    yield { data: [{ data: this.#currentData, context: this.#context }] };
+
+    for (;;) {
+      const [ok] = await waitOrAbort(this.#deferred.promise, options?.signal);
+      if (!ok) {
+        return;
+      }
+
+      yield { data: [{ data: this.#currentData, context: this.#context }] };
+    }
   }
 
   setData(data: JsonObject) {
-    this.subject.next(data);
+    this.#currentData = data;
+    this.#deferred.resolve();
+    this.#deferred = simpleDefer();
   }
 }

packages/config-loader/src/sources/RemoteConfigSource.ts

@@ -14,6 +14,106 @@
  * limitations under the License.
  */
 
-export class RemoteConfigSource implements ConfigSource {
-  static create(options: { logger: Logger }): ConfigSource {}
+import {
+  AsyncConfigSourceIterator,
+  ConfigSource,
+  ReadConfigDataOptions,
+} from './types';
+import isEqual from 'lodash/isEqual';
+import yaml from 'yaml';
+import { ResponseError } from '@backstage/errors';
+import { JsonObject } from '@backstage/types';
+
+const DEFAULT_RELOAD_INTERVAL_SECONDS = 60;
+
+export interface RemoteConfigSourceOptions {
+  url: string;
+  reloadIntervalSeconds?: number;
+}
+
+export class RemoteConfigSource implements ConfigSource {
+  static create(options: RemoteConfigSourceOptions): ConfigSource {
+    try {
+      // eslint-disable-next-line no-new
+      new URL(options.url);
+    } catch (error) {
+      throw new Error(
+        `Invalid URL provided to remote config source, '${options.url}', ${error}`,
+      );
+    }
+    return new RemoteConfigSource(options);
+  }
+
+  readonly #url: string;
+  readonly #reloadIntervalSeconds: number;
+
+  private constructor(options: RemoteConfigSourceOptions) {
+    this.#url = options.url;
+    this.#reloadIntervalSeconds =
+      options.reloadIntervalSeconds ?? DEFAULT_RELOAD_INTERVAL_SECONDS;
+  }
+
+  async *readConfigData(
+    options?: ReadConfigDataOptions | undefined,
+  ): AsyncConfigSourceIterator {
+    let data = await this.#load();
+
+    yield { data: [{ data, context: this.#url }] };
+
+    for (;;) {
+      if (options?.signal?.aborted) {
+        return;
+      }
+      const loadStart = Date.now();
+
+      try {
+        const newData = await this.#load(options?.signal);
+        if (newData && !isEqual(data, newData)) {
+          data = newData;
+          yield { data: [{ data, context: this.#url }] };
+        }
+      } catch (error) {
+        console.error(`Failed to read config from ${this.#url}, ${error}`);
+      }
+      const loadTime = Date.now() - loadStart;
+
+      await this.#wait(loadTime, options?.signal);
+    }
+  }
+
+  async #load(signal?: AbortSignal): Promise<JsonObject> {
+    const res = await fetch(this.#url, { signal });
+    if (!res.ok) {
+      throw ResponseError.fromResponse(res);
+    }
+
+    const content = await res.text();
+    const data = yaml.parse(content);
+    if (data === null) {
+      throw new Error('configuration data is null');
+    } else if (typeof data !== 'object') {
+      throw new Error('configuration data is not an object');
+    } else if (Array.isArray(data)) {
+      throw new Error(
+        'configuration data is an array, expected an object instead',
+      );
+    }
+    return data;
+  }
+
+  async #wait(loadTimeMs: number, signal?: AbortSignal) {
+    return new Promise<void>(resolve => {
+      const timeoutId = setTimeout(
+        onDone,
+        Math.max(0, this.#reloadIntervalSeconds * 1000 - loadTimeMs),
+      );
+      signal?.addEventListener('abort', onDone);
+
+      function onDone() {
+        clearTimeout(timeoutId);
+        signal?.removeEventListener('abort', onDone);
+        resolve();
+      }
+    });
+  }
 }

packages/config-loader/src/sources/SimpleBehaviorSubject.ts

@@ -1,61 +0,0 @@
-/*
- * Copyright 2023 The Backstage Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import ObservableImpl from 'zen-observable';
-
-/**
- * A simple behavior subject that doesn't implement the observable interface itself,
- * and assumes that the observable is never completed.
- */
-export class SimpleBehaviorSubject<T> {
-  private currentValue: T;
-  private terminatingError: Error | undefined;
-
-  readonly observable: ObservableImpl<T>;
-
-  constructor(value: T) {
-    this.currentValue = value;
-    this.terminatingError = undefined;
-    this.observable = new ObservableImpl<T>(subscriber => {
-      if (this.terminatingError) {
-        subscriber.error(this.terminatingError);
-        return () => {};
-      }
-
-      subscriber.next(this.currentValue);
-
-      this.subscribers.add(subscriber);
-      return () => {
-        this.subscribers.delete(subscriber);
-      };
-    });
-  }
-
-  private readonly subscribers = new Set<
-    ZenObservable.SubscriptionObserver<T>
-  >();
-
-  next(value: T) {
-    this.currentValue = value;
-    this.subscribers.forEach(subscriber => subscriber.next(value));
-  }
-
-  error(error: Error) {
-    this.terminatingError = error;
-    this.subscribers.forEach(subscriber => subscriber.error(error));
-    this.subscribers.clear();
-  }
-}

packages/config-loader/src/sources/StaticConfigSource.ts

@@ -15,47 +15,112 @@
  */
 
 import { JsonObject, Observable } from '@backstage/types';
-import ObservableImpl from 'zen-observable';
-import { ConfigSource, ConfigSourceData } from './types';
+import {
+  AsyncConfigSourceIterator,
+  ConfigSource,
+  ReadConfigDataOptions,
+} from './types';
+import { simpleDefer } from './utils';
 
-export class StaticConfigSource implements ConfigSource {
-  static create(options: {
-    data: JsonObject | Observable<JsonObject> | Promise<JsonObject>;
-    context?: string;
-  }): ConfigSource {
-    const { data, context = 'static-config' } = options;
-    if (!data) {
-      return new StaticConfigSource(ObservableImpl.of({}), context);
+export interface StaticConfigSourceOptions {
+  data:
+    | JsonObject
+    | Observable<JsonObject>
+    | PromiseLike<JsonObject>
+    | AsyncIterable<JsonObject>;
+  context?: string;
+}
+
+/** @internal */
+class StaticObservableConfigSource implements ConfigSource {
+  constructor(
+    private readonly data: Observable<JsonObject>,
+    private readonly context: string,
+  ) {}
+
+  async *readConfigData(
+    options?: ReadConfigDataOptions | undefined,
+  ): AsyncConfigSourceIterator {
+    const queue = new Array<JsonObject>();
+    let deferred = simpleDefer<void>();
+
+    const sub = this.data.subscribe({
+      next(value) {
+        queue.push(value);
+        deferred.resolve();
+        deferred = simpleDefer();
+      },
+      complete() {
+        queue.length = 0;
+        deferred.resolve();
+      },
+    });
+
+    options?.signal?.addEventListener('abort', () => {
+      sub.unsubscribe();
+      queue.length = 0;
+      deferred.resolve();
+    });
+
+    for (;;) {
+      await deferred.promise;
+      if (queue.length === 0) {
+        return;
+      }
+      while (queue.length > 0) {
+        yield { data: [{ data: queue.shift()!, context: this.context }] };
+      }
     }
-
-    if ('subscribe' in data && typeof data.subscribe === 'function') {
-      return new StaticConfigSource(
-        ObservableImpl.from(data as Observable<JsonObject>),
-        context,
-      );
-    }
-    if ('then' in data && typeof data.then === 'function') {
-      return new StaticConfigSource(
-        new ObservableImpl(subscriber => {
-          (data as Promise<JsonObject>).then(
-            value => subscriber.next(value),
-            error => subscriber.error(error),
-          );
-          return () => {};
-        }),
-        context,
-      );
-    }
-
-    return new StaticConfigSource(
-      ObservableImpl.of(data as JsonObject),
-      context,
-    );
-  }
-
-  readonly data$: Observable<ConfigSourceData[]>;
-
-  private constructor(observable: ObservableImpl<JsonObject>, context: string) {
-    this.data$ = observable.map(data => [{ context, data }]);
+  }
+}
+
+function isObservable<T>(value: {}): value is Observable<T> {
+  return 'subscribe' in value && typeof (value as any).subscribe === 'function';
+}
+
+function isAsyncIterable<T>(value: {}): value is AsyncIterable<T> {
+  return Symbol.asyncIterator in value;
+}
+
+export class StaticConfigSource implements ConfigSource {
+  static create(options: StaticConfigSourceOptions): ConfigSource {
+    const { data, context = 'static-config' } = options;
+    if (!data) {
+      return {
+        async *readConfigData(): AsyncConfigSourceIterator {
+          yield { data: [] };
+          return;
+        },
+      };
+    }
+
+    if (isObservable(data)) {
+      return new StaticObservableConfigSource(
+        data as Observable<JsonObject>,
+        context,
+      );
+    }
+
+    if (isAsyncIterable(data)) {
+      return {
+        async *readConfigData(): AsyncConfigSourceIterator {
+          for await (const value of data) {
+            yield { data: [{ data: value, context }] };
+          }
+        },
+      };
+    }
+
+    return new StaticConfigSource(data, context);
+  }
+
+  private constructor(
+    private readonly promise: JsonObject | PromiseLike<JsonObject>,
+    private readonly context: string,
+  ) {}
+
+  async *readConfigData(): AsyncConfigSourceIterator {
+    yield { data: [{ data: await this.promise, context: this.context }] };
+    return;
   }
 }

packages/config-loader/src/sources/types.ts

@@ -27,8 +27,15 @@ export interface ReadConfigDataOptions {
   signal?: AbortSignal;
 }
 
-export interface ConfigSource {
-  readConfigData(
-    options?: ReadConfigDataOptions,
-  ): AsyncIterator<{ data: ConfigSourceData[] }, void, void>;
+export interface AsyncConfigSourceIterator
+  extends AsyncIterator<{ data: ConfigSourceData[] }, void, void> {
+  [Symbol.asyncIterator](): AsyncIterator<
+    { data: ConfigSourceData[] },
+    void,
+    void
+  >;
+}
+
+export interface ConfigSource {
+  readConfigData(options?: ReadConfigDataOptions): AsyncConfigSourceIterator;
 }

packages/config-loader/src/sources/utils.ts

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2023 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/** @internal */
+export interface SimpleDeferred<T> {
+  promise: Promise<T>;
+  resolve(value: T): void;
+}
+
+/** @internal */
+export function simpleDefer<T>(): SimpleDeferred<T> {
+  let resolve: (value: T) => void;
+  const promise = new Promise<T>(_resolve => {
+    resolve = _resolve;
+  });
+  return { promise, resolve: resolve! };
+}
+
+/** @internal */
+export async function waitOrAbort<T>(
+  promise: PromiseLike<T>,
+  signal?: AbortSignal,
+): Promise<[ok: true, value: T] | [ok: false]> {
+  return new Promise((resolve, reject) => {
+    const onAbort = () => {
+      resolve([false]);
+    };
+    promise.then(
+      value => {
+        resolve([true, value]);
+        signal?.removeEventListener('abort', onAbort);
+      },
+      error => {
+        reject(error);
+        signal?.removeEventListener('abort', onAbort);
+      },
+    );
+    signal?.addEventListener('abort', onAbort);
+  });
+}