Support dynamic refresh (#21)

Author: Eskibear

examples/refresh.mjs

@@ -0,0 +1,44 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import * as dotenv from "dotenv";
+import { promisify } from "util";
+dotenv.config();
+const sleepInMs = promisify(setTimeout);
+
+/**
+ * This example retrives all settings with key following pattern "app.settings.*", i.e. starting with "app.settings.".
+ * With the option `trimKeyPrefixes`, it trims the prefix "app.settings." from keys for simplicity.
+ * Value of config "app.settings.message" will be printed.
+ * It also watches for changes to the key "app.settings.sentinel" and refreshes the configuration when it changes.
+ *
+ * Below environment variables are required for this example:
+ * - APPCONFIG_CONNECTION_STRING
+ */
+
+import { load } from "@azure/app-configuration-provider";
+const connectionString = process.env.APPCONFIG_CONNECTION_STRING;
+const settings = await load(connectionString, {
+    selectors: [{
+        keyFilter: "app.settings.*"
+    }],
+    trimKeyPrefixes: ["app.settings."],
+    refreshOptions: {
+        watchedSettings: [{ key: "app.settings.sentinel" }],
+        refreshIntervalInMs: 10 * 1000 // Default value is 30 seconds, shorted for this sample
+    }
+});
+
+console.log("Using Azure portal or CLI, update the `app.settings.message` value, and then update the `app.settings.sentinel` value in your App Configuration store.")
+
+// eslint-disable-next-line no-constant-condition
+while (true) {
+    // Refreshing the configuration setting
+    await settings.refresh();
+
+    // Current value of message
+    console.log(settings.get("message"));
+
+    // Waiting before the next refresh
+    await sleepInMs(5000);
+}

src/AzureAppConfiguration.ts

@@ -1,6 +1,19 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+import { Disposable } from "./common/disposable";
+
 export type AzureAppConfiguration = {
-    // methods for advanced features, e.g. refresh()
+    /**
+     * API to trigger refresh operation.
+     */
+    refresh(): Promise<void>;
+
+    /**
+     * API to register callback listeners, which will be called only when a refresh operation successfully updates key-values.
+     *
+     * @param listener - Callback funtion to be registered.
+     * @param thisArg - Optional. Value to use as `this` when executing callback.
+     */
+    onRefresh(listener: () => any, thisArg?: any): Disposable;
 } & ReadonlyMap<string, any>;

src/AzureAppConfigurationImpl.ts

@@ -1,28 +1,40 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-import { AppConfigurationClient, ConfigurationSetting, ListConfigurationSettingsOptions } from "@azure/app-configuration";
+import { AppConfigurationClient, ConfigurationSetting, ConfigurationSettingId, GetConfigurationSettingOptions, GetConfigurationSettingResponse, ListConfigurationSettingsOptions } from "@azure/app-configuration";
+import { RestError } from "@azure/core-rest-pipeline";
 import { AzureAppConfiguration } from "./AzureAppConfiguration";
 import { AzureAppConfigurationOptions } from "./AzureAppConfigurationOptions";
 import { IKeyValueAdapter } from "./IKeyValueAdapter";
 import { JsonKeyValueAdapter } from "./JsonKeyValueAdapter";
-import { KeyFilter, LabelFilter } from "./types";
+import { DefaultRefreshIntervalInMs, MinimumRefreshIntervalInMs } from "./RefreshOptions";
+import { Disposable } from "./common/disposable";
 import { AzureKeyVaultKeyValueAdapter } from "./keyvault/AzureKeyVaultKeyValueAdapter";
+import { RefreshTimer } from "./refresh/RefreshTimer";
 import { CorrelationContextHeaderName } from "./requestTracing/constants";
 import { createCorrelationContextHeader, requestTracingEnabled } from "./requestTracing/utils";
-import { SettingSelector } from "./types";
+import { KeyFilter, LabelFilter, SettingSelector } from "./types";
 
-export class AzureAppConfigurationImpl extends Map<string, unknown> implements AzureAppConfiguration {
+export class AzureAppConfigurationImpl extends Map<string, any> implements AzureAppConfiguration {
     #adapters: IKeyValueAdapter[] = [];
     /**
      * Trim key prefixes sorted in descending order.
      * Since multiple prefixes could start with the same characters, we need to trim the longest prefix first.
      */
     #sortedTrimKeyPrefixes: string[] | undefined;
     readonly #requestTracingEnabled: boolean;
-    #correlationContextHeader: string | undefined;
     #client: AppConfigurationClient;
     #options: AzureAppConfigurationOptions | undefined;
+    #isInitialLoadCompleted: boolean = false;
+
+    // Refresh
+    #refreshInterval: number = DefaultRefreshIntervalInMs;
+    #onRefreshListeners: Array<() => any> = [];
+    /**
+     * Aka watched settings.
+     */
+    #sentinels: ConfigurationSettingId[] = [];
+    #refreshTimer: RefreshTimer;
 
     constructor(
         client: AppConfigurationClient,
@@ -34,21 +46,54 @@ export class AzureAppConfigurationImpl extends Map<string, unknown> implements A
 
         // Enable request tracing if not opt-out
         this.#requestTracingEnabled = requestTracingEnabled();
-        if (this.#requestTracingEnabled) {
-            this.#enableRequestTracing();
-        }
 
         if (options?.trimKeyPrefixes) {
             this.#sortedTrimKeyPrefixes = [...options.trimKeyPrefixes].sort((a, b) => b.localeCompare(a));
         }
+
+        if (options?.refreshOptions?.enabled) {
+            const { watchedSettings, refreshIntervalInMs } = options.refreshOptions;
+            // validate watched settings
+            if (watchedSettings === undefined || watchedSettings.length === 0) {
+                throw new Error("Refresh is enabled but no watched settings are specified.");
+            }
+
+            // custom refresh interval
+            if (refreshIntervalInMs !== undefined) {
+                if (refreshIntervalInMs < MinimumRefreshIntervalInMs) {
+                    throw new Error(`The refresh interval cannot be less than ${MinimumRefreshIntervalInMs} milliseconds.`);
+
+                } else {
+                    this.#refreshInterval = refreshIntervalInMs;
+                }
+            }
+
+            for (const setting of watchedSettings) {
+                if (setting.key.includes("*") || setting.key.includes(",")) {
+                    throw new Error("The characters '*' and ',' are not supported in key of watched settings.");
+                }
+                if (setting.label?.includes("*") || setting.label?.includes(",")) {
+                    throw new Error("The characters '*' and ',' are not supported in label of watched settings.");
+                }
+                this.#sentinels.push(setting);
+            }
+
+            this.#refreshTimer = new RefreshTimer(this.#refreshInterval);
+        }
+
         // TODO: should add more adapters to process different type of values
         // feature flag, others
         this.#adapters.push(new AzureKeyVaultKeyValueAdapter(options?.keyVaultOptions));
         this.#adapters.push(new JsonKeyValueAdapter());
     }
 
-    async load() {
-        const keyValues: [key: string, value: unknown][] = [];
+
+    get #refreshEnabled(): boolean {
+        return !!this.#options?.refreshOptions?.enabled;
+    }
+
+    async #loadSelectedKeyValues(): Promise<ConfigurationSetting[]> {
+        const loadedSettings: ConfigurationSetting[] = [];
 
         // validate selectors
         const selectors = getValidSelectors(this.#options?.selectors);
@@ -60,25 +105,142 @@ export class AzureAppConfigurationImpl extends Map<string, unknown> implements A
             };
             if (this.#requestTracingEnabled) {
                 listOptions.requestOptions = {
-                    customHeaders: this.#customHeaders()
+                    customHeaders: {
+                        [CorrelationContextHeaderName]: createCorrelationContextHeader(this.#options, this.#isInitialLoadCompleted)
+                    }
                 }
             }
 
             const settings = this.#client.listConfigurationSettings(listOptions);
 
             for await (const setting of settings) {
-                if (setting.key) {
-                    const [key, value] = await this.#processAdapters(setting);
-                    const trimmedKey = this.#keyWithPrefixesTrimmed(key);
-                    keyValues.push([trimmedKey, value]);
+                loadedSettings.push(setting);
+            }
+        }
+        return loadedSettings;
+    }
+
+    /**
+     * Update etag of watched settings from loaded data. If a watched setting is not covered by any selector, a request will be sent to retrieve it.
+     */
+    async #updateWatchedKeyValuesEtag(existingSettings: ConfigurationSetting[]): Promise<void> {
+        if (!this.#refreshEnabled) {
+            return;
+        }
+
+        for (const sentinel of this.#sentinels) {
+            const matchedSetting = existingSettings.find(s => s.key === sentinel.key && s.label === sentinel.label);
+            if (matchedSetting) {
+                sentinel.etag = matchedSetting.etag;
+            } else {
+                // Send a request to retrieve key-value since it may be either not loaded or loaded with a different label or different casing
+                const { key, label } = sentinel;
+                const response = await this.#getConfigurationSettingWithTrace({ key, label });
+                if (response) {
+                    sentinel.etag = response.etag;
+                } else {
+                    sentinel.etag = undefined;
                 }
             }
         }
+    }
+
+    async #loadSelectedAndWatchedKeyValues() {
+        const keyValues: [key: string, value: unknown][] = [];
+
+        const loadedSettings = await this.#loadSelectedKeyValues();
+        await this.#updateWatchedKeyValuesEtag(loadedSettings);
+
+        // process key-values, watched settings have higher priority
+        for (const setting of loadedSettings) {
+            const [key, value] = await this.#processKeyValues(setting);
+            keyValues.push([key, value]);
+        }
+
+        this.clear(); // clear existing key-values in case of configuration setting deletion
         for (const [k, v] of keyValues) {
             this.set(k, v);
         }
     }
 
+    /**
+     * Load the configuration store for the first time.
+     */
+    async load() {
+        await this.#loadSelectedAndWatchedKeyValues();
+
+        // Mark all settings have loaded at startup.
+        this.#isInitialLoadCompleted = true;
+    }
+
+    /**
+     * Refresh the configuration store.
+     */
+    public async refresh(): Promise<void> {
+        if (!this.#refreshEnabled) {
+            throw new Error("Refresh is not enabled.");
+        }
+
+        // if still within refresh interval/backoff, return
+        if (!this.#refreshTimer.canRefresh()) {
+            return Promise.resolve();
+        }
+
+        // try refresh if any of watched settings is changed.
+        let needRefresh = false;
+        for (const sentinel of this.#sentinels.values()) {
+            const response = await this.#getConfigurationSettingWithTrace(sentinel, {
+                onlyIfChanged: true
+            });
+
+            if (response?.statusCode === 200 // created or changed
+                || (response === undefined && sentinel.etag !== undefined) // deleted
+            ) {
+                sentinel.etag = response?.etag;// update etag of the sentinel
+                needRefresh = true;
+                break;
+            }
+        }
+        if (needRefresh) {
+            try {
+                await this.#loadSelectedAndWatchedKeyValues();
+                this.#refreshTimer.reset();
+            } catch (error) {
+                // if refresh failed, backoff
+                this.#refreshTimer.backoff();
+                throw error;
+            }
+
+            // successfully refreshed, run callbacks in async
+            for (const listener of this.#onRefreshListeners) {
+                listener();
+            }
+        }
+    }
+
+    onRefresh(listener: () => any, thisArg?: any): Disposable {
+        if (!this.#refreshEnabled) {
+            throw new Error("Refresh is not enabled.");
+        }
+
+        const boundedListener = listener.bind(thisArg);
+        this.#onRefreshListeners.push(boundedListener);
+
+        const remove = () => {
+            const index = this.#onRefreshListeners.indexOf(boundedListener);
+            if (index >= 0) {
+                this.#onRefreshListeners.splice(index, 1);
+            }
+        }
+        return new Disposable(remove);
+    }
+
+    async #processKeyValues(setting: ConfigurationSetting<string>): Promise<[string, unknown]> {
+        const [key, value] = await this.#processAdapters(setting);
+        const trimmedKey = this.#keyWithPrefixesTrimmed(key);
+        return [trimmedKey, value];
+    }
+
     async #processAdapters(setting: ConfigurationSetting<string>): Promise<[string, unknown]> {
         for (const adapter of this.#adapters) {
             if (adapter.canProcess(setting)) {
@@ -99,18 +261,26 @@ export class AzureAppConfigurationImpl extends Map<string, unknown> implements A
         return key;
     }
 
-    #enableRequestTracing() {
-        this.#correlationContextHeader = createCorrelationContextHeader(this.#options);
-    }
-
-    #customHeaders() {
-        if (!this.#requestTracingEnabled) {
-            return undefined;
+    async #getConfigurationSettingWithTrace(configurationSettingId: ConfigurationSettingId, customOptions?: GetConfigurationSettingOptions): Promise<GetConfigurationSettingResponse | undefined> {
+        let response: GetConfigurationSettingResponse | undefined;
+        try {
+            const options = { ...customOptions ?? {} };
+            if (this.#requestTracingEnabled) {
+                options.requestOptions = {
+                    customHeaders: {
+                        [CorrelationContextHeaderName]: createCorrelationContextHeader(this.#options, this.#isInitialLoadCompleted)
+                    }
+                }
+            }
+            response = await this.#client.getConfigurationSetting(configurationSettingId, options);
+        } catch (error) {
+            if (error instanceof RestError && error.statusCode === 404) {
+                response = undefined;
+            } else {
+                throw error;
+            }
         }
-
-        const headers = {};
-        headers[CorrelationContextHeaderName] = this.#correlationContextHeader;
-        return headers;
+        return response;
     }
 }
 
@@ -143,4 +313,4 @@ function getValidSelectors(selectors?: SettingSelector[]) {
         }
         return selector;
     });
-}
+}

src/AzureAppConfigurationOptions.ts

@@ -3,6 +3,7 @@
 
 import { AppConfigurationClientOptions } from "@azure/app-configuration";
 import { KeyVaultOptions } from "./keyvault/KeyVaultOptions";
+import { RefreshOptions } from "./RefreshOptions";
 import { SettingSelector } from "./types";
 
 export const MaxRetries = 2;
@@ -35,4 +36,8 @@ export interface AzureAppConfigurationOptions {
      * Specifies options used to resolve Vey Vault references.
      */
     keyVaultOptions?: KeyVaultOptions;
+    /**
+     * Specifies options for dynamic refresh key-values.
+     */
+    refreshOptions?: RefreshOptions;
 }