Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: Adds LDDataSystemOptions for configuring the Data System. #794

Draft
wants to merge 1 commit into
base: ta/sdk-857/composite-datasource
Choose a base branch
from
Draft

chore: Adds LDDataSystemOptions for configuring the Data System. #794

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 44 additions & 49 deletions packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,28 +1,23 @@
import {
CompositeDataSource,
Transition,
TransitionConditions,
} from '../../../src/api/subsystem/DataSystem/CompositeDataSource';
import {
Data,
DataSourceState,
DataSystemInitializer,
DataSystemSynchronizer,
InitializerFactory,
SynchronizerFactory,
LDInitializerFactory,
LDSynchronizerFactory,
} from '../../../src/api/subsystem/DataSystem/DataSource';
import { Backoff } from '../../../src/datasource/Backoff';

function makeInitializerFactory(internal: DataSystemInitializer): InitializerFactory {
return {
create: () => internal,
};
function makeInitializerFactory(internal: DataSystemInitializer): LDInitializerFactory {
return () => internal;
}

function makeSynchronizerFactory(internal: DataSystemSynchronizer): SynchronizerFactory {
return {
create: () => internal,
};
function makeSynchronizerFactory(internal: DataSystemSynchronizer): LDSynchronizerFactory {
return () => internal;
}

function makeTestTransitionConditions(): TransitionConditions {
Expand Down Expand Up @@ -59,7 +54,7 @@ function makeZeroBackoff(): Backoff {

it('initializer gets basis, switch to syncrhonizer', async () => {
const mockInitializer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -74,7 +69,7 @@ it('initializer gets basis, switch to syncrhonizer', async () => {

const mockSynchronizer1Data = { key: 'sync1' };
const mockSynchronizer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand Down Expand Up @@ -102,19 +97,19 @@ it('initializer gets basis, switch to syncrhonizer', async () => {
}
});

underTest.run(callback, jest.fn());
underTest.start(callback, jest.fn());
});

expect(mockInitializer1.run).toHaveBeenCalledTimes(1);
expect(mockSynchronizer1.run).toHaveBeenCalledTimes(1);
expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
expect(callback).toHaveBeenCalledTimes(2);
expect(callback).toHaveBeenNthCalledWith(1, true, { key: 'init1' });
expect(callback).toHaveBeenNthCalledWith(2, false, { key: 'sync1' });
});

it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2, recover to synchronizer 1', async () => {
const mockInitializer1: DataSystemInitializer = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -130,7 +125,7 @@ it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2
let sync1RunCount = 0;
const mockSynchronizer1Data = { key: 'sync1' };
const mockSynchronizer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -143,7 +138,7 @@ it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2
message: 'I am error...man!',
}); // error that will lead to fallback
} else {
_dataCallback(false, mockSynchronizer1Data); // second run will lead to data
_dataCallback(false, mockSynchronizer1Data); // second start will lead to data
}
sync1RunCount += 1;
},
Expand All @@ -153,7 +148,7 @@ it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2

const mockSynchronizer2Data = { key: 'sync2' };
const mockSynchronizer2 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand Down Expand Up @@ -182,12 +177,12 @@ it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2
}
});

underTest.run(callback, jest.fn());
underTest.start(callback, jest.fn());
});

expect(mockInitializer1.run).toHaveBeenCalledTimes(1);
expect(mockSynchronizer1.run).toHaveBeenCalledTimes(2);
expect(mockSynchronizer2.run).toHaveBeenCalledTimes(1);
expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
expect(mockSynchronizer1.start).toHaveBeenCalledTimes(2);
expect(mockSynchronizer2.start).toHaveBeenCalledTimes(1);
expect(callback).toHaveBeenCalledTimes(3);
expect(callback).toHaveBeenNthCalledWith(1, true, { key: 'init1' });
expect(callback).toHaveBeenNthCalledWith(2, false, { key: 'sync2' }); // sync1 errors and fallsback
Expand All @@ -200,7 +195,7 @@ it('it reports error when all initializers fail', async () => {
message: 'I am initializer1 error!',
};
const mockInitializer1: DataSystemInitializer = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -218,7 +213,7 @@ it('it reports error when all initializers fail', async () => {
message: 'I am initializer2 error!',
};
const mockInitializer2: DataSystemInitializer = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand Down Expand Up @@ -247,11 +242,11 @@ it('it reports error when all initializers fail', async () => {
}
});

underTest.run(dataCallback, statusCallback);
underTest.start(dataCallback, statusCallback);
});

expect(mockInitializer1.run).toHaveBeenCalledTimes(1);
expect(mockInitializer2.run).toHaveBeenCalledTimes(1);
expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
expect(mockInitializer2.start).toHaveBeenCalledTimes(1);
expect(dataCallback).toHaveBeenCalledTimes(0);
expect(statusCallback).toHaveBeenNthCalledWith(
1,
Expand All @@ -273,7 +268,7 @@ it('it reports error when all initializers fail', async () => {

it('it can be stopped when in thrashing synchronizer fallback loop', async () => {
const mockInitializer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -288,7 +283,7 @@ it('it can be stopped when in thrashing synchronizer fallback loop', async () =>

const mockSynchronizer1Error = { name: 'Error', message: 'I am error...man!' };
const mockSynchronizer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand Down Expand Up @@ -317,11 +312,11 @@ it('it can be stopped when in thrashing synchronizer fallback loop', async () =>
}
});

underTest.run(dataCallback, statusCallback);
underTest.start(dataCallback, statusCallback);
});

expect(mockInitializer1.run).toHaveBeenCalled();
expect(mockSynchronizer1.run).toHaveBeenCalled();
expect(mockInitializer1.start).toHaveBeenCalled();
expect(mockSynchronizer1.start).toHaveBeenCalled();
expect(dataCallback).toHaveBeenNthCalledWith(1, true, { key: 'init1' });
underTest.stop();

Expand All @@ -341,7 +336,7 @@ it('it can be stopped when in thrashing synchronizer fallback loop', async () =>
it('it can be stopped and restarted', async () => {
const mockInitializer1Data = { key: 'init1' };
const mockInitializer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -356,7 +351,7 @@ it('it can be stopped and restarted', async () => {

const mockSynchronizer1Data = { key: 'sync1' };
const mockSynchronizer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand Down Expand Up @@ -384,13 +379,13 @@ it('it can be stopped and restarted', async () => {
resolve();
}
});
// first run
underTest.run(callback1, jest.fn());
// first start
underTest.start(callback1, jest.fn());
});

// check first run triggered underlying data sources
expect(mockInitializer1.run).toHaveBeenCalledTimes(1);
expect(mockSynchronizer1.run).toHaveBeenCalledTimes(1);
// check first start triggered underlying data sources
expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
expect(callback1).toHaveBeenCalledTimes(2);

// wait a moment for pending awaits to resolve the stop request
Expand All @@ -405,13 +400,13 @@ it('it can be stopped and restarted', async () => {
resolve();
}
});
// second run
underTest.run(callback2, jest.fn());
// second start
underTest.start(callback2, jest.fn());
});

// check that second run triggers underlying data sources again
expect(mockInitializer1.run).toHaveBeenCalledTimes(2);
expect(mockSynchronizer1.run).toHaveBeenCalledTimes(2);
// check that second start triggers underlying data sources again
expect(mockInitializer1.start).toHaveBeenCalledTimes(2);
expect(mockSynchronizer1.start).toHaveBeenCalledTimes(2);
expect(callback2).toHaveBeenCalledTimes(2);
});

Expand All @@ -429,7 +424,7 @@ it('it is well behaved with no initializers and no synchronizers configured', as
resolve();
});

underTest.run(jest.fn(), statusCallback);
underTest.start(jest.fn(), statusCallback);
});

expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Closed, {
Expand All @@ -441,7 +436,7 @@ it('it is well behaved with no initializers and no synchronizers configured', as

it('it is well behaved with an initializer and no synchronizers configured', async () => {
const mockInitializer1 = {
run: jest
start: jest
.fn()
.mockImplementation(
(
Expand All @@ -467,7 +462,7 @@ it('it is well behaved with an initializer and no synchronizers configured', asy
resolve();
});

underTest.run(jest.fn(), statusCallback);
underTest.start(jest.fn(), statusCallback);
});

expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Closed, {
Expand Down
17 changes: 9 additions & 8 deletions packages/shared/common/src/api/subsystem/DataSystem/CompositeDataSource.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ import {
Data,
DataSource,
DataSourceState,
InitializerFactory,
SynchronizerFactory,
LDInitializerFactory,
LDSynchronizerFactory,
} from './DataSource';

// TODO: SDK-858, specify these constants when CompositeDataSource is used.
Expand All @@ -32,6 +32,7 @@ interface TransitionRequest {
err?: Error;
}

// TODO SDK-858: move this out of API directory to neighbor datasource folder
/**
* The {@link CompositeDataSource} can combine a number of {@link DataSystemInitializer}s and {@link DataSystemSynchronizer}s
* into a single {@link DataSource}, implementing fallback and recovery logic internally to choose where data is sourced from.
Expand All @@ -52,8 +53,8 @@ export class CompositeDataSource implements DataSource {
* @param _synchronizers factories to create {@link DataSystemSynchronizer}s, in priority order.
*/
constructor(
private readonly _initializers: InitializerFactory[],
private readonly _synchronizers: SynchronizerFactory[],
private readonly _initializers: LDInitializerFactory[],
private readonly _synchronizers: LDSynchronizerFactory[],
private readonly _transitionConditions: TransitionConditions,
private readonly _backoff: Backoff,
) {
Expand All @@ -64,7 +65,7 @@ export class CompositeDataSource implements DataSource {
this._currentPosition = 0;
}

async run(
async start(
dataCallback: (basis: boolean, data: Data) => void,
statusCallback: (status: DataSourceState, err?: any) => void,
): Promise<void> {
Expand Down Expand Up @@ -128,7 +129,7 @@ export class CompositeDataSource implements DataSource {
}
},
);
currentDS.run(
currentDS.start(
(basis, data) => callbackHandler.dataHanlder(basis, data),
(status, err) => callbackHandler.statusHandler(status, err),
);
Expand Down Expand Up @@ -218,7 +219,7 @@ export class CompositeDataSource implements DataSource {
return undefined;
}

return this._initializers[this._currentPosition].create();
return this._initializers[this._currentPosition]();
}
// getting here indicates we are using a synchronizer

Expand All @@ -231,7 +232,7 @@ export class CompositeDataSource implements DataSource {
// this is only possible if no synchronizers were provided
return undefined;
}
return this._synchronizers[this._currentPosition].create();
return this._synchronizers[this._currentPosition]();
}

/**
Expand Down
15 changes: 6 additions & 9 deletions packages/shared/common/src/api/subsystem/DataSystem/DataSource.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ export interface DataSource {
* @param statusCallback that will be called when data source state changes or an unrecoverable error
* has been encountered.
*/
run(
start(
dataCallback: (basis: boolean, data: Data) => void,
statusCallback: (status: DataSourceState, err?: any) => void,
): void;
Expand All @@ -26,6 +26,11 @@ export interface DataSource {
stop(): void;
}

export type LDInitializerFactory = () => DataSystemInitializer;

export type LDSynchronizerFactory = () => DataSystemSynchronizer;

// TODO: renmae these to start with LD to help with API consumption
/**
* A data source that can be used to fetch the basis.
*/
Expand All @@ -35,11 +40,3 @@ export interface DataSystemInitializer extends DataSource {}
* A data source that can be used to fetch the basis or ongoing data changes.
*/
export interface DataSystemSynchronizer extends DataSource {}

export interface InitializerFactory {
create(): DataSystemInitializer;
}

export interface SynchronizerFactory {
create(): DataSystemSynchronizer;
}
4 changes: 2 additions & 2 deletions packages/shared/common/src/api/subsystem/DataSystem/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
export {
DataSystemInitializer,
DataSystemSynchronizer,
InitializerFactory,
SynchronizerFactory,
LDInitializerFactory as InitializerFactory,
LDSynchronizerFactory as SynchronizerFactory,
} from './DataSource';
export { CompositeDataSource } from './CompositeDataSource';
10 changes: 10 additions & 0 deletions packages/shared/common/src/api/subsystem/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,19 @@
import {
DataSystemInitializer,
DataSystemSynchronizer,
InitializerFactory,
SynchronizerFactory,
} from './DataSystem';
import LDContextDeduplicator from './LDContextDeduplicator';
import LDEventProcessor from './LDEventProcessor';
import LDEventSender, { LDDeliveryStatus, LDEventSenderResult, LDEventType } from './LDEventSender';
import { LDStreamProcessor } from './LDStreamProcessor';

export {
DataSystemInitializer,
DataSystemSynchronizer,
InitializerFactory,
SynchronizerFactory,
LDEventProcessor,
LDContextDeduplicator,
LDEventSender,
Expand Down
Loading