From 138f422fb75c69428555c068b81653c3c672fd1e Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 29 Mar 2023 15:41:30 +0200 Subject: [PATCH] config-loader: migrate all sources to async iterators Signed-off-by: Patrik Oldsberg --- packages/config-loader/package.json | 4 +- packages/config-loader/src/lib/env.test.ts | 146 ------------------ .../src/sources/ConfigSources.ts | 61 +++----- .../src/sources/EnvConfigSource.test.ts | 24 ++- .../src/sources/EnvConfigSource.ts | 4 +- .../src/sources/FileConfigSource.ts | 83 +++++++++- .../src/sources/MergedConfigSource.ts | 87 +++++++++++ .../src/sources/MutableConfigSource.ts | 39 ++++- .../src/sources/RemoteConfigSource.ts | 104 ++++++++++++- .../src/sources/SimpleBehaviorSubject.ts | 61 -------- .../src/sources/StaticConfigSource.ts | 145 ++++++++++++----- packages/config-loader/src/sources/types.ts | 15 +- packages/config-loader/src/sources/utils.ts | 53 +++++++ 13 files changed, 515 insertions(+), 311 deletions(-) delete mode 100644 packages/config-loader/src/lib/env.test.ts create mode 100644 packages/config-loader/src/sources/MergedConfigSource.ts delete mode 100644 packages/config-loader/src/sources/SimpleBehaviorSubject.ts create mode 100644 packages/config-loader/src/sources/utils.ts diff --git a/packages/config-loader/package.json b/packages/config-loader/package.json index 44c326a03a..02870be8ea 100644 --- a/packages/config-loader/package.json +++ b/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:^", diff --git a/packages/config-loader/src/lib/env.test.ts b/packages/config-loader/src/lib/env.test.ts deleted file mode 100644 index 6908f0adb0..0000000000 --- a/packages/config-loader/src/lib/env.test.ts +++ /dev/null @@ -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'"); - }); -}); diff --git a/packages/config-loader/src/sources/ConfigSources.ts b/packages/config-loader/src/sources/ConfigSources.ts index 061fbc991e..02780e9345 100644 --- a/packages/config-loader/src/sources/ConfigSources.ts +++ b/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(observer => { - const dataArr = new Array(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 { - 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); + } }); } } diff --git a/packages/config-loader/src/sources/EnvConfigSource.test.ts b/packages/config-loader/src/sources/EnvConfigSource.test.ts index c1fec52a7c..705f9bce93 100644 --- a/packages/config-loader/src/sources/EnvConfigSource.test.ts +++ b/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' } }] }, + ]); }); }); diff --git a/packages/config-loader/src/sources/EnvConfigSource.ts b/packages/config-loader/src/sources/EnvConfigSource.ts index e02f8617dc..6bb2402767 100644 --- a/packages/config-loader/src/sources/EnvConfigSource.ts +++ b/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; diff --git a/packages/config-loader/src/sources/FileConfigSource.ts b/packages/config-loader/src/sources/FileConfigSource.ts index 051cf74aef..aed5869b04 100644 --- a/packages/config-loader/src/sources/FileConfigSource.ts +++ b/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 => { + 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); + }); + } } diff --git a/packages/config-loader/src/sources/MergedConfigSource.ts b/packages/config-loader/src/sources/MergedConfigSource.ts new file mode 100644 index 0000000000..4aeab4cecf --- /dev/null +++ b/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( + iterator: AsyncIterator, + index: number, +): Promise]> { + return iterator.next().then( + r => [index, r] as const, + e => { + throw Object.assign(e, { index }); + }, + ); +} diff --git a/packages/config-loader/src/sources/MutableConfigSource.ts b/packages/config-loader/src/sources/MutableConfigSource.ts index 1d331be1bf..cb9401b3bc 100644 --- a/packages/config-loader/src/sources/MutableConfigSource.ts +++ b/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; - readonly data$: Observable; + #currentData: JsonObject; + #deferred: SimpleDeferred; + 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(); } } diff --git a/packages/config-loader/src/sources/RemoteConfigSource.ts b/packages/config-loader/src/sources/RemoteConfigSource.ts index 6117af65f5..5f991cf7f6 100644 --- a/packages/config-loader/src/sources/RemoteConfigSource.ts +++ b/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 { + 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(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(); + } + }); + } } diff --git a/packages/config-loader/src/sources/SimpleBehaviorSubject.ts b/packages/config-loader/src/sources/SimpleBehaviorSubject.ts deleted file mode 100644 index 54c6e8c1b4..0000000000 --- a/packages/config-loader/src/sources/SimpleBehaviorSubject.ts +++ /dev/null @@ -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 { - private currentValue: T; - private terminatingError: Error | undefined; - - readonly observable: ObservableImpl; - - constructor(value: T) { - this.currentValue = value; - this.terminatingError = undefined; - this.observable = new ObservableImpl(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 - >(); - - 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(); - } -} diff --git a/packages/config-loader/src/sources/StaticConfigSource.ts b/packages/config-loader/src/sources/StaticConfigSource.ts index 23fa7d1cfd..3cd5c0c028 100644 --- a/packages/config-loader/src/sources/StaticConfigSource.ts +++ b/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 | Promise; - context?: string; - }): ConfigSource { - const { data, context = 'static-config' } = options; - if (!data) { - return new StaticConfigSource(ObservableImpl.of({}), context); +export interface StaticConfigSourceOptions { + data: + | JsonObject + | Observable + | PromiseLike + | AsyncIterable; + context?: string; +} + +/** @internal */ +class StaticObservableConfigSource implements ConfigSource { + constructor( + private readonly data: Observable, + private readonly context: string, + ) {} + + async *readConfigData( + options?: ReadConfigDataOptions | undefined, + ): AsyncConfigSourceIterator { + const queue = new Array(); + let deferred = simpleDefer(); + + 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), - context, - ); - } - if ('then' in data && typeof data.then === 'function') { - return new StaticConfigSource( - new ObservableImpl(subscriber => { - (data as Promise).then( - value => subscriber.next(value), - error => subscriber.error(error), - ); - return () => {}; - }), - context, - ); - } - - return new StaticConfigSource( - ObservableImpl.of(data as JsonObject), - context, - ); - } - - readonly data$: Observable; - - private constructor(observable: ObservableImpl, context: string) { - this.data$ = observable.map(data => [{ context, data }]); + } +} + +function isObservable(value: {}): value is Observable { + return 'subscribe' in value && typeof (value as any).subscribe === 'function'; +} + +function isAsyncIterable(value: {}): value is AsyncIterable { + 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, + 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, + private readonly context: string, + ) {} + + async *readConfigData(): AsyncConfigSourceIterator { + yield { data: [{ data: await this.promise, context: this.context }] }; + return; } } diff --git a/packages/config-loader/src/sources/types.ts b/packages/config-loader/src/sources/types.ts index 5b0dd1d467..cf1ae80418 100644 --- a/packages/config-loader/src/sources/types.ts +++ b/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; } diff --git a/packages/config-loader/src/sources/utils.ts b/packages/config-loader/src/sources/utils.ts new file mode 100644 index 0000000000..5dfe66aaed --- /dev/null +++ b/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 { + promise: Promise; + resolve(value: T): void; +} + +/** @internal */ +export function simpleDefer(): SimpleDeferred { + let resolve: (value: T) => void; + const promise = new Promise(_resolve => { + resolve = _resolve; + }); + return { promise, resolve: resolve! }; +} + +/** @internal */ +export async function waitOrAbort( + promise: PromiseLike, + 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); + }); +}