From bf98b94438024917e286628c2d8143110f6d9130 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wies=C5=82aw=20=C5=A0olt=C3=A9s?= Date: Sat, 12 Sep 2026 22:06:06 +0200 Subject: [PATCH 1/2] Support ReactiveWeb models, collection deltas and .NET resource lifetimes --- CHANGELOG.md | 13 ++ demo/index.html | 2 +- docs/COMPATIBILITY.md | 18 +++ docs/release-notes.md | 23 ++-- package-lock.json | 4 +- package.json | 4 +- scripts/package-test.mjs | 13 +- src/helpers.js | 10 +- src/lifecycle.js | 134 ++++++++++++++++---- src/operators.js | 16 ++- test/reactive-interop-types.ts | 28 +++++ test/reactive-interop.test.js | 215 +++++++++++++++++++++++++++++++++ types/index.d.ts | 46 ++++--- types/lifecycle.d.ts | 2 + types/operators.d.ts | 2 + 15 files changed, 465 insertions(+), 65 deletions(-) create mode 100644 test/reactive-interop-types.ts create mode 100644 test/reactive-interop.test.js diff --git a/CHANGELOG.md b/CHANGELOG.md index 44d36ea..b88c54e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.1.1 + +- Observe native ReactiveWeb `Changed`/`PropertyChanged` streams without proxies, + including nested path replacement, unspecified-property notifications and + shared observer cleanup. +- Support `.Dispose()` and `.DisposeAsync()` resources in lifecycle operators. +- Bind original change batches through `ApplyChanges`/`applyChanges` targets and + support collections with PascalCase editing methods. +- Release item subscriptions after synchronous source errors and continue cleanup + when individual disposers throw. +- Share `ItemWithIndex` identity between helper and kernel/root entry points. +- Add native reactive-model, lifecycle, binding and strict type regression tests. + ## 0.1.0 First public npm release of DynamicDataWeb as `@wieslawsoltes/dynamicdataweb`. diff --git a/demo/index.html b/demo/index.html index a29f821..ff68df7 100644 --- a/demo/index.html +++ b/demo/index.html @@ -2,7 +2,7 @@ DynamicData RxJS · Collection Lab -
D↗DynamicData RxJS
COLLECTION LAB v0.1.0
↓ Download package
+
D↗DynamicData RxJS
COLLECTION LAB v0.1.0
↓ Download package
REACTIVE COLLECTIONS / SOURCECACHE

Data in motion.

One source. Composable operators. A view that stays in sync.

diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index 0b02a77..abc5796 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -59,3 +59,21 @@ The implementation chunks large range insertions to avoid JavaScript argument-co The tests in `test/core.test.js`, `test/operators.test.js`, `test/advanced.test.js`, `test/lifecycle.test.js`, and `test/extras.test.js` cover source transactions, duplicate list occurrences, change replay, filter/projection transitions, sorted windows, grouping/join ownership, timers, cancellation/cleanup, and utility behavior. They are targeted JavaScript regressions; the full upstream C# fixture suite has not been mechanically translated or differentially executed against this package. The remaining compatibility work includes overload-by-overload differential tests, identical change-batch compression, full expression-tree property dependency tracking, the full .NET collection binding/context option contract, numeric-type fidelity where needed, and helper/type members marked missing in the generated mapping. Platform thread/parallel APIs require a separately designed web worker contract. Export aliases and interface declarations should not be used to close those items without exercising their behavior. + +## Native ReactiveWeb interoperability (0.1.1) + +The property operators recognize structural RxJS `Changed`, `PropertyChanged` +and lowercase equivalents in addition to the port's observable-object proxies. +ReactiveWeb `PropertyName`/`Value`/`OldValue` events and unspecified-property +notifications refresh the matching subscriptions. Nested paths rewire when an +intermediate object is replaced, and aliases do not duplicate subscriptions. +Lifecycle resources may expose `.Dispose()` or `.DisposeAsync()` as well as +JavaScript/RxJS disposal methods. + +`Bind` forwards the original batch to targets exposing `ApplyChanges` or +`applyChanges`, which lets ReactiveWeb retain indexed changes and object +lifetimes. PascalCase `Edit`/`Clear`/`AddRange` collection targets are also +recognized. These structural hooks do not add a dependency on ReactiveWeb. + +`test/reactive-interop.test.js` and `test/reactive-interop-types.ts` verify these +contracts, cleanup after synchronous errors and shared public helper identity. diff --git a/docs/release-notes.md b/docs/release-notes.md index 08819f6..8fb89b1 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -1,13 +1,22 @@ -# DynamicDataWeb 0.1.0 +DynamicDataWeb 0.1.1 adds native ReactiveWeb integration and lifecycle fixes. -First public npm release: `@wieslawsoltes/dynamicdataweb@0.1.0`. +- ReactiveObject property streams now drive AutoRefresh, property observation, + property filters and grouping directly, including nested object replacement. +- Collection binding accepts original change batches through ApplyChanges and + supports .NET-style collection editing. +- Lifecycle operators dispose .Dispose/.DisposeAsync resources and correctly + release subscriptions after synchronous errors and throwing cleanup callbacks. +- Helper/root ItemWithIndex exports share their class identity. + +Install: ```sh -npm install @wieslawsoltes/dynamicdataweb@0.1.0 rxjs +npm install @wieslawsoltes/dynamicdataweb@0.1.1 rxjs ``` -Includes keyed caches, ordered lists, incremental operators, per-instance fluent .NET-style naming, TypeScript declarations, and independently usable browser bundles. CommonJS subpaths share their module graph and retain the application's external RxJS peer. - -The release pipeline verifies source and installed consumers on Node 22 and 24, publishes immutable archives with SHA-256 checksums, deploys the showcase, and publishes the same npm tarball with provenance. Public registry checks verify downloaded bytes, installed consumers and fresh package-name installation. +The same verified tarball is distributed through GitHub Packages and npm with +provenance. The release includes standalone browser modules, a static showcase, +SHA-256 checksums and complete source archives. -See [the live collection laboratory](https://wieslawsoltes.github.io/DynamicDataWeb/) and the source compatibility report for supported contracts and remaining .NET overload adaptations. +The compatibility report documents the web adaptations; this release does not +claim exact equivalence to every DynamicData C# overload or native collection API. diff --git a/package-lock.json b/package-lock.json index 63dbd85..c2e504f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@wieslawsoltes/dynamicdataweb", - "version": "0.1.0", + "version": "0.1.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@wieslawsoltes/dynamicdataweb", - "version": "0.1.0", + "version": "0.1.1", "license": "MIT", "devDependencies": { "esbuild": "^0.28.2", diff --git a/package.json b/package.json index d8cb295..e9b65ed 100644 --- a/package.json +++ b/package.json @@ -1,13 +1,13 @@ { "name": "@wieslawsoltes/dynamicdataweb", - "version": "0.1.0", + "version": "0.1.1", "description": "Reactive keyed caches and ordered lists for JavaScript and TypeScript, powered by RxJS.", "main": "./dist/cjs/index.js", "scripts": { "test": "node --test test/*.test.js", "build": "node scripts/build.mjs && node scripts/prepare-demo.mjs", "benchmark": "node scripts/benchmark.mjs", - "check:types": "tsc --noEmit --strict --skipLibCheck --module nodenext --moduleResolution nodenext test/types.ts", + "check:types": "tsc --noEmit --strict --skipLibCheck --module nodenext --moduleResolution nodenext test/types.ts test/reactive-interop-types.ts", "demo": "node scripts/serve.mjs", "audit:api": "node docs/generate-api-mapping.mjs", "prepack": "npm run build", diff --git a/scripts/package-test.mjs b/scripts/package-test.mjs index 96481eb..70da3f8 100644 --- a/scripts/package-test.mjs +++ b/scripts/package-test.mjs @@ -41,7 +41,7 @@ assert.deepEqual(Reflect.ownKeys(Observable.prototype), before, 'Import must not for (const subpath of ${JSON.stringify(subpaths.slice(1))}) { const module = LOAD(${name} + subpath.slice(1)); assert(Object.keys(module).length > 0, 'Subpath exports must be usable'); - for (const [key, value] of Object.entries(module)) if (key in root && !(subpath === './helpers' && key === 'ItemWithIndex')) assert.equal(value, root[key], 'Shared module identity: ' + subpath + '/' + key); + for (const [key, value] of Object.entries(module)) if (key in root) assert.equal(value, root[key], 'Shared module identity: ' + subpath + '/' + key); } const source = new root.SourceCache(item => item.id); assert(source.connect() instanceof Observable, 'RxJS must remain an external peer'); @@ -60,6 +60,17 @@ const listSubscription = list.connect().pipe(root.toCollection()).subscribe(item list.addRange(['a', 'b', 'a']); list.move(2, 0); assert.deepEqual(lists.at(-1), ['a', 'a', 'b']); listSubscription.unsubscribe(); list.dispose(); +const { Subject } = LOAD('rxjs'); +const model = { id: 1, Score: 1, Changed: new Subject() }; +const native = new root.SourceCache(item => item.id); const nativeValues = []; +const nativeSubscription = native.Connect().pipe(root.AutoRefresh('Score'), root.Filter(item => item.Score >= 2), root.Bind(nativeValues)).subscribe(); +native.AddOrUpdate(model); assert.equal(nativeValues.length, 0); +model.Score = 2; model.Changed.next({ Sender: model, PropertyName: 'Score', Value: 2, OldValue: 1 }); +assert.equal(nativeValues[0], model, 'Native property streams must update cache pipelines'); +let disposed = 0; +const lifetime = native.Connect().pipe(root.SubscribeMany(() => ({ Dispose() { disposed++; } }))).subscribe(); +native.RemoveKey(1); assert.equal(disposed, 1); assert.equal(nativeValues.length, 0); +lifetime.unsubscribe(); nativeSubscription.unsubscribe(); native.Dispose(); `; await writeFile(join(consumer, 'consumer.mjs'), `import assert from 'node:assert/strict';\nimport { Observable } from 'rxjs';\n${behavior.replaceAll('LOAD(', 'await import(')}`); await writeFile(join(consumer, 'consumer.cjs'), `const assert = require('node:assert/strict');\nconst { Observable } = require('rxjs');\n${behavior.replaceAll('LOAD(', 'require(')}`); diff --git a/src/helpers.js b/src/helpers.js index c3f8f65..25af97d 100644 --- a/src/helpers.js +++ b/src/helpers.js @@ -1,6 +1,7 @@ import { Subject, Subscription, asyncScheduler, observeOn, retry, timer } from 'rxjs'; import { ChangeSet, ChangeType, ListChange, ObservableCache, Optional, SourceCache, SourceList } from './core.js'; import { ChangeSummary, collectUpdateStats } from './extras.js'; +import { ItemWithIndex } from './kernel.js'; const streamOf = source => typeof source?.connect === 'function' ? source.connect() : source; const equality = comparer => typeof comparer === 'function' ? comparer : comparer?.equals?.bind(comparer) ?? comparer?.Equals?.bind(comparer) ?? Object.is; @@ -44,14 +45,7 @@ export function getChangeType(reason) { throw new RangeError(`Unknown list change reason: ${reason}`); } -export class ItemWithIndex { - constructor(item, index) { this.item = item; this.index = index; } - get Item() { return this.item; } - get Index() { return this.index; } - equals(other) { return other instanceof ItemWithIndex && Object.is(this.item, other.item); } - Equals(other) { return this.equals(other); } - toString() { return `${this.item} (${this.index})`; } -} +export { ItemWithIndex } from './kernel.js'; export function indexOfOptional(source, item, comparer) { const equals = equality(comparer); diff --git a/src/lifecycle.js b/src/lifecycle.js index 2294ad9..477d495 100644 --- a/src/lifecycle.js +++ b/src/lifecycle.js @@ -1,26 +1,29 @@ /** Per-item observable lifetimes, property notifications and time-aware change-set operators. */ -import { Observable, Subject, ReplaySubject, Subscription, asyncScheduler, from, of, isObservable, filter as rxFilter, map, distinctUntilChanged, debounceTime, bufferTime, skip, finalize, combineLatest, fromEvent } from 'rxjs'; +import { Observable, Subject, ReplaySubject, Subscription, asyncScheduler, from, of, isObservable, filter as rxFilter, map, distinctUntilChanged, debounceTime, bufferTime, skip, finalize, combineLatest, fromEvent, share } from 'rxjs'; import { ChangeSet, SourceCache, SourceList, applyChanges, snapshotChanges } from './core.js'; import { filter, transformMany } from './operators.js'; const cs = (changes, kind = 'cache') => new ChangeSet(changes, kind); -const release = resource => { if (typeof resource === 'function') resource(); else if (resource?.unsubscribe) resource.unsubscribe(); else if (resource?.dispose) resource.dispose(); else if (resource?.[Symbol.dispose]) resource[Symbol.dispose](); }; +const release = resource => { if (typeof resource === 'function') resource(); else if (resource?.unsubscribe) resource.unsubscribe(); else if (typeof resource?.dispose === 'function') resource.dispose(); else if (typeof resource?.Dispose === 'function') resource.Dispose(); else if (resource?.[Symbol.dispose]) resource[Symbol.dispose](); }; const getChanges = changes => changes?.changes ?? changes; /** Tracks list occurrences independently, so duplicate references retain distinct lifetimes. */ function tracker(hooks = {}) { - const state = { kind: 'cache', entries: [], byKey: new Map() }; + const state = { kind: 'cache', entries: [], byKey: new Map(), disposed: false }; const add = (item, key, index = state.entries.length, previous, prior) => { + if (state.disposed) return; if (prior) prior.prior = undefined; const entry = { item, key, prior, active: true, subscription: new Subscription() }; if (state.kind === 'cache') { const old = state.byKey.get(key); if (old) { index = state.entries.indexOf(old); remove(old); } state.byKey.set(key, entry); } state.entries.splice(Math.max(0, index), 0, entry); hooks.add?.(entry, previous); return entry; }; - const remove = entry => { if (!entry) return; entry.active = false; const index = state.entries.indexOf(entry); if (index >= 0) state.entries.splice(index, 1); if (state.kind === 'cache' && state.byKey.get(entry.key) === entry) state.byKey.delete(entry.key); entry.subscription.unsubscribe(); hooks.remove?.(entry, index); }; + const remove = entry => { if (!entry) return; entry.active = false; const index = state.entries.indexOf(entry); if (index >= 0) state.entries.splice(index, 1); if (state.kind === 'cache' && state.byKey.get(entry.key) === entry) state.byKey.delete(entry.key); try { entry.subscription.unsubscribe(); } finally { hooks.remove?.(entry, index); } }; const at = change => state.kind === 'cache' ? state.byKey.get(change.key) : state.entries[change.currentIndex >= 0 ? change.currentIndex : change.previousIndex >= 0 ? change.previousIndex : state.entries.findIndex(entry => entry.item === change.current)]; state.apply = changes => { + if (state.disposed) return; state.kind = changes.kind ?? state.kind; for (const change of getChanges(changes)) { + if (state.disposed) break; const reason = change.reason; if (reason === 'add') add(change.current, change.key, change.currentIndex >= 0 ? change.currentIndex : state.entries.length); else if (reason === 'addRange') { const range = change.range ?? { items: change.current ?? [], index: change.currentIndex }; let index = range.index >= 0 ? range.index : state.entries.length; for (const item of range.items) add(item, undefined, index++); } @@ -34,20 +37,75 @@ function tracker(hooks = {}) { if (state.kind === 'cache' && Array.isArray(changes.keys)) { const ordered = changes.keys.map(key => state.byKey.get(key)).filter(Boolean); const included = new Set(ordered); state.entries = [...ordered, ...state.entries.filter(entry => !included.has(entry))]; } }; state.erase = remove; - state.dispose = () => { for (const entry of [...state.entries]) remove(entry); }; + state.dispose = () => { + state.disposed = true; + const errors = []; + for (const entry of [...state.entries]) { try { remove(entry); } catch (error) { errors.push(error); } } + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, 'Item subscriptions failed to dispose'); + }; return state; } const notifications = new WeakMap(); +const notificationStreams = new WeakMap(); const proxyTargets = new WeakMap(); const objectProxies = new WeakMap(); -function propertyEvents(item) { const target = proxyTargets.get(item) ?? item; if (target == null || (typeof target !== 'object' && typeof target !== 'function')) throw new TypeError('Property notifications require an object'); let subject = notifications.get(target); if (!subject) notifications.set(target, subject = new Subject()); return subject; } -function readProperty(item, property) { if (typeof property === 'function') return property(item); if (property == null) return item; return String(property).split('.').reduce((value, part) => value?.[part], item); } -export function notifyPropertyChanged(item, propertyName, previous) { propertyEvents(item).next({ sender: item, propertyName, value: readProperty(item, propertyName), previous }); } +const isObject = value => value != null && (typeof value === 'object' || typeof value === 'function'); +const allProperties = name => name == null || name === ''; +function propertySubject(item) { + const target = proxyTargets.get(item) ?? item; + if (!isObject(target)) throw new TypeError('Property notifications require an object'); + let subject = notifications.get(target); + if (!subject) notifications.set(target, subject = new Subject()); + return subject; +} +/** Select one alias only: ReactiveWeb exposes the same stream under several names. */ +function nativePropertyEvents(item) { + for (const name of ['PropertyChanged', 'Changed', 'propertyChanged', 'changed']) { + const stream = item[name]; + if (isObservable(stream)) return stream; + } +} +function normalizePropertyEvent(item, event) { + if (event == null || typeof event !== 'object') return { sender: item, propertyName: event, value: undefined, previous: undefined }; + return { + sender: event.sender ?? event.Sender ?? item, + propertyName: 'propertyName' in event ? event.propertyName : event.PropertyName, + value: 'value' in event ? event.value : event.Value, + previous: 'previous' in event ? event.previous : 'oldValue' in event ? event.oldValue : event.OldValue + }; +} +/** A ref-counted connection avoids duplicate native subscriptions for shared list items. */ +function propertyEvents(item) { + const target = proxyTargets.get(item) ?? item; + propertySubject(target); + let stream = notificationStreams.get(target); + if (!stream) { + stream = new Observable(observer => { + const resources = new Subscription(); + resources.add(propertySubject(target).subscribe(observer)); + try { + const native = nativePropertyEvents(target); + if (native) resources.add(native.subscribe({ + next: event => observer.next(normalizePropertyEvent(target, event)), + error: error => observer.error(error), + complete: () => observer.complete() + })); + } catch (error) { observer.error(error); } + return resources; + }).pipe(share()); + notificationStreams.set(target, stream); + } + return stream; +} +function readProperty(item, property) { if (typeof property === 'function') return property(item); if (property == null) return item; if (typeof property === 'symbol') return item?.[property]; return String(property).split('.').reduce((value, part) => value?.[part], item); } +export function notifyPropertyChanged(item, propertyName, previous) { propertySubject(item).next({ sender: item, propertyName, value: readProperty(item, propertyName), previous }); } /** Shallow property notification Proxy. Keep and mutate the returned object. */ export function createObservableObject(item) { if (proxyTargets.has(item)) return item; if (objectProxies.has(item)) return objectProxies.get(item); + if (isObject(item) && nativePropertyEvents(item)) return item; const proxy = new Proxy(item, { set(target, property, value) { const previous = target[property]; const result = Reflect.set(target, property, value); if (result && !Object.is(previous, value)) notifyPropertyChanged(proxy, property, previous); return result; }, deleteProperty(target, property) { const existed = Reflect.has(target, property); const previous = target[property]; const result = Reflect.deleteProperty(target, property); if (result && existed) notifyPropertyChanged(proxy, property, previous); return result; } @@ -59,30 +117,49 @@ function changedProperty(item, property, notifyInitial = true) { if (typeof property === 'string' && property.includes('.')) { const parts = property.split('.'); return new Observable(observer => { - let connections = new Subscription(), currentValue = readProperty(item, property); + const resources = new Subscription(), connections = new Map(); + let currentValue; + resources.add(() => { for (const entry of connections.values()) entry.subscription.unsubscribe(); connections.clear(); }); const rewire = () => { - connections.unsubscribe(); connections = new Subscription(); + const desired = new Map(); let current = item; - for (let depth = 0; depth < parts.length; depth++) { - if (current == null || typeof current !== 'object' && typeof current !== 'function') break; - const part = parts[depth], remaining = parts.slice(depth).join('.'); - connections.add(propertyEvents(current).subscribe({ + for (let depth = 0; depth < parts.length && isObject(current); depth++) { + const target = proxyTargets.get(current) ?? current; + let names = desired.get(target); + if (!names) desired.set(target, names = new Set()); + names.add(parts[depth]); names.add(parts.slice(depth).join('.')); + current = current[parts[depth]]; + } + for (const [target, entry] of connections) { + if (!desired.has(target)) { connections.delete(target); entry.subscription.unsubscribe(); } + else entry.names = desired.get(target); + } + for (const [target, names] of desired) { + if (connections.has(target) || observer.closed) continue; + const entry = { names, subscription: new Subscription() }; + connections.set(target, entry); + entry.subscription.add(propertyEvents(target).subscribe({ next(event) { - if (event.propertyName != null && event.propertyName !== part && event.propertyName !== remaining) return; + if (!allProperties(event.propertyName) && !entry.names.has(event.propertyName)) return; const previous = currentValue; - try { currentValue = readProperty(item, property); rewire(); observer.next({ sender: item, propertyName: property, value: currentValue, previous }); } catch(error) { observer.error(error); } - }, error: error => observer.error(error) + try { currentValue = readProperty(item, property); rewire(); observer.next({ sender: item, propertyName: property, value: currentValue, previous }); } catch (error) { observer.error(error); } + }, error: error => observer.error(error), complete() { if (target === (proxyTargets.get(item) ?? item)) observer.complete(); } })); - current = current[part]; } }; - try { rewire(); if (notifyInitial) observer.next({ sender: item, propertyName: property, value: currentValue, previous: undefined }); } catch(error) { observer.error(error); } - return () => connections.unsubscribe(); + try { currentValue = readProperty(item, property); rewire(); if (notifyInitial && !observer.closed) observer.next({ sender: item, propertyName: property, value: currentValue, previous: undefined }); } catch (error) { observer.error(error); } + return resources; }); } return new Observable(observer => { - const sub = propertyEvents(item).subscribe({ next(event) { if (typeof property === 'function' || property == null || event.propertyName === property) observer.next({ sender: item, propertyName: property, value: readProperty(item, property), previous: event.previous }); }, error: error => observer.error(error), complete: () => observer.complete() }); - if (notifyInitial) observer.next({ sender: item, propertyName: property, value: readProperty(item, property), previous: undefined }); + const sub = propertyEvents(item).subscribe({ + next(event) { + if (typeof property === 'function' || property == null || allProperties(event.propertyName) || event.propertyName === property) { + try { observer.next({ sender: item, propertyName: property ?? event.propertyName, value: readProperty(item, property), previous: event.previous }); } catch (error) { observer.error(error); } + } + }, error: error => observer.error(error), complete: () => observer.complete() + }); + try { if (notifyInitial && !observer.closed) observer.next({ sender: item, propertyName: property, value: readProperty(item, property), previous: undefined }); } catch (error) { observer.error(error); } return sub; }); } @@ -347,14 +424,21 @@ function disposalOperator(disposer, completedAccessor) { const check = () => { if (ended && pending === 0 && !completion.isStopped) { completion.next(undefined); completion.complete(); } }; const dispose = item => { try { const result = disposer(item); if (result?.then) { pending++; Promise.resolve(result).then(() => { pending--; check(); }, error => { pending--; completion.error(error); }); } } catch (error) { if (completedAccessor) completion.error(error); else throw error; } }; const state = tracker({ remove: entry => queue.push(entry) }); - const flush = () => { const removed = queue; queue = []; for (const entry of removed) if (!state.entries.some(current => current.prior === entry && Object.is(current.item, entry.item))) dispose(entry.item); }; + const flush = () => { + const removed = queue; queue = []; const errors = []; + for (const entry of removed) if (!state.entries.some(current => current.prior === entry && Object.is(current.item, entry.item))) { + try { dispose(entry.item); } catch (error) { errors.push(error); } + } + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, 'Items failed to dispose'); + }; resources.add(() => { state.dispose(); flush(); ended = true; check(); }); resources.add(source.subscribe({ next(changes) { try { state.apply(changes); observer.next(changes); flush(); } catch(error) { observer.error(error); } }, error:error => observer.error(error), complete:() => observer.complete() })); return resources; }); } /** Completion accessor receives a separate observable that awaits async cleanup, even on unsubscribe. */ -export function asyncDisposeMany(completedAccessor, disposer = item => item?.[Symbol.asyncDispose] ? item[Symbol.asyncDispose]() : item?.disposeAsync ? item.disposeAsync() : release(item)) { return disposalOperator(disposer, completedAccessor); } +export function asyncDisposeMany(completedAccessor, disposer = item => item?.[Symbol.asyncDispose] ? item[Symbol.asyncDispose]() : item?.disposeAsync ? item.disposeAsync() : item?.DisposeAsync ? item.DisposeAsync() : release(item)) { return disposalOperator(disposer, completedAccessor); } export const batchIf = bufferIf; export function bufferInitial(duration, scheduler = asyncScheduler) { if (!(duration >= 0)) throw new RangeError('Initial duration must be non-negative'); @@ -370,7 +454,7 @@ export const finallySafe = action => finalize(action); export const ConnectionStatus = Object.freeze({ Pending:'pending', Loaded:'loaded', Errored:'errored', Completed:'completed' }); export function monitorStatus() { return source => new Observable(observer => {let loaded=false;observer.next(ConnectionStatus.Pending);return source.subscribe({next(){if(!loaded){loaded=true;observer.next(ConnectionStatus.Loaded);}},error(){observer.next(ConnectionStatus.Errored);observer.complete();},complete(){observer.next(ConnectionStatus.Completed);observer.complete();}});}); } export const watchValue = key => source => (typeof source.connect === 'function' ? source.connect() : source).pipe(watch(key),map(change=>change.current)); -export function whenAnyPropertyChanged(itemOrChanges, ...propertyNames) { const observe=item=>propertyEvents(item).pipe(rxFilter(event=>!propertyNames.length||propertyNames.includes(event.propertyName)),map(()=>item));return isObservable(itemOrChanges)?itemOrChanges.pipe(mergeMany(observe)):observe(itemOrChanges); } +export function whenAnyPropertyChanged(itemOrChanges, ...propertyNames) { const observe=item=>propertyEvents(item).pipe(rxFilter(event=>!propertyNames.length||allProperties(event.propertyName)||propertyNames.includes(event.propertyName)),map(()=>item));return isObservable(itemOrChanges)?itemOrChanges.pipe(mergeMany(observe)):observe(itemOrChanges); } /** whenChanged(object, [property names/selectors], (object, ...values) => result). */ export function whenChanged(item, properties, resultSelector) { const fields=Array.isArray(properties)?properties:[properties]; diff --git a/src/operators.js b/src/operators.js index 021864a..05e8b5e 100644 --- a/src/operators.js +++ b/src/operators.js @@ -515,11 +515,23 @@ export function bind(target) { const model = new Model(); return source.subscribe({ next: changes => { try { + const apply = typeof target.ApplyChanges === 'function' ? target.ApplyChanges : target.applyChanges; + if (typeof apply === 'function') { + apply.call(target, changes); + observer.next(changes); + return; + } model.ingest(changes); const items = changes.items ? [...changes.items] : model.entries.map(e => e.value); if (typeof target === 'function') target(items, changes); else if (Array.isArray(target)) { target.length = 0; for (const item of items) target.push(item); } else if (typeof target.load === 'function') target.load(items); - else if (typeof target.edit === 'function') target.edit(updater => { updater.clear(); if (updater.addRange) updater.addRange(items); else updater.addOrUpdate(items); }); + else if (typeof target.edit === 'function' || typeof target.Edit === 'function') (typeof target.edit === 'function' ? target.edit : target.Edit).call(target, updater => { + const collection = updater ?? target; + const clear = collection.clear ?? collection.Clear; + const add = collection.addRange ?? collection.AddRange ?? collection.addOrUpdate ?? collection.AddOrUpdate; + if (typeof clear !== 'function' || typeof add !== 'function') throw new TypeError('Binding collection updater requires Clear/clear and AddRange/addRange or AddOrUpdate/addOrUpdate.'); + clear.call(collection); add.call(collection, items); + }); else if (typeof target.next === 'function') target.next(items); else throw new TypeError('Unsupported binding target.'); observer.next(changes); @@ -531,7 +543,7 @@ export function bind(target) { export const bindToObservableList = bind; export const bindToObservableCollection = bind; export const sortAndBind = (targetOrComparer, comparerOrTarget, options) => { - const isTarget = value => Array.isArray(value) || !!value?.load || !!value?.edit; + const isTarget = value => Array.isArray(value) || !!value?.load || !!value?.edit || !!value?.Edit || !!value?.applyChanges || !!value?.ApplyChanges; const comparerFirst = isTarget(comparerOrTarget) && !isTarget(targetOrComparer); const target = comparerFirst ? comparerOrTarget : targetOrComparer, comparer = comparerFirst ? targetOrComparer : comparerOrTarget; return source => source.pipe(sort(comparer, options), bind(target)); diff --git a/test/reactive-interop-types.ts b/test/reactive-interop-types.ts new file mode 100644 index 0000000..4c809c7 --- /dev/null +++ b/test/reactive-interop-types.ts @@ -0,0 +1,28 @@ +import { SourceList, SourceCache, type ChangeSet, type DisposableLike, type ChangeSetBindingTarget, type PascalCaseCollectionTarget, subscribeMany, bind, autoRefresh, filterOnProperty, whenPropertyChanged, observeProperty, notifyPropertyChanged, ObservableChangeSet } from '@wieslawsoltes/dynamicdataweb'; +import type { PropertyValue, ReactivePropertyNotification } from '@wieslawsoltes/dynamicdataweb/lifecycle'; +import { Subject, type Observable } from 'rxjs'; + +class Item { + readonly Changed = new Subject>(); + constructor(public Id: string, public Name: string, public Address: { City: string }) {} +} +const source = new SourceCache((item: Item) => item.Id); +const target: ChangeSetBindingTarget = { ApplyChanges(changes: ChangeSet) { changes.forEach(change => { const key: string = change.key; void key; }); } }; +source.connect().pipe(bind(target)); +source.Connect().Bind(target).AutoRefresh('Address.City'); +source.connect().pipe(autoRefresh('Address.City'), filterOnProperty('Address.City', item => item.Address.City !== '')); +const cleanup: DisposableLike = { Dispose() {} }; +source.connect().pipe(subscribeMany(() => cleanup)); +source.Connect().SubscribeMany(() => ({ Dispose() {} })); +ObservableChangeSet.createList((list: SourceList) => { list.add(new Item('a', 'Ada', { City: 'London' })); return cleanup; }); +const collection: PascalCaseCollectionTarget = { Edit(action) { action({ Clear() {}, AddRange(items: Iterable) { Array.from(items); } }); } }; +source.connect().pipe(bind(collection)); +const item = new Item('b', 'Grace', { City: 'Paris' }); +const property: Observable> = whenPropertyChanged(item, 'Address.City'); +const projected: Observable = observeProperty(item, value => value.Name); +const nested: Observable = observeProperty(item, 'Address.City'); +notifyPropertyChanged(item, ''); notifyPropertyChanged(item, null); notifyPropertyChanged(item); +void property; void projected; void nested; +// @ts-expect-error a delta target receives keyed ChangeSets, not item arrays. +const invalid: ChangeSetBindingTarget = { ApplyChanges(changes: Item[]) {} }; +void invalid; diff --git a/test/reactive-interop.test.js b/test/reactive-interop.test.js new file mode 100644 index 0000000..0733789 --- /dev/null +++ b/test/reactive-interop.test.js @@ -0,0 +1,215 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { Observable, Subject, firstValueFrom } from 'rxjs'; +import { ChangeSet, SourceCache, SourceList } from '../src/core.js'; +import { autoRefresh, whenPropertyChanged, whenAnyPropertyChanged, observeProperty, filterOnProperty, createObservableObject, notifyPropertyChanged, subscribeMany, disposeMany, asyncDisposeMany, ObservableChangeSet } from '../src/lifecycle.js'; +import { bind, sortAndBind } from '../src/operators.js'; + +// The public ReactiveWeb ReactiveObject protocol, without importing it or creating a cycle. +class ReactiveFixture { + constructor(properties = {}, alias = 'all') { + Object.assign(this, properties); + this.events = new Subject(); this.active = 0; this.connections = 0; + const stream = new Observable(observer => { + this.active++; this.connections++; + const inner = this.events.subscribe(observer); + return () => { this.active--; inner.unsubscribe(); }; + }); + if (alias === 'all') this.PropertyChanged = this.Changed = this.changed = stream; + else this[alias] = stream; + } + SetValue(PropertyName, Value) { + const OldValue = this[PropertyName]; this[PropertyName] = Value; + this.events.next({ Sender: this, PropertyName, Value, OldValue }); + } + Dispose() { this.events.complete(); } +} + +test('native ReactiveObject aliases share one connection and normalize PascalCase events', () => { + const item = new ReactiveFixture({ Name: 'Ada' }); + const first = [], second = []; + const a = whenPropertyChanged(item, 'Name').subscribe(event => first.push(event)); + const b = observeProperty(item, 'Name').subscribe(value => second.push(value)); + assert.equal(item.active, 1); assert.equal(item.connections, 1); + item.SetValue('Name', 'Grace'); + assert.deepEqual(first.map(event => [event.sender, event.propertyName, event.value, event.previous]), [[item, 'Name', 'Ada', undefined], [item, 'Name', 'Grace', 'Ada']]); + assert.deepEqual(second, ['Ada', 'Grace']); + a.unsubscribe(); assert.equal(item.active, 1); + b.unsubscribe(); assert.equal(item.active, 0); + const c = observeProperty(item, 'Name').subscribe(); assert.equal(item.active, 1); + item.Dispose(); assert.equal(c.closed, true); assert.equal(item.active, 0); +}); + +test('Changed-only, camelCase and all-property notifications integrate without proxy wrapping', () => { + for (const alias of ['Changed', 'PropertyChanged', 'changed', 'propertyChanged']) { + const item = new ReactiveFixture({ Count: 1 }, alias), values = [], any = []; + assert.equal(createObservableObject(item), item); + const a = observeProperty(item, 'Count').subscribe(value => values.push(value)); + const b = whenAnyPropertyChanged(item, 'Count').subscribe(value => any.push(value)); + item.Count = 2; item.events.next({ sender: item, propertyName: 'Count', value: 2, previous: 1 }); + for (const PropertyName of ['', null, undefined]) { item.Count++; item.events.next({ Sender: item, PropertyName }); } + item.events.next({ PropertyName: 'Unrelated' }); + assert.deepEqual(values, [1, 2, 3, 4, 5]); assert.equal(any.length, 4); + a.unsubscribe(); b.unsubscribe(); assert.equal(item.active, 0); + } +}); + +test('plain-object explicit all-property notifications and Symbol properties remain observable', () => { + const token = Symbol('state'), item = { value: 1, [token]: 0 }, values = [], symbols = []; + const a = observeProperty(item, 'value').subscribe(value => values.push(value)); + const b = observeProperty(item, token).subscribe(value => symbols.push(value)); + item.value = 2; notifyPropertyChanged(item, ''); + item[token] = 3; notifyPropertyChanged(item, token, 0); + assert.deepEqual(values, [1, 2]); assert.deepEqual(symbols, [0, 3]); + a.unsubscribe(); b.unsubscribe(); +}); + +test('native nested paths replace and detach children while preserving parent subscriptions', () => { + const first = new ReactiveFixture({ City: 'Paris' }), second = new ReactiveFixture({ City: 'Oslo' }); + const item = new ReactiveFixture({ Address: first }), seen = []; + const subscription = whenPropertyChanged(item, 'Address.City').subscribe(event => seen.push([event.value, event.previous])); + first.SetValue('City', 'London'); + assert.equal(item.connections, 1); assert.equal(first.connections, 1); + item.SetValue('Address', second); + assert.equal(first.active, 0); assert.equal(second.active, 1); assert.equal(item.connections, 1); + first.SetValue('City', 'Stale'); second.SetValue('City', 'Rome'); + item.SetValue('Address', null); assert.equal(second.active, 0); + item.SetValue('Address', first); first.City = 'Warsaw'; first.events.next({ PropertyName: '' }); + assert.deepEqual(seen, [['Paris', undefined], ['London', 'Paris'], ['Oslo', 'London'], ['Rome', 'Oslo'], [undefined, 'Rome'], ['Stale', undefined], ['Warsaw', 'Stale']]); + item.Dispose(); assert.equal(subscription.closed, true); assert.equal(first.active, 0); assert.equal(item.active, 0); +}); + +test('nested paths through the same object subscribe once and emit once per native event', () => { + const item = new ReactiveFixture({ Name: 'A' }); item.Self = item; + const values = [], subscription = observeProperty(item, 'Self.Name').subscribe(value => values.push(value)); + item.SetValue('Name', 'B'); + assert.deepEqual(values, ['A', 'B']); assert.equal(item.active, 1); assert.equal(item.connections, 1); + subscription.unsubscribe(); assert.equal(item.active, 0); +}); + +test('AutoRefresh retains duplicate occurrences, removes listeners and replaces keyed objects', () => { + const item = new ReactiveFixture({ Id: 1, Name: 'A' }), list = new SourceList(), batches = []; + const subscription = list.connect().pipe(autoRefresh('Name')).subscribe(changes => batches.push(changes)); + list.addRange([item, item]); assert.equal(item.active, 1); + item.SetValue('Name', 'B'); + assert.deepEqual(batches.flatMap(batch => batch.filter(change => change.reason === 'refresh').map(change => change.currentIndex)), [0, 1]); + list.removeAt(0); assert.equal(item.active, 1); + const before = batches.length; item.SetValue('Name', 'C'); assert.equal(batches.length, before + 1); + list.clear(); assert.equal(item.active, 0); + const removed = batches.length; item.SetValue('Name', 'D'); assert.equal(batches.length, removed); + subscription.unsubscribe(); list.dispose(); + const cache = new SourceCache(value => value.Id), replacement = new ReactiveFixture({ Id: 1, Name: 'New' }); + const cached = cache.connect().pipe(autoRefresh()).subscribe(); + cache.addOrUpdate(item); cache.addOrUpdate(replacement); + assert.equal(item.active, 0); assert.equal(replacement.active, 1); + cached.unsubscribe(); assert.equal(replacement.active, 0); cache.dispose(); +}); + +test('FilterOnProperty follows native nested properties and shared children across items', () => { + const shared = new ReactiveFixture({ Enabled: false }), other = new ReactiveFixture({ Enabled: true }); + const first = new ReactiveFixture({ Id: 1, Options: shared }), second = new ReactiveFixture({ Id: 2, Options: shared }); + const cache = new SourceCache(item => item.Id), rows = []; + const subscription = cache.connect().pipe(filterOnProperty('Options.Enabled', item => item.Options.Enabled), bind(rows)).subscribe(); + cache.addOrUpdate([first, second]); assert.deepEqual(rows, []); assert.equal(shared.active, 1); + shared.SetValue('Enabled', true); assert.deepEqual(rows, [first, second]); + first.SetValue('Options', other); shared.SetValue('Enabled', false); assert.deepEqual(rows, [first]); + cache.removeKey(2); assert.equal(shared.active, 0); assert.equal(second.active, 0); + subscription.unsubscribe(); assert.equal(first.active, 0); assert.equal(other.active, 0); cache.dispose(); +}); + +test('native errors and property getter errors reach observers and detach subscriptions', () => { + const leaf = new ReactiveFixture({ Value: 1 }), item = new ReactiveFixture({ Child: leaf }), errors = []; + const subscription = whenPropertyChanged(item, 'Child.Value').subscribe({ error: error => errors.push(error) }); + const error = new Error('native source failed'); leaf.events.error(error); + assert.deepEqual(errors, [error]); assert.equal(subscription.closed, true); assert.equal(item.active, 0); assert.equal(leaf.active, 0); + const failing = new ReactiveFixture({ Name: 'first' }), failures = []; + const simple = observeProperty(failing, 'Name').subscribe({ error: error => failures.push(error) }); + Object.defineProperty(failing, 'Name', { get() { throw error; } }); + failing.events.next({ PropertyName: 'Name' }); + assert.equal(simple.closed, true); assert.deepEqual(failures, [error]); assert.equal(failing.active, 0); +}); + +test('Bind delta contract preserves batches, ranges, refreshes and sorting metadata', () => { + for (const method of ['ApplyChanges', 'applyChanges']) { + const source = new Subject(), received = [], forwarded = []; + const target = { [method](changes) { assert.equal(this, target); received.push(changes); }, edit() { throw new Error('delta target must take priority'); } }; + const subscription = source.pipe(bind(target)).subscribe(changes => forwarded.push(changes)); + const batch = new ChangeSet([{ reason: 'addRange', range: { items: ['a', 'a'], index: 0 } }, { reason: 'refresh', current: 'a', currentIndex: 1 }], 'list'); + batch.items = ['a', 'a']; batch.keys = [1, 2]; source.next(batch); + assert.equal(received[0], batch); assert.equal(forwarded[0], batch); + assert.equal(received[0].kind, 'list'); assert.equal(received[0].items, batch.items); subscription.unsubscribe(); + } + const source = new SourceCache(item => item.Id), batches = []; + const sorted = source.connect().pipe(sortAndBind((a, b) => a.Order - b.Order, { ApplyChanges(changes) { batches.push(changes); } })).subscribe(); + source.addOrUpdate([{ Id: 'b', Order: 2 }, { Id: 'a', Order: 1 }]); + assert.deepEqual(batches.at(-1).keys, ['a', 'b']); sorted.unsubscribe(); source.dispose(); +}); + +test('Bind accepts PascalCase collection edits, preserves this and forwards target errors', () => { + const source = new SourceList(), rows = []; + const target = { Edit(action) { assert.equal(this, target); action(this); }, Clear() { assert.equal(this, target); rows.length = 0; }, AddRange(items) { assert.equal(this, target); rows.push(...items); } }; + const subscription = source.connect().pipe(bind(target)).subscribe(); + source.addRange([1, 2]); source.move(0, 1); assert.deepEqual(rows, [2, 1]); + source.removeAt(0); assert.deepEqual(rows, [1]); subscription.unsubscribe(); source.dispose(); + const events = new Subject(), expected = new Error('target failed'), errors = []; + const broken = events.pipe(bind({ ApplyChanges() { throw expected; } })).subscribe({ error: error => errors.push(error) }); + events.next(new ChangeSet([], 'list')); + assert.deepEqual(errors, [expected]); assert.equal(events.observed, false); assert.equal(broken.closed, true); +}); + +test('SubscribeMany, DisposeMany and ObservableChangeSet accept Dispose-only resources', () => { + const source = new SourceList(), disposed = [], own = []; + const first = { Dispose() { disposed.push('first'); } }, second = { Dispose() { disposed.push('second'); } }; + const subscription = source.connect().pipe(subscribeMany(item => ({ Dispose() { own.push(item); } })), disposeMany()).subscribe(); + source.addRange([first, second]); source.remove(first); + assert.deepEqual(disposed, ['first']); assert.deepEqual(own, [first]); + subscription.unsubscribe(); subscription.unsubscribe(); + assert.deepEqual(disposed, ['first', 'second']); assert.deepEqual(own, [first, second]); source.dispose(); + let cleaned = 0; + const created = ObservableChangeSet.createList(list => { list.add(1); return { Dispose() { cleaned++; } }; }).subscribe(); + created.unsubscribe(); assert.equal(cleaned, 1); +}); + +test('AsyncDisposeMany waits for PascalCase DisposeAsync completion', async () => { + const source = new SourceList(); let resolve, completed, calls = 0; + const resource = { DisposeAsync() { calls++; return new Promise(done => { resolve = done; }); } }; + const subscription = source.connect().pipe(asyncDisposeMany(completion => { completed = firstValueFrom(completion); })).subscribe(); + source.add(resource); subscription.unsubscribe(); + assert.equal(calls, 1); let finished = false; completed.then(() => { finished = true; }); + await Promise.resolve(); assert.equal(finished, false); resolve(); await completed; assert.equal(finished, true); source.dispose(); +}); + +test('helpers and kernel share ItemWithIndex identity for optional lookups', async () => { + const helpers = await import('../src/helpers.js'), kernel = await import('../src/kernel.js'); + assert.equal(helpers.ItemWithIndex, kernel.ItemWithIndex); + const item = {}, found = helpers.indexOfOptional([item], item).value; + assert.equal(found instanceof kernel.ItemWithIndex, true); + assert.equal(found.Equals(new kernel.ItemWithIndex(item, 99)), true); +}); + +test('synchronous native errors stop attaching later items in the same change batch', () => { + const expected = new Error('already failed'), failed = new ReactiveFixture(), live = new ReactiveFixture(), source = new SourceList(), errors = []; + failed.events.error(expected); + const subscription = source.connect().pipe(autoRefresh()).subscribe({ error: error => errors.push(error) }); + source.addRange([failed, live]); + assert.equal(subscription.closed, true); assert.deepEqual(errors, [expected]); + assert.equal(failed.active, 0); assert.equal(live.active, 0); assert.equal(live.connections, 0); + source.dispose(); +}); + +test('SubscribeMany releases every resource even if one Dispose method throws', () => { + const source = new SourceList(), disposed = [], expected = new Error('cleanup failed'); + const subscription = source.connect().pipe(subscribeMany(item => ({ Dispose() { disposed.push(item); if (item === 1) throw expected; } }))).subscribe(); + source.addRange([1, 2, 3]); + assert.throws(() => subscription.unsubscribe(), /cleanup failed/); + assert.deepEqual(disposed, [1, 2, 3]); + source.add(4); assert.deepEqual(disposed, [1, 2, 3]); source.dispose(); +}); + +test('DisposeMany attempts every item cleanup when a Dispose method throws', () => { + const source = new SourceList(), disposed = []; + const subscription = source.connect().pipe(disposeMany()).subscribe(); + source.addRange([1, 2, 3].map(id => ({ Dispose() { disposed.push(id); if (id !== 2) throw new Error(`cleanup ${id}`); } }))); + assert.throws(() => subscription.unsubscribe(), /Items failed to dispose/); + assert.deepEqual(disposed, [1, 2, 3]); source.dispose(); +}); diff --git a/types/index.d.ts b/types/index.d.ts index 39caf5d..6fe2922 100644 --- a/types/index.d.ts +++ b/types/index.d.ts @@ -8,6 +8,8 @@ export type Comparer = ((left: T, right: T) => number) | { compare(left: T, r export type EqualityComparer = ((left: T, right: T) => boolean) | { equals(left: T, right: T): boolean } | { Equals(left: T, right: T): boolean }; export type ChangeOperator = OperatorFunction, ChangeSet>; export type ChangeSource = Observable> | ReadonlyCache | ReadonlyList; +export type DisposableLike = TeardownLogic | { dispose(): void } | { Dispose(): void }; +export type PropertyPath = keyof T | `${Extract}.${string}`; export interface Disposable { dispose(): void; Dispose(): void; unsubscribe(): void; readonly closed?: boolean; } export interface FluentObservable extends Observable { Subscribe: Observable['subscribe']; Pipe: Observable['pipe']; @@ -19,7 +21,7 @@ export interface DataObservable extends FluentObservable(factory: AsyncSelector, options?: AsyncOptions): DataObservable; FilterOnObservable(selector: Selector, K>, options?: AsyncOptions): DataObservable; TransformOnObservable(selector: AsyncSelector, options?: AsyncOptions): DataObservable; - AutoRefresh(property?: keyof T | ((item: T) => unknown), options?: RefreshOptions): DataObservable; + AutoRefresh(property?: PropertyPath | ((item: T) => unknown), options?: RefreshOptions): DataObservable; AutoRefreshOnObservable(selector: Selector, K>, options?: RefreshOptions): DataObservable; Sort(comparer?: Comparer | Observable>, options?: SortConfiguration | Observable): DataObservable; SortBy(selector: KeySelector, direction?: SortDirection, options?: SortConfiguration): DataObservable; @@ -27,9 +29,9 @@ export interface DataObservable extends FluentObservable): DataObservable; Virtualize(requests: VirtualRequestLike | Observable): DataObservable; Top(size: number): DataObservable; Top(comparer: Comparer | Observable>, size: number): DataObservable; Reverse(): DataObservable; - SortAndBind(target: BindingTarget, comparer?: Comparer, options?: SortConfiguration): DataObservable; - SortAndBind(comparer: Comparer | Observable>, target: BindingTarget, options?: SortConfiguration): DataObservable; - Bind(target: BindingTarget): DataObservable; + SortAndBind(target: BindingTarget, comparer?: Comparer, options?: SortConfiguration): DataObservable; + SortAndBind(comparer: Comparer | Observable>, target: BindingTarget, options?: SortConfiguration): DataObservable; + Bind(target: BindingTarget): DataObservable; ToCollection(): FluentObservable; ToSortedCollection(selectorOrComparer?: KeySelector | Comparer, direction?: SortDirection): FluentObservable; AsObservableCache(keySelector?: KeySelector): ObservableCache; @@ -53,7 +55,7 @@ export interface DataObservable extends FluentObservable, fallback?: number): FluentObservable; IsEmpty(): FluentObservable; IsNotEmpty(): FluentObservable; DisposeMany(disposer?: (item: T) => void): DataObservable; - SubscribeMany(selector: Selector): DataObservable; + SubscribeMany(selector: Selector): DataObservable; MergeMany(selector: Selector, K>): FluentObservable; OnItemAdded(action: Selector): DataObservable; OnItemUpdated(action: (current: T, previous: T, key: K) => void): DataObservable; @@ -216,7 +218,11 @@ export interface PageRequestLike { page: number; size: number; } export interface VirtualRequestLike { startIndex: number; size: number; } export interface PageResponse { page: number; size: number; totalSize: number; pages: number; } export interface VirtualResponse { startIndex: number; size: number; totalSize: number; } -export type BindingTarget = T[] | SourceList | ((items: T[], changes: ChangeSet) => void) | { next(items: T[]): void }; +/** A delta target consumes the original batch, including sorting and range metadata. */ +export type ChangeSetBindingTarget = { ApplyChanges(changes: ChangeSet): unknown } | { applyChanges(changes: ChangeSet): unknown }; +export interface PascalCaseListUpdater { Clear(): unknown; AddRange(items: Iterable): unknown; } +export interface PascalCaseCollectionTarget { Edit(action: (collection: PascalCaseListUpdater) => void): unknown; } +export type BindingTarget = T[] | SourceList | SourceCache | ChangeSetBindingTarget | PascalCaseCollectionTarget | ((items: T[], changes: ChangeSet) => void) | { load(items: T[]): unknown } | { next(items: T[]): void }; export function filter(predicate: Predicate | Observable>, reapplyOrOptions?: Observable | FilterOptions | boolean, suppressEmptyChangeSets?: boolean): ChangeOperator; export function filter(state: Observable, predicate: (state: S, item: T, key: K) => boolean, suppressEmptyChangeSets?: boolean): ChangeOperator; export const filterImmutable: typeof filter; @@ -244,10 +250,10 @@ export const sortAndVirtualize: typeof sortAndVirtualise; export function sortBy(selector: KeySelector, direction?: SortDirection, options?: SortConfiguration): ChangeOperator; export function sortBy(property: keyof T, direction?: SortDirection, options?: SortConfiguration): ChangeOperator; export function toCollection(): OperatorFunction, T[]>; -export function bind(target: BindingTarget): ChangeOperator; +export function bind(target: BindingTarget): ChangeOperator; export const bindToObservableList: typeof bind; export const bindToObservableCollection: typeof bind; -export function sortAndBind(target: BindingTarget, comparer?: Comparer, options?: SortConfiguration): ChangeOperator; -export function sortAndBind(comparer: Comparer | Observable>, target: BindingTarget, options?: SortConfiguration): ChangeOperator; +export function sortAndBind(target: BindingTarget, comparer?: Comparer, options?: SortConfiguration): ChangeOperator; +export function sortAndBind(comparer: Comparer | Observable>, target: BindingTarget, options?: SortConfiguration): ChangeOperator; export function asObservableCache(source: Observable>, keySelector?: KeySelector): ObservableCache; export function asObservableCache(keySelector?: KeySelector): (source: Observable>) => ObservableCache; export function asObservableList(source: Observable>): ObservableList; @@ -261,23 +267,29 @@ export interface RefreshOptions extends TimerOptions { throttle?: number; buffer export interface BufferOptions extends TimerOptions { initialPauseState?: boolean; timeout?: number; } export interface AsyncOptions { onError?: (error: { error: unknown; item: T; key: K }) => void; waitForCompletion?: boolean; maximumConcurrency?: number; transformOnRefresh?: boolean; } export type AsyncSelector = (item: T, key: K, previous: R | undefined, signal: AbortSignal) => ObservableInput | R; -export interface PropertyValue { sender: T; propertyName: PropertyKey; value: V; previous?: V; } +export interface PropertyValue { sender: T; propertyName: PropertyKey | ((item: T) => V) | null | undefined; value: V; previous?: V; } +/** PropertyChanged/Changed observables are discovered structurally; no ReactiveWeb dependency is required. */ +export interface ReactivePropertyNotification { Sender: T; PropertyName: PropertyKey | null | undefined; Value?: V; OldValue?: V; } export function notifyPropertyChanged(item: T, propertyName: P, previous?: T[P]): void; +export function notifyPropertyChanged(item: T, propertyName?: null | ''): void; export function createObservableObject(item: T): T; export const observableObject: typeof createObservableObject; export function whenPropertyChanged(itemOrChanges: T | Observable>, property: P, notifyInitial?: boolean): Observable>; export function whenPropertyChanged(itemOrChanges: T | Observable>, property: (item: T) => V, notifyInitial?: boolean): Observable>; +export function whenPropertyChanged(itemOrChanges: T | Observable>, property: `${Extract}.${string}`, notifyInitial?: boolean): Observable>; +export function observeProperty(itemOrChanges: T | Observable>, property: (item: T) => V, notifyInitial?: boolean): Observable; +export function observeProperty(itemOrChanges: T | Observable>, property: `${Extract}.${string}`, notifyInitial?: boolean): Observable; export function observeProperty(itemOrChanges: T | Observable>, property: P, notifyInitial?: boolean): Observable; export const whenValueChanged: typeof observeProperty; export function autoRefreshOnObservable(selector: Selector, K>, options?: RefreshOptions): ChangeOperator; -export function autoRefresh(property?: keyof T | ((item: T) => unknown), options?: RefreshOptions): ChangeOperator; +export function autoRefresh(property?: PropertyPath | ((item: T) => unknown), options?: RefreshOptions): ChangeOperator; export function filterOnObservable(selector: Selector | boolean, K>, options?: AsyncOptions): ChangeOperator; export function transformOnObservable(selector: AsyncSelector, options?: AsyncOptions): ChangeOperator; export function transformAsync(factory: AsyncSelector, options?: AsyncOptions): ChangeOperator; export function transformSafeAsync(factory: AsyncSelector, onError: NonNullable['onError']>, options?: AsyncOptions): ChangeOperator; export function mergeMany(selector: Selector, K>): OperatorFunction, R>; export function mergeManyItems(selector: Selector, K>): OperatorFunction, { item: T; key: K; value: R }>; -export function subscribeMany(selector: Selector): ChangeOperator; +export function subscribeMany(selector: Selector): ChangeOperator; export function disposeMany(disposer?: (item: T) => void): ChangeOperator; export function onItemAdded(action: Selector): ChangeOperator; export function onItemUpdated(action: (current: T, previous: T, key: K) => void): ChangeOperator; @@ -301,14 +313,14 @@ export function toObservableChangeSet(keySelector: KeySelector, opti export function toObservableChangeSet(options?: ChangeSetConversionOptions): OperatorFunction, ChangeSet>; export function toObservableChangeSet(options: ChangeSetConversionOptions & { keySelector: KeySelector }): OperatorFunction, ChangeSet>; export const ObservableChangeSet: Readonly<{ - create(subscribe: (source: SourceCache) => TeardownLogic | Disposable, keySelector: KeySelector): Observable>; - create(subscribe: (source: SourceList) => TeardownLogic | Disposable): Observable>; - createCache(subscribe: (source: SourceCache) => TeardownLogic | Disposable, keySelector: KeySelector): Observable>; - createList(subscribe: (source: SourceList) => TeardownLogic | Disposable): Observable>; + create(subscribe: (source: SourceCache) => DisposableLike, keySelector: KeySelector): Observable>; + create(subscribe: (source: SourceList) => DisposableLike): Observable>; + createCache(subscribe: (source: SourceCache) => DisposableLike, keySelector: KeySelector): Observable>; + createList(subscribe: (source: SourceList) => DisposableLike): Observable>; }>; export function asyncDisposeMany(completedAccessor?: (completion: Observable) => void, disposer?: (item: T) => PromiseLike | void): ChangeOperator; export function bufferInitial(duration: number, scheduler?: SchedulerLike): ChangeOperator; -export function filterOnProperty(property: keyof T | ((item: T) => unknown), predicate: Predicate, options?: RefreshOptions): ChangeOperator; +export function filterOnProperty(property: PropertyPath | ((item: T) => unknown), predicate: Predicate, options?: RefreshOptions): ChangeOperator; export function finallySafe(action: () => void): MonoTypeOperatorFunction; export const ConnectionStatus: Readonly<{ Pending: 'pending'; Loaded: 'loaded'; Errored: 'errored'; Completed: 'completed' }>; export type ConnectionStatusValue = typeof ConnectionStatus[keyof typeof ConnectionStatus]; diff --git a/types/lifecycle.d.ts b/types/lifecycle.d.ts index 625e269..ccc2b8d 100644 --- a/types/lifecycle.d.ts +++ b/types/lifecycle.d.ts @@ -1,2 +1,4 @@ // Module entry point: the same generic contracts as the public package. export { ConnectionStatus, ObservableChangeSet, asyncDisposeMany, autoRefresh, autoRefreshOnObservable, batch, batchIf, bufferIf, bufferInitial, createObservableObject, deferUntilLoaded, disposeMany, expireAfter, filterOnObservable, filterOnProperty, finallySafe, limitSizeTo, mergeMany, mergeManyItems, monitorStatus, notEmpty, notifyPropertyChanged, observableObject, observeCollectionChanges, observeProperty, onItemAdded, onItemRefreshed, onItemRemoved, onItemUpdated, refCount, skipInitial, subscribeMany, switch, switchLatest, toObservableChangeSet, transformAsync, transformManyAsync, transformManySafeAsync, transformOnObservable, transformSafeAsync, trueForAll, trueForAny, watch, watchValue, whenAnyPropertyChanged, whenChanged, whenPropertyChanged, whenValueChanged } from './index.js'; + +export type { DisposableLike, PropertyPath, PropertyValue, ReactivePropertyNotification } from './index.js'; diff --git a/types/operators.d.ts b/types/operators.d.ts index 0a3f1a4..aea34b7 100644 --- a/types/operators.d.ts +++ b/types/operators.d.ts @@ -1,2 +1,4 @@ // Module entry point: the same generic contracts as the public package. export { AsObservableCache, AsObservableList, Bind, BindToObservableCollection, BindToObservableList, Cast, CastToObject, ChangeKey, Convert, DistinctValues, Filter, FilterImmutable, FilterWithState, OfType, Page, RemoveKey, Reverse, Sort, SortAndBind, SortAndPage, SortAndVirtualise, SortAndVirtualize, SortBy, ToCollection, Top, Transform, TransformImmutable, TransformMany, TransformSafe, TransformWithInlineUpdate, Virtualise, Virtualize, asObservableCache, asObservableList, bind, bindToObservableCollection, bindToObservableList, cast, castToObject, changeKey, convert, distinctValues, filter, filterImmutable, filterWithState, ofType, page, removeKey, reverse, sort, sortAndBind, sortAndPage, sortAndVirtualise, sortAndVirtualize, sortBy, toCollection, top, transform, transformImmutable, transformMany, transformSafe, transformWithInlineUpdate, virtualise, virtualize } from './index.js'; + +export type { BindingTarget, ChangeSetBindingTarget, PascalCaseCollectionTarget, PascalCaseListUpdater } from './index.js'; From fa50607a397fb62ff120321f0c8f05740f18adec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wies=C5=82aw=20=C5=A0olt=C3=A9s?= Date: Sat, 12 Sep 2026 22:12:57 +0200 Subject: [PATCH 2/2] Preserve queued source changes before reentrant completion --- src/core.js | 28 +++++++++++++++++++----- test/core.test.js | 55 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 5 deletions(-) diff --git a/src/core.js b/src/core.js index e4b8616..36904bf 100644 --- a/src/core.js +++ b/src/core.js @@ -230,13 +230,12 @@ class SourceBase { if (predicate != null && typeof predicate !== 'function') throw new TypeError('Predicate must be a function'); return observableDecorator(new Observable(observer => { if (this._disposed) { observer.complete(); return; } - let version = this._version - Number(!!this._inPreview), starting = true; + let version = this._version - Number(!!this._inPreview), starting = true, draining = false; const pending = []; let projection; try { projection = predicate ? (this.kind === 'cache' ? cacheFilter(this._data, predicate) : listFilter(this._data, predicate)) : null; } catch (error) { observer.error(error); return; } const receive = event => { - if (starting) { pending.push(event); return; } if (observer.closed) return; try { const segments = event.segments.filter(segment => segment.version > version); @@ -247,16 +246,35 @@ class SourceBase { if (changes.length || !suppressEmpty) observer.next(changes); } catch (error) { observer.error(error); } }; + // Initial snapshot callbacks can mutate and terminate the source before they return. + // Keep both deltas and terminal events in order, including writes made while draining. + const drain = () => { + if (starting || draining) return; + draining = true; + try { + for (let index = 0; index < pending.length && !observer.closed; index++) { + const notification = pending[index]; + if (notification.kind === 'next') receive(notification.event); + else if (notification.kind === 'error') observer.error(notification.error); + else observer.complete(); + } + } finally { pending.length = 0; draining = false; } + }; + const enqueue = notification => { if (!observer.closed) { pending.push(notification); drain(); } }; const stream = preview ? this._previews : this._events; if (stream.hasError) { observer.error(stream.thrownError); return; } - const sub = stream.isStopped ? null : stream.subscribe({ next: receive, error: e => observer.error(e), complete: () => observer.complete() }); + const sub = stream.isStopped ? null : stream.subscribe({ + next: event => enqueue({ kind: 'next', event }), + error: error => enqueue({ kind: 'error', error }), + complete: () => enqueue({ kind: 'complete' }) + }); if (!preview) { const initial = projection ? projection.initial() : snapshotChanges(this._data, this.kind, this.keySelector); if (initial.length || !suppressEmpty) observer.next(initial); } starting = false; - for (const event of pending) receive(event); - if (stream.isStopped) observer.complete(); + if (!sub) pending.push({ kind: 'complete' }); + drain(); return sub; })); } diff --git a/test/core.test.js b/test/core.test.js index e1fbe2b..830f21d 100644 --- a/test/core.test.js +++ b/test/core.test.js @@ -150,3 +150,58 @@ test('constructed and appended operator records expose Pascal fields and typed l assert.equal(list[0].Range.Count, 2); assert.equal(list[0].Range.Index, 0); assert.equal(list.keys, undefined); }); + + +test('initial list and cache subscribers receive queued changes before source disposal completes', () => { + for (const kind of ['list', 'cache']) { + const source = kind === 'list' ? new SourceList([1]) : new SourceCache(value => value); + if (kind === 'cache') source.addOrUpdate(1); + const state = kind === 'list' ? [] : new Map(), trace = []; + const subscription = source.connect().subscribe({ + next(changes) { + applyChanges(state, changes); trace.push(Array.from(state.values())); + if (trace.length === 1) { if (kind === 'list') source.add(2); else source.addOrUpdate(2); source.dispose(); } + }, complete() { trace.push('complete'); } + }); + assert.deepEqual(trace, [[1], [1, 2], 'complete']); + assert.deepEqual(Array.from(state.values()), source.items); assert.equal(subscription.closed, true); + assert.equal(source._events.observed, false); + } +}); + +test('initial subscribers receive queued projected changes before a source error', () => { + for (const kind of ['list', 'cache']) { + const source = kind === 'list' ? new SourceList([1]) : new SourceCache(value => value); + if (kind === 'cache') source.addOrUpdate(1); + const state = kind === 'list' ? [] : new Map(), trace = [], expected = new Error('source failed'); + const subscription = source.connect(value => value % 2 !== 0).subscribe({ + next(changes) { + applyChanges(state, changes); trace.push(Array.from(state.values())); + if (trace.length === 1) { + if (kind === 'list') source.addRange([2, 3]); else source.addOrUpdate([2, 3]); + source._error(expected); + } + }, error(error) { assert.equal(error, expected); trace.push('error'); } + }); + assert.deepEqual(trace, [[1], [1, 3], 'error']); assert.equal(subscription.closed, true); + assert.equal(source._events.observed, false); + } +}); + +test('changes produced while draining startup notifications stay behind earlier queued deltas', () => { + for (const kind of ['list', 'cache']) { + const source = kind === 'list' ? new SourceList([1]) : new SourceCache(value => value); + const add = value => kind === 'list' ? source.add(value) : source.addOrUpdate(value); + if (kind === 'cache') add(1); + const state = kind === 'list' ? [] : new Map(), trace = []; + source.connect().subscribe({ + next(changes) { + applyChanges(state, changes); trace.push(Array.from(state.values())); + if (trace.length === 1) { add(2); add(3); } + else if (trace.length === 2) { add(4); source.dispose(); } + }, complete() { trace.push('complete'); } + }); + assert.deepEqual(trace, [[1], [1, 2], [1, 2, 3], [1, 2, 3, 4], 'complete']); + assert.deepEqual(Array.from(state.values()), source.items); + } +});