push.rocks/smartstate
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(smartstate): Add middleware, computed, batching, selector memoization, AbortSignal support, and Web Component Context Protocol provider

Browse Source
This commit is contained in:
Jürgen Kunz
2026-02-27 11:40:07 +00:00
parent 39aa63bdb3
commit 575477df09
13 changed files with 1091 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
ts/smartstate.classes.statepart.ts
View File

@@ -1,21 +1,54 @@
import * as plugins from './smartstate.plugins.js';
import { Observable, shareReplay, takeUntil } from 'rxjs';
import { StateAction, type IActionDef } from './smartstate.classes.stateaction.js';
import type { Smartstate } from './smartstate.classes.smartstate.js';
export type TMiddleware<TPayload> = (
newState: TPayload,
oldState: TPayload | undefined,
) => TPayload | Promise<TPayload>;
/**
* creates an Observable that emits once when the given AbortSignal fires
*/
function fromAbortSignal(signal: AbortSignal): Observable<void> {
return new Observable<void>((subscriber) => {
if (signal.aborted) {
subscriber.next();
subscriber.complete();
return;
}
const handler = () => {
subscriber.next();
subscriber.complete();
};
signal.addEventListener('abort', handler);
return () => signal.removeEventListener('abort', handler);
});
}
export class StatePart<TStatePartName, TStatePayload> {
public name: TStatePartName;
public state = new plugins.smartrx.rxjs.Subject<TStatePayload>();
public stateStore: TStatePayload | undefined;
public smartstateRef?: Smartstate<any>;
private cumulativeDeferred = plugins.smartpromise.cumulativeDefer();
private pendingCumulativeNotification: ReturnType<typeof setTimeout> | null = null;
private pendingBatchNotification = false;
private webStoreOptions: plugins.webstore.IWebStoreOptions;
private webStore: plugins.webstore.WebStore<TStatePayload> | null = null; // Add WebStore instance
private webStore: plugins.webstore.WebStore<TStatePayload> | null = null;
private middlewares: TMiddleware<TStatePayload>[] = [];
// Selector memoization
private selectorCache = new WeakMap<Function, plugins.smartrx.rxjs.Observable<any>>();
private defaultSelectObservable: plugins.smartrx.rxjs.Observable<TStatePayload> | null = null;
constructor(nameArg: TStatePartName, webStoreOptionsArg?: plugins.webstore.IWebStoreOptions) {
this.name = nameArg;
// Initialize WebStore if webStoreOptions are provided
if (webStoreOptionsArg) {
this.webStoreOptions = webStoreOptionsArg;
}
@@ -43,23 +76,43 @@ export class StatePart<TStatePartName, TStatePayload> {
return this.stateStore;
}
/**
* adds a middleware that intercepts setState calls.
* middleware can transform the state or throw to reject it.
* returns a removal function.
*/
public addMiddleware(middleware: TMiddleware<TStatePayload>): () => void {
this.middlewares.push(middleware);
return () => {
const idx = this.middlewares.indexOf(middleware);
if (idx !== -1) {
this.middlewares.splice(idx, 1);
}
};
}
/**
* sets the stateStore to the new state
* @param newStateArg
*/
public async setState(newStateArg: TStatePayload) {
// Run middleware chain
let processedState = newStateArg;
for (const mw of this.middlewares) {
processedState = await mw(processedState, this.stateStore);
}
// Validate state structure
if (!this.validateState(newStateArg)) {
if (!this.validateState(processedState)) {
throw new Error(`Invalid state structure for state part '${this.name}'`);
}
// Save to WebStore first to ensure atomicity - if save fails, memory state remains unchanged
// Save to WebStore first to ensure atomicity
if (this.webStore) {
await this.webStore.set(String(this.name), newStateArg);
await this.webStore.set(String(this.name), processedState);
}
// Update in-memory state after successful persistence
this.stateStore = newStateArg;
this.stateStore = processedState;
await this.notifyChange();
return this.stateStore;
@@ -67,11 +120,8 @@ export class StatePart<TStatePartName, TStatePayload> {
/**
* Validates state structure - can be overridden for custom validation
* @param stateArg
*/
protected validateState(stateArg: any): stateArg is TStatePayload {
// Basic validation - ensure state is not null/undefined
// Subclasses can override for more specific validation
return stateArg !== null && stateArg !== undefined;
}
@@ -82,6 +132,14 @@ export class StatePart<TStatePartName, TStatePayload> {
if (!this.stateStore) {
return;
}
// If inside a batch, defer the notification
if (this.smartstateRef?.isBatching) {
this.pendingBatchNotification = true;
this.smartstateRef.registerPendingNotification(this);
return;
}
const createStateHash = async (stateArg: any) => {
return await plugins.smarthashWeb.sha256FromString(plugins.smartjson.stableOneWayStringify(stateArg));
};
@@ -99,10 +157,9 @@ export class StatePart<TStatePartName, TStatePayload> {
private lastStateNotificationPayloadHash: any;
/**
* creates a cumulative notification by adding a change notification at the end of the call stack;
* creates a cumulative notification by adding a change notification at the end of the call stack
*/
public notifyChangeCumulative() {
// Debounce: clear any pending notification
if (this.pendingCumulativeNotification) {
clearTimeout(this.pendingCumulativeNotification);
}
@@ -116,27 +173,56 @@ export class StatePart<TStatePartName, TStatePayload> {
}
/**
* selects a state or a substate
* selects a state or a substate.
* supports an optional AbortSignal for automatic unsubscription.
* memoizes observables by selector function reference.
*/
public select<T = TStatePayload>(
selectorFn?: (state: TStatePayload) => T
selectorFn?: (state: TStatePayload) => T,
options?: { signal?: AbortSignal }
): plugins.smartrx.rxjs.Observable<T> {
if (!selectorFn) {
selectorFn = (state: TStatePayload) => <T>(<any>state);
const hasSignal = options?.signal != null;
// Check memoization cache (only for non-signal selects)
if (!hasSignal) {
if (!selectorFn) {
if (this.defaultSelectObservable) {
return this.defaultSelectObservable as unknown as plugins.smartrx.rxjs.Observable<T>;
}
} else if (this.selectorCache.has(selectorFn)) {
return this.selectorCache.get(selectorFn)!;
}
}
const mapped = this.state.pipe(
const effectiveSelectorFn = selectorFn || ((state: TStatePayload) => <T>(<any>state));
let mapped = this.state.pipe(
plugins.smartrx.rxjs.ops.startWith(this.getState()),
plugins.smartrx.rxjs.ops.filter((stateArg): stateArg is TStatePayload => stateArg !== undefined),
plugins.smartrx.rxjs.ops.map((stateArg) => {
try {
return selectorFn(stateArg);
return effectiveSelectorFn(stateArg);
} catch (e) {
console.error(`Selector error in state part '${this.name}':`, e);
return undefined;
}
})
);
return mapped;
if (hasSignal) {
mapped = mapped.pipe(takeUntil(fromAbortSignal(options.signal)));
return mapped;
}
// Apply shareReplay for caching and store in memo cache
const shared = mapped.pipe(shareReplay({ bufferSize: 1, refCount: true }));
if (!selectorFn) {
this.defaultSelectObservable = shared as unknown as plugins.smartrx.rxjs.Observable<TStatePayload>;
} else {
this.selectorCache.set(selectorFn, shared);
}
return shared;
}
/**
@@ -159,18 +245,32 @@ export class StatePart<TStatePartName, TStatePayload> {
}
/**
* waits until a certain part of the state becomes available
* @param selectorFn
* @param timeoutMs - optional timeout in milliseconds to prevent indefinite waiting
* waits until a certain part of the state becomes available.
* supports optional timeout and AbortSignal.
*/
public async waitUntilPresent<T = TStatePayload>(
selectorFn?: (state: TStatePayload) => T,
timeoutMs?: number
optionsOrTimeout?: number | { timeoutMs?: number; signal?: AbortSignal }
): Promise<T> {
// Parse backward-compatible args
let timeoutMs: number | undefined;
let signal: AbortSignal | undefined;
if (typeof optionsOrTimeout === 'number') {
timeoutMs = optionsOrTimeout;
} else if (optionsOrTimeout) {
timeoutMs = optionsOrTimeout.timeoutMs;
signal = optionsOrTimeout.signal;
}
const done = plugins.smartpromise.defer<T>();
const selectedObservable = this.select(selectorFn);
let resolved = false;
// Check if already aborted
if (signal?.aborted) {
throw new Error('Aborted');
}
const subscription = selectedObservable.subscribe((value) => {
if (value && !resolved) {
resolved = true;
@@ -189,12 +289,29 @@ export class StatePart<TStatePartName, TStatePayload> {
}, timeoutMs);
}
// Handle abort signal
const abortHandler = signal ? () => {
if (!resolved) {
resolved = true;
subscription.unsubscribe();
if (timeoutId) clearTimeout(timeoutId);
done.reject(new Error('Aborted'));
}
} : undefined;
if (signal && abortHandler) {
signal.addEventListener('abort', abortHandler);
}
try {
const result = await done.promise;
return result;
} finally {
subscription.unsubscribe();
if (timeoutId) clearTimeout(timeoutId);
if (signal && abortHandler) {
signal.removeEventListener('abort', abortHandler);
}
}
}