feat: implement the custom data source option

this enables users to specify file data source initializer

Author: joker23

packages/shared/common/__tests__/internal/fdv2/FDv2ChangeSetBuilder.test.ts renamed to packages/shared/common/__tests__/internal/fdv2/FDv1PayloadAdaptor.test.ts

@@ -1004,3 +1004,157 @@ it('consumes cancellation tokens correctly', async () => {
   // eslint-disable-next-line no-underscore-dangle
   expect(underTest._cancelTokens.length).toEqual(0);
 });
+
+it('handles multiple initializers with fallback when first initializer fails and second succeeds', async () => {
+  const mockInitializer1Error = {
+    name: 'Error',
+    message: 'First initializer failed',
+  };
+  const mockInitializer1: DataSource = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Closed, mockInitializer1Error);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const mockInitializer2Data = { key: 'init2' };
+  const mockInitializer2: DataSource = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Valid);
+          _dataCallback(true, mockInitializer2Data);
+          _statusCallback(DataSourceState.Closed);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const mockSynchronizer1Data = { key: 'sync1' };
+  const mockSynchronizer1 = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Valid);
+          _dataCallback(false, mockSynchronizer1Data);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const underTest = new CompositeDataSource(
+    [makeDataSourceFactory(mockInitializer1), makeDataSourceFactory(mockInitializer2)],
+    [makeDataSourceFactory(mockSynchronizer1)],
+    [],
+    undefined,
+    makeTestTransitionConditions(),
+    makeZeroBackoff(),
+  );
+
+  let dataCallback;
+  const statusCallback = jest.fn();
+  await new Promise<void>((resolve) => {
+    dataCallback = jest.fn((_: boolean, data: any) => {
+      if (data === mockSynchronizer1Data) {
+        resolve();
+      }
+    });
+
+    underTest.start(dataCallback, statusCallback);
+  });
+
+  expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
+  expect(mockInitializer2.start).toHaveBeenCalledTimes(1);
+  expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
+  expect(statusCallback).toHaveBeenCalledTimes(5);
+  expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Initializing, undefined);
+  expect(statusCallback).toHaveBeenNthCalledWith(2, DataSourceState.Interrupted, mockInitializer1Error);
+  expect(statusCallback).toHaveBeenNthCalledWith(3, DataSourceState.Valid, undefined);
+  expect(statusCallback).toHaveBeenNthCalledWith(4, DataSourceState.Interrupted, undefined);
+  expect(statusCallback).toHaveBeenNthCalledWith(5, DataSourceState.Valid, undefined);
+});
+
+it('does not run second initializer when first initializer succeeds', async () => {
+  const mockInitializer1Data = { key: 'init1' };
+  const mockInitializer1: DataSource = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Valid);
+          _dataCallback(true, mockInitializer1Data);
+          _statusCallback(DataSourceState.Closed);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const mockInitializer2: DataSource = {
+    start: jest.fn(),
+    stop: jest.fn(),
+  };
+
+  const mockSynchronizer1Data = { key: 'sync1' };
+  const mockSynchronizer1 = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Valid);
+          _dataCallback(false, mockSynchronizer1Data);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const underTest = new CompositeDataSource(
+    [makeDataSourceFactory(mockInitializer1), makeDataSourceFactory(mockInitializer2)],
+    [makeDataSourceFactory(mockSynchronizer1)],
+    [],
+    undefined,
+    makeTestTransitionConditions(),
+    makeZeroBackoff(),
+  );
+
+  let dataCallback;
+  const statusCallback = jest.fn();
+  await new Promise<void>((resolve) => {
+    dataCallback = jest.fn((_: boolean, data: any) => {
+      if (data === mockSynchronizer1Data) {
+        resolve();
+      }
+    });
+
+    underTest.start(dataCallback, statusCallback);
+  });
+
+  expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
+  expect(mockInitializer2.start).toHaveBeenCalledTimes(0);
+  expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
+});

packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts

@@ -34,6 +34,7 @@ import { Hook } from './api/integrations/Hook';
 import { BigSegmentStoreMembership } from './api/interfaces';
 import { LDWaitForInitializationOptions } from './api/LDWaitForInitializationOptions';
 import {
+  isCustomOptions,
   isPollingOnlyOptions,
   isStandardOptions,
   isStreamingOnlyOptions,
@@ -65,9 +66,10 @@ import FlagsStateBuilder from './FlagsStateBuilder';
 import HookRunner from './hooks/HookRunner';
 import MigrationOpEventToInputEvent from './MigrationOpEventConversion';
 import MigrationOpTracker from './MigrationOpTracker';
-import Configuration, { DEFAULT_POLL_INTERVAL } from './options/Configuration';
+import Configuration, { DEFAULT_POLL_INTERVAL, DEFAULT_STREAM_RECONNECT_DELAY } from './options/Configuration';
 import { ServerInternalOptions } from './options/ServerInternalOptions';
 import VersionedDataKinds from './store/VersionedDataKinds';
+import FileDataInitializerFDv2 from './data_sources/fileDataInitilizerFDv2';
 
 const { ClientMessages, ErrorKinds, NullEventProcessor } = internal;
 enum InitState {
@@ -323,60 +325,96 @@ function constructFDv2(
   if (!(config.offline || config.dataSystem!.useLdd)) {
     // make the FDv2 composite datasource with initializers/synchronizers
     const initializers: subsystem.LDDataSourceFactory[] = [];
+    const synchronizers: subsystem.LDDataSourceFactory[] = [];
+    const fdv1FallbackSynchronizers: subsystem.LDDataSourceFactory[] = [];
+
+    if (isCustomOptions(dataSystem.dataSource)) {
+      const {
+        initializers: initializerConfigs = [],
+        synchronizers: synchronizerConfigs = [],
+      } = dataSystem.dataSource;
+
+      initializerConfigs.forEach((initializerConfig) => {
+        switch (initializerConfig.type) {
+          case 'file':
+            initializers.push(() => new FileDataInitializerFDv2(initializerConfig, platform, config.logger));
+            break;
+          case 'polling':
+            const { pollInterval = DEFAULT_POLL_INTERVAL } = initializerConfig;
+            initializers.push(() => new PollingProcessorFDv2(new Requestor(config, platform.requests, baseHeaders, '/sdk/poll', config.logger), pollInterval, config.logger));
+            break;
+          default:
+            throw new Error('Unsupported initializer type');
+        }
+      });
+      synchronizerConfigs.forEach((synchronizerConfig) => {
+        switch (synchronizerConfig.type) {
+          case 'streaming':
+            const { streamInitialReconnectDelay = DEFAULT_STREAM_RECONNECT_DELAY } = synchronizerConfig;
+            synchronizers.push(() => new StreamingProcessorFDv2(clientContext, '/sdk/stream', [], baseHeaders, diagnosticsManager, streamInitialReconnectDelay));
+            break;
+          case 'polling':
+            const { pollInterval = DEFAULT_POLL_INTERVAL } = synchronizerConfig;
+            synchronizers.push(() => new PollingProcessorFDv2(new Requestor(config, platform.requests, baseHeaders, '/sdk/poll', config.logger), pollInterval, config.logger));
+            break;
+          default:
+            throw new Error('Unsupported synchronizer type');
+          }
+      });
+    } else {
+      // use one shot initializer for performance and cost if we can do a combination of polling and streaming
+      if (isStandardOptions(dataSystem.dataSource)) {
+        initializers.push(
+          () =>
+            new OneShotInitializerFDv2(
+              new Requestor(config, platform.requests, baseHeaders, '/sdk/poll', config.logger),
+              config.logger,
+            ),
+        );
+      }
 
-    // use one shot initializer for performance and cost if we can do a combination of polling and streaming
-    if (isStandardOptions(dataSystem.dataSource)) {
-      initializers.push(
-        () =>
-          new OneShotInitializerFDv2(
-            new Requestor(config, platform.requests, baseHeaders, '/sdk/poll', config.logger),
-            config.logger,
-          ),
-      );
-    }
+      // if streaming is configured, add streaming synchronizer
+      if (isStandardOptions(dataSystem.dataSource) || isStreamingOnlyOptions(dataSystem.dataSource)) {
+        const reconnectDelay = dataSystem.dataSource.streamInitialReconnectDelay;
+        synchronizers.push(
+          () =>
+            new StreamingProcessorFDv2(
+              clientContext,
+              '/sdk/stream',
+              [],
+              baseHeaders,
+              diagnosticsManager,
+              reconnectDelay,
+            ),
+        );
+      }
 
-    const synchronizers: subsystem.LDDataSourceFactory[] = [];
-    // if streaming is configured, add streaming synchronizer
-    if (isStandardOptions(dataSystem.dataSource) || isStreamingOnlyOptions(dataSystem.dataSource)) {
-      const reconnectDelay = dataSystem.dataSource.streamInitialReconnectDelay;
-      synchronizers.push(
-        () =>
-          new StreamingProcessorFDv2(
-            clientContext,
-            '/sdk/stream',
-            [],
-            baseHeaders,
-            diagnosticsManager,
-            reconnectDelay,
-          ),
-      );
-    }
+      let pollingInterval = DEFAULT_POLL_INTERVAL;
+      // if polling is configured, add polling synchronizer
+      if (isStandardOptions(dataSystem.dataSource) || isPollingOnlyOptions(dataSystem.dataSource)) {
+        pollingInterval = dataSystem.dataSource.pollInterval ?? DEFAULT_POLL_INTERVAL;
+        synchronizers.push(
+          () =>
+            new PollingProcessorFDv2(
+              new Requestor(config, platform.requests, baseHeaders, '/sdk/poll', logger),
+              pollingInterval,
+              logger,
+            ),
+        );
+      }
 
-    let pollingInterval = DEFAULT_POLL_INTERVAL;
-    // if polling is configured, add polling synchronizer
-    if (isStandardOptions(dataSystem.dataSource) || isPollingOnlyOptions(dataSystem.dataSource)) {
-      pollingInterval = dataSystem.dataSource.pollInterval ?? DEFAULT_POLL_INTERVAL;
-      synchronizers.push(
+      // This is short term handling and will be removed once FDv2 adoption is sufficient.
+      fdv1FallbackSynchronizers.push(
         () =>
           new PollingProcessorFDv2(
-            new Requestor(config, platform.requests, baseHeaders, '/sdk/poll', logger),
+            new Requestor(config, platform.requests, baseHeaders, '/sdk/latest-all', config.logger),
             pollingInterval,
-            logger,
+            config.logger,
+            true,
           ),
-      );
+        );
     }
 
-    // This is short term handling and will be removed once FDv2 adoption is sufficient.
-    const fdv1FallbackSynchronizers = [
-      () =>
-        new PollingProcessorFDv2(
-          new Requestor(config, platform.requests, baseHeaders, '/sdk/latest-all', config.logger),
-          pollingInterval,
-          config.logger,
-          true,
-        ),
-    ];
-
     dataSource = new CompositeDataSource(
       initializers,
       synchronizers,

packages/shared/sdk-server/src/LDClientImpl.ts

@@ -73,7 +73,8 @@ export interface LDDataSystemOptions {
 export type DataSourceOptions =
   | StandardDataSourceOptions
   | StreamingDataSourceOptions
-  | PollingDataSourceOptions;
+  | PollingDataSourceOptions
+  | CustomDataSourceOptions;
 
 export type DataSourceConfiguration =
   | FileSystemDataSourceConfiguration
@@ -187,3 +188,7 @@ export function isStreamingOnlyOptions(u: any): u is StreamingDataSourceOptions
 export function isPollingOnlyOptions(u: any): u is PollingDataSourceOptions {
   return u.dataSourceOptionsType === 'pollingOnly';
 }
+
+export function isCustomOptions(u: any): u is CustomDataSourceOptions {
+  return u.dataSourceOptionsType === 'custom';
+}

packages/shared/sdk-server/src/api/options/LDDataSystemOptions.ts

@@ -24,7 +24,7 @@ export default class FileDataInitializerFDv2 implements subsystemCommon.DataSour
   private _yamlParser?: (data: string) => any;
   private _fileLoader?: FileLoader;
 
-  constructor(options: FileSystemDataSourceConfiguration, platform: Platform, logger: LDLogger) {
+  constructor(options: FileSystemDataSourceConfiguration, platform: Platform, logger?: LDLogger) {
     this._validateInputs(options, platform);
 
     this._paths = options.paths;

packages/shared/sdk-server/src/data_sources/fileDataInitilizerFDv2.ts

@@ -16,13 +16,15 @@ import { LDBigSegmentsOptions, LDOptions, LDProxyOptions, LDTLSOptions } from '.
 import { Hook } from '../api/integrations';
 import {
   DataSourceOptions,
+  isCustomOptions,
   isPollingOnlyOptions,
   isStandardOptions,
   isStreamingOnlyOptions,
   LDDataSystemOptions,
   PollingDataSourceOptions,
   StandardDataSourceOptions,
   StreamingDataSourceOptions,
+  CustomDataSourceOptions,
 } from '../api/options/LDDataSystemOptions';
 import {
   LDDataSourceUpdates,
@@ -78,7 +80,7 @@ const validations: Record<string, TypeValidator> = {
 };
 
 export const DEFAULT_POLL_INTERVAL = 30;
-const DEFAULT_STREAM_RECONNECT_DELAY = 1;
+export const DEFAULT_STREAM_RECONNECT_DELAY = 1;
 
 const defaultStandardDataSourceOptions: StandardDataSourceOptions = {
   dataSourceOptionsType: 'standard',
@@ -237,6 +239,9 @@ function validateDataSystemOptions(options: Options): {
         options.dataSource,
         defaultPollingDataSourceOptions,
       ));
+    } else if (isCustomOptions(options.dataSource)) {
+      validatedDataSourceOptions = options.dataSource;
+      errors = [];
     } else {
       // provided datasource options don't fit any expected form, drop them and use defaults
       validatedDataSourceOptions = defaultStandardDataSourceOptions;