style: format state subscriptions and their adopted controls

This commit is contained in:
Sameh Khamis committed 2026-09-13 05:43:05 -07:00
1 parent 45d97376d0
commit e0652b19f2
12 files changed
+299 -88

No files matched your search

+39 -11
View File
@@ -2,17 +2,27 @@
function copyState(value, ancestors = new Set()) {
if (value === null) return value;
if (typeof value !== 'object') {
if (['string', 'number', 'boolean', 'undefined'].includes(typeof value)) return value;
if (['string', 'number', 'boolean', 'undefined'].includes(typeof value))
return value;
throw new TypeError('Product state must contain plain data');
}
if (ancestors.has(value)) throw new TypeError('Circular product state');
if (!Array.isArray(value) && Object.getPrototypeOf(value) !== Object.prototype && Object.getPrototypeOf(value) !== null) {
if (
!Array.isArray(value) &&
Object.getPrototypeOf(value) !== Object.prototype &&
Object.getPrototypeOf(value) !== null
) {
throw new TypeError('Product state must contain plain records');
}
const next = new Set(ancestors).add(value);
const copy = Array.isArray(value)
? value.map((item) => copyState(item, next))
: Object.fromEntries(Object.entries(value).map(([key, item]) => [key, copyState(item, next)]));
: Object.fromEntries(
Object.entries(value).map(([key, item]) => [
key,
copyState(item, next),
]),
);
return Object.freeze(copy);
}
@@ -42,28 +52,46 @@ export function createStateChannel(readSnapshot) {
return Object.freeze({
getSnapshot,
subscribe(listener, { emitCurrent = true } = {}) {
if (typeof listener !== 'function') throw new TypeError('Expected a state listener');
if (typeof listener !== 'function')
throw new TypeError('Expected a state listener');
if (destroyed) return () => {};
const state = emitCurrent ? getSnapshot() : null;
const entry = { listener, revision };
listeners.add(entry);
if (emitCurrent) deliver(entry, Object.freeze({
state, change: null, revision, initial: true,
}));
if (emitCurrent)
deliver(
entry,
Object.freeze({
state,
change: null,
revision,
initial: true,
}),
);
return () => listeners.delete(entry);
},
publish(change = null) {
if (destroyed) return false;
pending.push(Object.freeze({
state: getSnapshot(), change: copyState(change), revision: ++revision, initial: false,
}));
pending.push(
Object.freeze({
state: getSnapshot(),
change: copyState(change),
revision: ++revision,
initial: false,
}),
);
if (delivering) return true;
delivering = true;
try {
while (!destroyed && pending.length) {
const notification = pending.shift();
for (const entry of [...listeners]) {
if (destroyed || !listeners.has(entry) || entry.revision >= notification.revision) continue;
if (
destroyed ||
!listeners.has(entry) ||
entry.revision >= notification.revision
)
continue;
entry.revision = notification.revision;
deliver(entry, notification);
}
+17 -9
View File
@@ -22,7 +22,9 @@ test('snapshots and action data are immutable copies of live product state', ()
test('reentrant updates retain order and a new subscriber never regresses behind its initial snapshot', () => {
let count = 0;
const channel = createStateChannel(() => ({ count }));
const first = [], second = [], joined = [];
const first = [],
second = [],
joined = [];
channel.subscribe(({ state, initial }) => {
if (initial) return;
first.push(state.count);
@@ -32,7 +34,9 @@ test('reentrant updates retain order and a new subscriber never regresses behind
channel.subscribe(({ state }) => joined.push(state.count));
}
});
channel.subscribe(({ state, initial }) => { if (!initial) second.push(state.count); });
channel.subscribe(({ state, initial }) => {
if (!initial) second.push(state.count);
});
count = 1;
channel.publish();
assert.deepEqual(first, [1, 2]);
@@ -45,11 +49,14 @@ test('unsubscribe and destroy during delivery revoke pending callbacks and futur
const channel = createStateChannel(() => ({ reads: ++reads }));
let late = 0;
let remove;
channel.subscribe(() => {
remove();
channel.publish();
channel.destroy();
}, { emitCurrent: false });
channel.subscribe(
() => {
remove();
channel.publish();
channel.destroy();
},
{ emitCurrent: false },
);
remove = channel.subscribe(() => late++, { emitCurrent: false });
channel.publish();
const stoppedReads = reads;
@@ -67,7 +74,9 @@ test('a failed subscriber cannot prevent other controls from receiving an update
const original = console.error;
console.error = (message) => errors.push(message);
try {
channel.subscribe(() => { throw new Error('consumer failure'); });
channel.subscribe(() => {
throw new Error('consumer failure');
});
let received = 0;
channel.subscribe(() => received++);
channel.publish();
@@ -79,7 +88,6 @@ test('a failed subscriber cannot prevent other controls from receiving an update
}
});
test('an invalid initial snapshot cannot retain a subscriber without a cleanup handle', () => {
let live = { action() {} };
const channel = createStateChannel(() => live);