Refactor State Providers (#8273)
* Delete A Lot Of Code * Fix Tests * Create SingleUserState Provider Once * Update Manual Instantiations * Fix Service Factory * Delete More * Delete Unused `updatePromise` * `postStorageSave` -> `doStorageSave` * Update Comment * Fix jslib-services
This commit is contained in:
parent
4f8fa57b9d
commit
1d76e80afb
|
@ -409,8 +409,7 @@ export default class MainBackground {
|
|||
);
|
||||
this.activeUserStateProvider = new DefaultActiveUserStateProvider(
|
||||
this.accountService,
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
this.singleUserStateProvider,
|
||||
);
|
||||
this.derivedStateProvider = new BackgroundDerivedStateProvider(
|
||||
this.memoryStorageForStateProviders,
|
||||
|
|
|
@ -9,20 +9,15 @@ import {
|
|||
|
||||
import { CachedServices, FactoryOptions, factory } from "./factory-options";
|
||||
import {
|
||||
StateEventRegistrarServiceInitOptions,
|
||||
stateEventRegistrarServiceFactory,
|
||||
} from "./state-event-registrar-service.factory";
|
||||
import {
|
||||
StorageServiceProviderInitOptions,
|
||||
storageServiceProviderFactory,
|
||||
} from "./storage-service-provider.factory";
|
||||
SingleUserStateProviderInitOptions,
|
||||
singleUserStateProviderFactory,
|
||||
} from "./single-user-state-provider.factory";
|
||||
|
||||
type ActiveUserStateProviderFactory = FactoryOptions;
|
||||
|
||||
export type ActiveUserStateProviderInitOptions = ActiveUserStateProviderFactory &
|
||||
AccountServiceInitOptions &
|
||||
StorageServiceProviderInitOptions &
|
||||
StateEventRegistrarServiceInitOptions;
|
||||
SingleUserStateProviderInitOptions;
|
||||
|
||||
export async function activeUserStateProviderFactory(
|
||||
cache: { activeUserStateProvider?: ActiveUserStateProvider } & CachedServices,
|
||||
|
@ -35,8 +30,7 @@ export async function activeUserStateProviderFactory(
|
|||
async () =>
|
||||
new DefaultActiveUserStateProvider(
|
||||
await accountServiceFactory(cache, opts),
|
||||
await storageServiceProviderFactory(cache, opts),
|
||||
await stateEventRegistrarServiceFactory(cache, opts),
|
||||
await singleUserStateProviderFactory(cache, opts),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
|
|
@ -294,8 +294,7 @@ export class Main {
|
|||
|
||||
this.activeUserStateProvider = new DefaultActiveUserStateProvider(
|
||||
this.accountService,
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
this.singleUserStateProvider,
|
||||
);
|
||||
|
||||
this.derivedStateProvider = new DefaultDerivedStateProvider(
|
||||
|
|
|
@ -124,13 +124,14 @@ export class Main {
|
|||
storageServiceProvider,
|
||||
);
|
||||
|
||||
const singleUserStateProvider = new DefaultSingleUserStateProvider(
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
);
|
||||
|
||||
const stateProvider = new DefaultStateProvider(
|
||||
new DefaultActiveUserStateProvider(
|
||||
accountService,
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
),
|
||||
new DefaultSingleUserStateProvider(storageServiceProvider, stateEventRegistrarService),
|
||||
new DefaultActiveUserStateProvider(accountService, singleUserStateProvider),
|
||||
singleUserStateProvider,
|
||||
globalStateProvider,
|
||||
new DefaultDerivedStateProvider(this.memoryStorageForStateProviders),
|
||||
);
|
||||
|
|
|
@ -954,7 +954,7 @@ const typesafeProviders: Array<SafeProvider> = [
|
|||
safeProvider({
|
||||
provide: ActiveUserStateProvider,
|
||||
useClass: DefaultActiveUserStateProvider,
|
||||
deps: [AccountServiceAbstraction, StorageServiceProvider, StateEventRegistrarService],
|
||||
deps: [AccountServiceAbstraction, SingleUserStateProvider],
|
||||
}),
|
||||
safeProvider({
|
||||
provide: SingleUserStateProvider,
|
||||
|
|
|
@ -45,13 +45,13 @@ describe("EnvironmentService", () => {
|
|||
storageServiceProvider = new StorageServiceProvider(diskStorageService, memoryStorageService);
|
||||
|
||||
accountService = mockAccountServiceWith(undefined);
|
||||
const singleUserStateProvider = new DefaultSingleUserStateProvider(
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
);
|
||||
stateProvider = new DefaultStateProvider(
|
||||
new DefaultActiveUserStateProvider(
|
||||
accountService,
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
),
|
||||
new DefaultSingleUserStateProvider(storageServiceProvider, stateEventRegistrarService),
|
||||
new DefaultActiveUserStateProvider(accountService, singleUserStateProvider),
|
||||
singleUserStateProvider,
|
||||
new DefaultGlobalStateProvider(storageServiceProvider),
|
||||
new DefaultDerivedStateProvider(memoryStorageService),
|
||||
);
|
||||
|
|
|
@ -3,14 +3,12 @@ import { mock } from "jest-mock-extended";
|
|||
import { mockAccountServiceWith, trackEmissions } from "../../../../spec";
|
||||
import { AuthenticationStatus } from "../../../auth/enums/authentication-status";
|
||||
import { UserId } from "../../../types/guid";
|
||||
import { StorageServiceProvider } from "../../services/storage-service.provider";
|
||||
import { StateEventRegistrarService } from "../state-event-registrar.service";
|
||||
import { SingleUserStateProvider } from "../user-state.provider";
|
||||
|
||||
import { DefaultActiveUserStateProvider } from "./default-active-user-state.provider";
|
||||
|
||||
describe("DefaultActiveUserStateProvider", () => {
|
||||
const storageServiceProvider = mock<StorageServiceProvider>();
|
||||
const stateEventRegistrarService = mock<StateEventRegistrarService>();
|
||||
const singleUserStateProvider = mock<SingleUserStateProvider>();
|
||||
const userId = "userId" as UserId;
|
||||
const accountInfo = {
|
||||
id: userId,
|
||||
|
@ -22,11 +20,7 @@ describe("DefaultActiveUserStateProvider", () => {
|
|||
let sut: DefaultActiveUserStateProvider;
|
||||
|
||||
beforeEach(() => {
|
||||
sut = new DefaultActiveUserStateProvider(
|
||||
accountService,
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
);
|
||||
sut = new DefaultActiveUserStateProvider(accountService, singleUserStateProvider);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
|
|
|
@ -1,56 +1,40 @@
|
|||
import { Observable, map } from "rxjs";
|
||||
import { Observable, distinctUntilChanged, map } from "rxjs";
|
||||
|
||||
import { AccountService } from "../../../auth/abstractions/account.service";
|
||||
import { UserId } from "../../../types/guid";
|
||||
import { StorageServiceProvider } from "../../services/storage-service.provider";
|
||||
import { KeyDefinition } from "../key-definition";
|
||||
import { StateEventRegistrarService } from "../state-event-registrar.service";
|
||||
import { UserKeyDefinition, isUserKeyDefinition } from "../user-key-definition";
|
||||
import { ActiveUserState } from "../user-state";
|
||||
import { ActiveUserStateProvider } from "../user-state.provider";
|
||||
import { ActiveUserStateProvider, SingleUserStateProvider } from "../user-state.provider";
|
||||
|
||||
import { DefaultActiveUserState } from "./default-active-user-state";
|
||||
|
||||
export class DefaultActiveUserStateProvider implements ActiveUserStateProvider {
|
||||
private cache: Record<string, ActiveUserState<unknown>> = {};
|
||||
|
||||
activeUserId$: Observable<UserId | undefined>;
|
||||
|
||||
constructor(
|
||||
private readonly accountService: AccountService,
|
||||
private readonly storageServiceProvider: StorageServiceProvider,
|
||||
private readonly stateEventRegistrarService: StateEventRegistrarService,
|
||||
private readonly singleUserStateProvider: SingleUserStateProvider,
|
||||
) {
|
||||
this.activeUserId$ = this.accountService.activeAccount$.pipe(map((account) => account?.id));
|
||||
this.activeUserId$ = this.accountService.activeAccount$.pipe(
|
||||
map((account) => account?.id),
|
||||
// To avoid going to storage when we don't need to, only get updates when there is a true change.
|
||||
distinctUntilChanged((a, b) => (a == null || b == null ? a == b : a === b)), // Treat null and undefined as equal
|
||||
);
|
||||
}
|
||||
|
||||
get<T>(keyDefinition: KeyDefinition<T> | UserKeyDefinition<T>): ActiveUserState<T> {
|
||||
if (!isUserKeyDefinition(keyDefinition)) {
|
||||
keyDefinition = UserKeyDefinition.fromBaseKeyDefinition(keyDefinition);
|
||||
}
|
||||
const [location, storageService] = this.storageServiceProvider.get(
|
||||
keyDefinition.stateDefinition.defaultStorageLocation,
|
||||
keyDefinition.stateDefinition.storageLocationOverrides,
|
||||
);
|
||||
const cacheKey = this.buildCacheKey(location, keyDefinition);
|
||||
const existingUserState = this.cache[cacheKey];
|
||||
if (existingUserState != null) {
|
||||
// I have to cast out of the unknown generic but this should be safe if rules
|
||||
// around domain token are made
|
||||
return existingUserState as ActiveUserState<T>;
|
||||
}
|
||||
|
||||
const newUserState = new DefaultActiveUserState<T>(
|
||||
// All other providers cache the creation of their corresponding `State` objects, this instance
|
||||
// doesn't need to do that since it calls `SingleUserStateProvider` it will go through their caching
|
||||
// layer, because of that, the creation of this instance is quite simple and not worth caching.
|
||||
return new DefaultActiveUserState(
|
||||
keyDefinition,
|
||||
this.accountService,
|
||||
storageService,
|
||||
this.stateEventRegistrarService,
|
||||
this.activeUserId$,
|
||||
this.singleUserStateProvider,
|
||||
);
|
||||
this.cache[cacheKey] = newUserState;
|
||||
return newUserState;
|
||||
}
|
||||
|
||||
private buildCacheKey(location: string, keyDefinition: UserKeyDefinition<unknown>) {
|
||||
return `${location}_${keyDefinition.fullName}`;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -3,19 +3,21 @@
|
|||
* @jest-environment ../shared/test.environment.ts
|
||||
*/
|
||||
import { any, mock } from "jest-mock-extended";
|
||||
import { BehaviorSubject, firstValueFrom, of, timeout } from "rxjs";
|
||||
import { BehaviorSubject, firstValueFrom, map, of, timeout } from "rxjs";
|
||||
import { Jsonify } from "type-fest";
|
||||
|
||||
import { awaitAsync, trackEmissions } from "../../../../spec";
|
||||
import { FakeStorageService } from "../../../../spec/fake-storage.service";
|
||||
import { AccountInfo, AccountService } from "../../../auth/abstractions/account.service";
|
||||
import { AccountInfo } from "../../../auth/abstractions/account.service";
|
||||
import { AuthenticationStatus } from "../../../auth/enums/authentication-status";
|
||||
import { UserId } from "../../../types/guid";
|
||||
import { StorageServiceProvider } from "../../services/storage-service.provider";
|
||||
import { StateDefinition } from "../state-definition";
|
||||
import { StateEventRegistrarService } from "../state-event-registrar.service";
|
||||
import { UserKeyDefinition } from "../user-key-definition";
|
||||
|
||||
import { DefaultActiveUserState } from "./default-active-user-state";
|
||||
import { DefaultSingleUserStateProvider } from "./default-single-user-state.provider";
|
||||
|
||||
class TestState {
|
||||
date: Date;
|
||||
|
@ -41,23 +43,35 @@ const testKeyDefinition = new UserKeyDefinition<TestState>(testStateDefinition,
|
|||
});
|
||||
|
||||
describe("DefaultActiveUserState", () => {
|
||||
const accountService = mock<AccountService>();
|
||||
let diskStorageService: FakeStorageService;
|
||||
const storageServiceProvider = mock<StorageServiceProvider>();
|
||||
const stateEventRegistrarService = mock<StateEventRegistrarService>();
|
||||
let activeAccountSubject: BehaviorSubject<{ id: UserId } & AccountInfo>;
|
||||
|
||||
let singleUserStateProvider: DefaultSingleUserStateProvider;
|
||||
|
||||
let userState: DefaultActiveUserState<TestState>;
|
||||
|
||||
beforeEach(() => {
|
||||
activeAccountSubject = new BehaviorSubject<{ id: UserId } & AccountInfo>(undefined);
|
||||
accountService.activeAccount$ = activeAccountSubject;
|
||||
|
||||
diskStorageService = new FakeStorageService();
|
||||
userState = new DefaultActiveUserState(
|
||||
testKeyDefinition,
|
||||
accountService,
|
||||
diskStorageService,
|
||||
storageServiceProvider.get.mockReturnValue(["disk", diskStorageService]);
|
||||
|
||||
singleUserStateProvider = new DefaultSingleUserStateProvider(
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
);
|
||||
|
||||
activeAccountSubject = new BehaviorSubject<{ id: UserId } & AccountInfo>(undefined);
|
||||
|
||||
userState = new DefaultActiveUserState(
|
||||
testKeyDefinition,
|
||||
activeAccountSubject.asObservable().pipe(map((a) => a?.id)),
|
||||
singleUserStateProvider,
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.resetAllMocks();
|
||||
});
|
||||
|
||||
const makeUserId = (id: string) => {
|
||||
|
@ -223,7 +237,16 @@ describe("DefaultActiveUserState", () => {
|
|||
await changeActiveUser("1");
|
||||
|
||||
// This should always return a value right await
|
||||
const value = await firstValueFrom(userState.state$);
|
||||
const value = await firstValueFrom(
|
||||
userState.state$.pipe(
|
||||
timeout({
|
||||
first: 20,
|
||||
with: () => {
|
||||
throw new Error("Did not emit data from newly active user.");
|
||||
},
|
||||
}),
|
||||
),
|
||||
);
|
||||
expect(value).toEqual(user1Data);
|
||||
|
||||
// Make it such that there is no active user
|
||||
|
@ -392,9 +415,7 @@ describe("DefaultActiveUserState", () => {
|
|||
await changeActiveUser(undefined);
|
||||
// Act
|
||||
|
||||
// FIXME: Verify that this floating promise is intentional. If it is, add an explanatory comment and ensure there is proper error handling.
|
||||
// eslint-disable-next-line @typescript-eslint/no-floating-promises
|
||||
expect(async () => await userState.update(() => null)).rejects.toThrow(
|
||||
await expect(async () => await userState.update(() => null)).rejects.toThrow(
|
||||
"No active user at this time.",
|
||||
);
|
||||
});
|
||||
|
@ -563,7 +584,7 @@ describe("DefaultActiveUserState", () => {
|
|||
});
|
||||
|
||||
it("does not await updates if the active user changes", async () => {
|
||||
const initialUserId = (await firstValueFrom(accountService.activeAccount$)).id;
|
||||
const initialUserId = (await firstValueFrom(activeAccountSubject)).id;
|
||||
expect(initialUserId).toBe(userId);
|
||||
trackEmissions(userState.state$);
|
||||
await awaitAsync(); // storage updates are behind a promise
|
||||
|
|
|
@ -1,118 +1,27 @@
|
|||
import {
|
||||
Observable,
|
||||
map,
|
||||
switchMap,
|
||||
firstValueFrom,
|
||||
filter,
|
||||
timeout,
|
||||
merge,
|
||||
share,
|
||||
ReplaySubject,
|
||||
timer,
|
||||
tap,
|
||||
throwError,
|
||||
distinctUntilChanged,
|
||||
withLatestFrom,
|
||||
} from "rxjs";
|
||||
import { Observable, map, switchMap, firstValueFrom, timeout, throwError, NEVER } from "rxjs";
|
||||
|
||||
import { AccountService } from "../../../auth/abstractions/account.service";
|
||||
import { UserId } from "../../../types/guid";
|
||||
import {
|
||||
AbstractStorageService,
|
||||
ObservableStorageService,
|
||||
} from "../../abstractions/storage.service";
|
||||
import { StateEventRegistrarService } from "../state-event-registrar.service";
|
||||
import { StateUpdateOptions, populateOptionsWithDefault } from "../state-update-options";
|
||||
import { StateUpdateOptions } from "../state-update-options";
|
||||
import { UserKeyDefinition } from "../user-key-definition";
|
||||
import { ActiveUserState, CombinedState, activeMarker } from "../user-state";
|
||||
|
||||
import { getStoredValue } from "./util";
|
||||
|
||||
const FAKE = Symbol("fake");
|
||||
import { SingleUserStateProvider } from "../user-state.provider";
|
||||
|
||||
export class DefaultActiveUserState<T> implements ActiveUserState<T> {
|
||||
[activeMarker]: true;
|
||||
private updatePromise: Promise<[UserId, T]> | null = null;
|
||||
|
||||
private activeUserId$: Observable<UserId | null>;
|
||||
|
||||
combinedState$: Observable<CombinedState<T>>;
|
||||
state$: Observable<T>;
|
||||
|
||||
constructor(
|
||||
protected keyDefinition: UserKeyDefinition<T>,
|
||||
private accountService: AccountService,
|
||||
private chosenStorageLocation: AbstractStorageService & ObservableStorageService,
|
||||
private stateEventRegistrarService: StateEventRegistrarService,
|
||||
private activeUserId$: Observable<UserId | null>,
|
||||
private singleUserStateProvider: SingleUserStateProvider,
|
||||
) {
|
||||
this.activeUserId$ = this.accountService.activeAccount$.pipe(
|
||||
// We only care about the UserId but we do want to know about no user as well.
|
||||
map((a) => a?.id),
|
||||
// To avoid going to storage when we don't need to, only get updates when there is a true change.
|
||||
distinctUntilChanged((a, b) => (a == null || b == null ? a == b : a === b)), // Treat null and undefined as equal
|
||||
);
|
||||
|
||||
const userChangeAndInitial$ = this.activeUserId$.pipe(
|
||||
// If the user has changed, we no longer need to lock an update call
|
||||
// since that call will be for a user that is no longer active.
|
||||
tap(() => (this.updatePromise = null)),
|
||||
switchMap(async (userId) => {
|
||||
// We've switched or started off with no active user. So,
|
||||
// emit a fake value so that we can fill our share buffer.
|
||||
if (userId == null) {
|
||||
return FAKE;
|
||||
}
|
||||
|
||||
const fullKey = this.keyDefinition.buildKey(userId);
|
||||
const data = await getStoredValue(
|
||||
fullKey,
|
||||
this.chosenStorageLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
);
|
||||
return [userId, data] as CombinedState<T>;
|
||||
}),
|
||||
);
|
||||
|
||||
const latestStorage$ = this.chosenStorageLocation.updates$.pipe(
|
||||
// Use withLatestFrom so that we do NOT emit when activeUserId changes because that
|
||||
// is taken care of above, but we do want to have the latest user id
|
||||
// when we get a storage update so we can filter the full key
|
||||
withLatestFrom(
|
||||
this.activeUserId$.pipe(
|
||||
// Null userId is already taken care of through the userChange observable above
|
||||
filter((u) => u != null),
|
||||
// Take the userId and build the fullKey that we can now create
|
||||
map((userId) => [userId, this.keyDefinition.buildKey(userId)] as const),
|
||||
),
|
||||
this.combinedState$ = this.activeUserId$.pipe(
|
||||
switchMap((userId) =>
|
||||
userId != null
|
||||
? this.singleUserStateProvider.get(userId, this.keyDefinition).combinedState$
|
||||
: NEVER,
|
||||
),
|
||||
// Filter to only storage updates that pertain to our key
|
||||
filter(([storageUpdate, [_userId, fullKey]]) => storageUpdate.key === fullKey),
|
||||
switchMap(async ([storageUpdate, [userId, fullKey]]) => {
|
||||
// We can shortcut on updateType of "remove"
|
||||
// and just emit null.
|
||||
if (storageUpdate.updateType === "remove") {
|
||||
return [userId, null] as CombinedState<T>;
|
||||
}
|
||||
|
||||
return [
|
||||
userId,
|
||||
await getStoredValue(
|
||||
fullKey,
|
||||
this.chosenStorageLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
),
|
||||
] as CombinedState<T>;
|
||||
}),
|
||||
);
|
||||
|
||||
this.combinedState$ = merge(userChangeAndInitial$, latestStorage$).pipe(
|
||||
share({
|
||||
connector: () => new ReplaySubject<CombinedState<T> | typeof FAKE>(1),
|
||||
resetOnRefCountZero: () => timer(this.keyDefinition.cleanupDelayMs),
|
||||
}),
|
||||
// Filter out FAKE AFTER the share so that we can fill the ReplaySubjects
|
||||
// buffer with something and avoid emitting when there is no active user.
|
||||
filter<CombinedState<T>>((d) => d !== (FAKE as unknown)),
|
||||
);
|
||||
|
||||
// State should just be combined state without the user id
|
||||
|
@ -123,52 +32,17 @@ export class DefaultActiveUserState<T> implements ActiveUserState<T> {
|
|||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine> = {},
|
||||
): Promise<[UserId, T]> {
|
||||
options = populateOptionsWithDefault(options);
|
||||
try {
|
||||
if (this.updatePromise != null) {
|
||||
await this.updatePromise;
|
||||
}
|
||||
this.updatePromise = this.internalUpdate(configureState, options);
|
||||
const [userId, newState] = await this.updatePromise;
|
||||
return [userId, newState];
|
||||
} finally {
|
||||
this.updatePromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async internalUpdate<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine>,
|
||||
): Promise<[UserId, T]> {
|
||||
const [userId, key, currentState] = await this.getStateForUpdate();
|
||||
const combinedDependencies =
|
||||
options.combineLatestWith != null
|
||||
? await firstValueFrom(options.combineLatestWith.pipe(timeout(options.msTimeout)))
|
||||
: null;
|
||||
|
||||
if (!options.shouldUpdate(currentState, combinedDependencies)) {
|
||||
return [userId, currentState];
|
||||
}
|
||||
|
||||
const newState = configureState(currentState, combinedDependencies);
|
||||
await this.saveToStorage(key, newState);
|
||||
if (newState != null && currentState == null) {
|
||||
// Only register this state as something clearable on the first time it saves something
|
||||
// worth deleting. This is helpful in making sure there is less of a race to adding events.
|
||||
await this.stateEventRegistrarService.registerEvents(this.keyDefinition);
|
||||
}
|
||||
return [userId, newState];
|
||||
}
|
||||
|
||||
/** For use in update methods, does not wait for update to complete before yielding state.
|
||||
* The expectation is that that await is already done
|
||||
*/
|
||||
protected async getStateForUpdate() {
|
||||
const userId = await firstValueFrom(
|
||||
this.activeUserId$.pipe(
|
||||
timeout({
|
||||
first: 1000,
|
||||
with: () => throwError(() => new Error("Timeout while retrieving active user.")),
|
||||
with: () =>
|
||||
throwError(
|
||||
() =>
|
||||
new Error(
|
||||
`Timeout while retrieving active user for key ${this.keyDefinition.fullName}.`,
|
||||
),
|
||||
),
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
@ -177,15 +51,12 @@ export class DefaultActiveUserState<T> implements ActiveUserState<T> {
|
|||
`Error storing ${this.keyDefinition.fullName} for the active user: No active user at this time.`,
|
||||
);
|
||||
}
|
||||
const fullKey = this.keyDefinition.buildKey(userId);
|
||||
|
||||
return [
|
||||
userId,
|
||||
fullKey,
|
||||
await getStoredValue(fullKey, this.chosenStorageLocation, this.keyDefinition.deserializer),
|
||||
] as const;
|
||||
}
|
||||
|
||||
protected saveToStorage(key: string, data: T): Promise<void> {
|
||||
return this.chosenStorageLocation.save(key, data);
|
||||
await this.singleUserStateProvider
|
||||
.get(userId, this.keyDefinition)
|
||||
.update(configureState, options),
|
||||
];
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,120 +1,20 @@
|
|||
import {
|
||||
Observable,
|
||||
ReplaySubject,
|
||||
defer,
|
||||
filter,
|
||||
firstValueFrom,
|
||||
merge,
|
||||
share,
|
||||
switchMap,
|
||||
timeout,
|
||||
timer,
|
||||
} from "rxjs";
|
||||
|
||||
import {
|
||||
AbstractStorageService,
|
||||
ObservableStorageService,
|
||||
} from "../../abstractions/storage.service";
|
||||
import { GlobalState } from "../global-state";
|
||||
import { KeyDefinition, globalKeyBuilder } from "../key-definition";
|
||||
import { StateUpdateOptions, populateOptionsWithDefault } from "../state-update-options";
|
||||
|
||||
import { getStoredValue } from "./util";
|
||||
|
||||
export class DefaultGlobalState<T> implements GlobalState<T> {
|
||||
private storageKey: string;
|
||||
private updatePromise: Promise<T> | null = null;
|
||||
|
||||
readonly state$: Observable<T>;
|
||||
import { StateBase } from "./state-base";
|
||||
|
||||
export class DefaultGlobalState<T>
|
||||
extends StateBase<T, KeyDefinition<T>>
|
||||
implements GlobalState<T>
|
||||
{
|
||||
constructor(
|
||||
private keyDefinition: KeyDefinition<T>,
|
||||
private chosenLocation: AbstractStorageService & ObservableStorageService,
|
||||
keyDefinition: KeyDefinition<T>,
|
||||
chosenLocation: AbstractStorageService & ObservableStorageService,
|
||||
) {
|
||||
this.storageKey = globalKeyBuilder(this.keyDefinition);
|
||||
const initialStorageGet$ = defer(() => {
|
||||
return getStoredValue(this.storageKey, this.chosenLocation, this.keyDefinition.deserializer);
|
||||
});
|
||||
|
||||
const latestStorage$ = this.chosenLocation.updates$.pipe(
|
||||
filter((s) => s.key === this.storageKey),
|
||||
switchMap(async (storageUpdate) => {
|
||||
if (storageUpdate.updateType === "remove") {
|
||||
return null;
|
||||
}
|
||||
|
||||
return await getStoredValue(
|
||||
this.storageKey,
|
||||
this.chosenLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
);
|
||||
}),
|
||||
);
|
||||
|
||||
this.state$ = merge(initialStorageGet$, latestStorage$).pipe(
|
||||
share({
|
||||
connector: () => new ReplaySubject<T>(1),
|
||||
resetOnRefCountZero: () => timer(this.keyDefinition.cleanupDelayMs),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async update<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine> = {},
|
||||
): Promise<T> {
|
||||
options = populateOptionsWithDefault(options);
|
||||
if (this.updatePromise != null) {
|
||||
await this.updatePromise;
|
||||
}
|
||||
|
||||
try {
|
||||
this.updatePromise = this.internalUpdate(configureState, options);
|
||||
const newState = await this.updatePromise;
|
||||
return newState;
|
||||
} finally {
|
||||
this.updatePromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async internalUpdate<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine>,
|
||||
): Promise<T> {
|
||||
const currentState = await this.getStateForUpdate();
|
||||
const combinedDependencies =
|
||||
options.combineLatestWith != null
|
||||
? await firstValueFrom(options.combineLatestWith.pipe(timeout(options.msTimeout)))
|
||||
: null;
|
||||
|
||||
if (!options.shouldUpdate(currentState, combinedDependencies)) {
|
||||
return currentState;
|
||||
}
|
||||
|
||||
const newState = configureState(currentState, combinedDependencies);
|
||||
await this.chosenLocation.save(this.storageKey, newState);
|
||||
return newState;
|
||||
}
|
||||
|
||||
/** For use in update methods, does not wait for update to complete before yielding state.
|
||||
* The expectation is that that await is already done
|
||||
*/
|
||||
private async getStateForUpdate() {
|
||||
return await getStoredValue(
|
||||
this.storageKey,
|
||||
this.chosenLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
);
|
||||
}
|
||||
|
||||
async getFromState(): Promise<T> {
|
||||
if (this.updatePromise != null) {
|
||||
return await this.updatePromise;
|
||||
}
|
||||
return await getStoredValue(
|
||||
this.storageKey,
|
||||
this.chosenLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
);
|
||||
super(globalKeyBuilder(keyDefinition), chosenLocation, keyDefinition);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,17 +1,4 @@
|
|||
import {
|
||||
Observable,
|
||||
ReplaySubject,
|
||||
combineLatest,
|
||||
defer,
|
||||
filter,
|
||||
firstValueFrom,
|
||||
merge,
|
||||
of,
|
||||
share,
|
||||
switchMap,
|
||||
timeout,
|
||||
timer,
|
||||
} from "rxjs";
|
||||
import { Observable, combineLatest, of } from "rxjs";
|
||||
|
||||
import { UserId } from "../../../types/guid";
|
||||
import {
|
||||
|
@ -19,105 +6,31 @@ import {
|
|||
ObservableStorageService,
|
||||
} from "../../abstractions/storage.service";
|
||||
import { StateEventRegistrarService } from "../state-event-registrar.service";
|
||||
import { StateUpdateOptions, populateOptionsWithDefault } from "../state-update-options";
|
||||
import { UserKeyDefinition } from "../user-key-definition";
|
||||
import { CombinedState, SingleUserState } from "../user-state";
|
||||
|
||||
import { getStoredValue } from "./util";
|
||||
import { StateBase } from "./state-base";
|
||||
|
||||
export class DefaultSingleUserState<T> implements SingleUserState<T> {
|
||||
private storageKey: string;
|
||||
private updatePromise: Promise<T> | null = null;
|
||||
|
||||
readonly state$: Observable<T>;
|
||||
export class DefaultSingleUserState<T>
|
||||
extends StateBase<T, UserKeyDefinition<T>>
|
||||
implements SingleUserState<T>
|
||||
{
|
||||
readonly combinedState$: Observable<CombinedState<T>>;
|
||||
|
||||
constructor(
|
||||
readonly userId: UserId,
|
||||
private keyDefinition: UserKeyDefinition<T>,
|
||||
private chosenLocation: AbstractStorageService & ObservableStorageService,
|
||||
keyDefinition: UserKeyDefinition<T>,
|
||||
chosenLocation: AbstractStorageService & ObservableStorageService,
|
||||
private stateEventRegistrarService: StateEventRegistrarService,
|
||||
) {
|
||||
this.storageKey = this.keyDefinition.buildKey(this.userId);
|
||||
const initialStorageGet$ = defer(() => {
|
||||
return getStoredValue(this.storageKey, this.chosenLocation, this.keyDefinition.deserializer);
|
||||
});
|
||||
|
||||
const latestStorage$ = chosenLocation.updates$.pipe(
|
||||
filter((s) => s.key === this.storageKey),
|
||||
switchMap(async (storageUpdate) => {
|
||||
if (storageUpdate.updateType === "remove") {
|
||||
return null;
|
||||
}
|
||||
|
||||
return await getStoredValue(
|
||||
this.storageKey,
|
||||
this.chosenLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
);
|
||||
}),
|
||||
);
|
||||
|
||||
this.state$ = merge(initialStorageGet$, latestStorage$).pipe(
|
||||
share({
|
||||
connector: () => new ReplaySubject<T>(1),
|
||||
resetOnRefCountZero: () => timer(this.keyDefinition.cleanupDelayMs),
|
||||
}),
|
||||
);
|
||||
|
||||
super(keyDefinition.buildKey(userId), chosenLocation, keyDefinition);
|
||||
this.combinedState$ = combineLatest([of(userId), this.state$]);
|
||||
}
|
||||
|
||||
async update<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine> = {},
|
||||
): Promise<T> {
|
||||
options = populateOptionsWithDefault(options);
|
||||
if (this.updatePromise != null) {
|
||||
await this.updatePromise;
|
||||
}
|
||||
|
||||
try {
|
||||
this.updatePromise = this.internalUpdate(configureState, options);
|
||||
const newState = await this.updatePromise;
|
||||
return newState;
|
||||
} finally {
|
||||
this.updatePromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async internalUpdate<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine>,
|
||||
): Promise<T> {
|
||||
const currentState = await this.getStateForUpdate();
|
||||
const combinedDependencies =
|
||||
options.combineLatestWith != null
|
||||
? await firstValueFrom(options.combineLatestWith.pipe(timeout(options.msTimeout)))
|
||||
: null;
|
||||
|
||||
if (!options.shouldUpdate(currentState, combinedDependencies)) {
|
||||
return currentState;
|
||||
}
|
||||
|
||||
const newState = configureState(currentState, combinedDependencies);
|
||||
await this.chosenLocation.save(this.storageKey, newState);
|
||||
if (newState != null && currentState == null) {
|
||||
// Only register this state as something clearable on the first time it saves something
|
||||
// worth deleting. This is helpful in making sure there is less of a race to adding events.
|
||||
protected override async doStorageSave(newState: T, oldState: T): Promise<void> {
|
||||
await super.doStorageSave(newState, oldState);
|
||||
if (newState != null && oldState == null) {
|
||||
await this.stateEventRegistrarService.registerEvents(this.keyDefinition);
|
||||
}
|
||||
return newState;
|
||||
}
|
||||
|
||||
/** For use in update methods, does not wait for update to complete before yielding state.
|
||||
* The expectation is that that await is already done
|
||||
*/
|
||||
private async getStateForUpdate() {
|
||||
return await getStoredValue(
|
||||
this.storageKey,
|
||||
this.chosenLocation,
|
||||
this.keyDefinition.deserializer,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -34,11 +34,7 @@ describe("Specific State Providers", () => {
|
|||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
);
|
||||
activeSut = new DefaultActiveUserStateProvider(
|
||||
mockAccountServiceWith(null),
|
||||
storageServiceProvider,
|
||||
stateEventRegistrarService,
|
||||
);
|
||||
activeSut = new DefaultActiveUserStateProvider(mockAccountServiceWith(null), singleSut);
|
||||
globalSut = new DefaultGlobalStateProvider(storageServiceProvider);
|
||||
});
|
||||
|
||||
|
@ -67,21 +63,25 @@ describe("Specific State Providers", () => {
|
|||
},
|
||||
);
|
||||
|
||||
describe.each([
|
||||
const globalAndSingle = [
|
||||
{
|
||||
getMethod: (keyDefinition: KeyDefinition<boolean>) => globalSut.get(keyDefinition),
|
||||
expectedInstance: DefaultGlobalState,
|
||||
},
|
||||
{
|
||||
// Use a static user id so that it has the same signature as the rest and then write special tests
|
||||
// handling differing user id
|
||||
getMethod: (keyDefinition: KeyDefinition<boolean>) => singleSut.get(fakeUser1, keyDefinition),
|
||||
expectedInstance: DefaultSingleUserState,
|
||||
},
|
||||
];
|
||||
|
||||
describe.each([
|
||||
{
|
||||
getMethod: (keyDefinition: KeyDefinition<boolean>) => activeSut.get(keyDefinition),
|
||||
expectedInstance: DefaultActiveUserState,
|
||||
},
|
||||
{
|
||||
getMethod: (keyDefinition: KeyDefinition<boolean>) => globalSut.get(keyDefinition),
|
||||
expectedInstance: DefaultGlobalState,
|
||||
},
|
||||
...globalAndSingle,
|
||||
])("common behavior %s", ({ getMethod, expectedInstance }) => {
|
||||
it("returns expected instance", () => {
|
||||
const state = getMethod(fakeDiskKeyDefinition);
|
||||
|
@ -90,12 +90,6 @@ describe("Specific State Providers", () => {
|
|||
expect(state).toBeInstanceOf(expectedInstance);
|
||||
});
|
||||
|
||||
it("returns cached instance on repeated request", () => {
|
||||
const stateFirst = getMethod(fakeDiskKeyDefinition);
|
||||
const stateCached = getMethod(fakeDiskKeyDefinition);
|
||||
expect(stateFirst).toStrictEqual(stateCached);
|
||||
});
|
||||
|
||||
it("returns different instances when the storage location differs", () => {
|
||||
const stateDisk = getMethod(fakeDiskKeyDefinition);
|
||||
const stateMemory = getMethod(fakeMemoryKeyDefinition);
|
||||
|
@ -115,6 +109,14 @@ describe("Specific State Providers", () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe.each(globalAndSingle)("Global And Single Behavior", ({ getMethod }) => {
|
||||
it("returns cached instance on repeated request", () => {
|
||||
const stateFirst = getMethod(fakeDiskKeyDefinition);
|
||||
const stateCached = getMethod(fakeDiskKeyDefinition);
|
||||
expect(stateFirst).toStrictEqual(stateCached);
|
||||
});
|
||||
});
|
||||
|
||||
describe("DefaultSingleUserStateProvider only behavior", () => {
|
||||
const fakeUser2 = "00000000-0000-1000-a000-000000000002" as UserId;
|
||||
|
||||
|
|
|
@ -0,0 +1,109 @@
|
|||
import {
|
||||
Observable,
|
||||
ReplaySubject,
|
||||
defer,
|
||||
filter,
|
||||
firstValueFrom,
|
||||
merge,
|
||||
share,
|
||||
switchMap,
|
||||
timeout,
|
||||
timer,
|
||||
} from "rxjs";
|
||||
import { Jsonify } from "type-fest";
|
||||
|
||||
import { StorageKey } from "../../../types/state";
|
||||
import {
|
||||
AbstractStorageService,
|
||||
ObservableStorageService,
|
||||
} from "../../abstractions/storage.service";
|
||||
import { StateUpdateOptions, populateOptionsWithDefault } from "../state-update-options";
|
||||
|
||||
import { getStoredValue } from "./util";
|
||||
|
||||
// The parts of a KeyDefinition this class cares about to make it work
|
||||
type KeyDefinitionRequirements<T> = {
|
||||
deserializer: (jsonState: Jsonify<T>) => T;
|
||||
cleanupDelayMs: number;
|
||||
};
|
||||
|
||||
export abstract class StateBase<T, KeyDef extends KeyDefinitionRequirements<T>> {
|
||||
private updatePromise: Promise<T>;
|
||||
|
||||
readonly state$: Observable<T>;
|
||||
|
||||
constructor(
|
||||
protected readonly key: StorageKey,
|
||||
protected readonly storageService: AbstractStorageService & ObservableStorageService,
|
||||
protected readonly keyDefinition: KeyDef,
|
||||
) {
|
||||
const storageUpdate$ = storageService.updates$.pipe(
|
||||
filter((storageUpdate) => storageUpdate.key === key),
|
||||
switchMap(async (storageUpdate) => {
|
||||
if (storageUpdate.updateType === "remove") {
|
||||
return null;
|
||||
}
|
||||
|
||||
return await getStoredValue(key, storageService, keyDefinition.deserializer);
|
||||
}),
|
||||
);
|
||||
|
||||
this.state$ = merge(
|
||||
defer(() => getStoredValue(key, storageService, keyDefinition.deserializer)),
|
||||
storageUpdate$,
|
||||
).pipe(
|
||||
share({
|
||||
connector: () => new ReplaySubject(1),
|
||||
resetOnRefCountZero: () => timer(keyDefinition.cleanupDelayMs),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async update<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine> = {},
|
||||
): Promise<T> {
|
||||
options = populateOptionsWithDefault(options);
|
||||
if (this.updatePromise != null) {
|
||||
await this.updatePromise;
|
||||
}
|
||||
|
||||
try {
|
||||
this.updatePromise = this.internalUpdate(configureState, options);
|
||||
const newState = await this.updatePromise;
|
||||
return newState;
|
||||
} finally {
|
||||
this.updatePromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async internalUpdate<TCombine>(
|
||||
configureState: (state: T, dependency: TCombine) => T,
|
||||
options: StateUpdateOptions<T, TCombine>,
|
||||
): Promise<T> {
|
||||
const currentState = await this.getStateForUpdate();
|
||||
const combinedDependencies =
|
||||
options.combineLatestWith != null
|
||||
? await firstValueFrom(options.combineLatestWith.pipe(timeout(options.msTimeout)))
|
||||
: null;
|
||||
|
||||
if (!options.shouldUpdate(currentState, combinedDependencies)) {
|
||||
return currentState;
|
||||
}
|
||||
|
||||
const newState = configureState(currentState, combinedDependencies);
|
||||
await this.doStorageSave(newState, currentState);
|
||||
return newState;
|
||||
}
|
||||
|
||||
protected async doStorageSave(newState: T, oldState: T) {
|
||||
await this.storageService.save(this.key, newState);
|
||||
}
|
||||
|
||||
/** For use in update methods, does not wait for update to complete before yielding state.
|
||||
* The expectation is that that await is already done
|
||||
*/
|
||||
private async getStateForUpdate() {
|
||||
return await getStoredValue(this.key, this.storageService, this.keyDefinition.deserializer);
|
||||
}
|
||||
}
|
Loading…
Reference in New Issue