68 lines
4.8 KiB
TypeScript
68 lines
4.8 KiB
TypeScript
/**
|
|
* 按元素 ID 精确路由的只读查询源。一个上游 editor 订阅服务所有元素桶；
|
|
* 普通提交用 changes 找受影响 ID，整份文档替换或加载覆盖变化则做必要的保守复查。
|
|
*/
|
|
import type {Commit,Id} from '../canvas-core/types.ts';
|
|
import type {ElementLookup} from '../canvas-core/queries.ts';
|
|
import type {EngineSnapshot} from '../canvas-runtime/contracts.ts';
|
|
import {equal} from '../canvas-core/values.ts';
|
|
import type {ExternalSource} from './selectors.ts';
|
|
|
|
/** 注入根读取、发布订阅与元素查询，绑定层不另建可以编辑的元素表。 */
|
|
export type ElementSourcePorts={
|
|
getSnapshot():EngineSnapshot;
|
|
subscribe(listener:(commit:Commit,snapshot:EngineSnapshot)=>void):()=>void;
|
|
lookup(snapshot:EngineSnapshot,id:Id):ElementLookup;
|
|
};
|
|
/** One upstream subscription, ID buckets and cached lookup values, never another document.
|
|
* Explicit changes use ID routing; replacement/coverage cuts retain a checked fallback. */
|
|
/** 创建惰性订阅中心；每个 forElement 源缓存自身读取值，同 ID 的监听共享上游桶。 */
|
|
export function createElementSources(ports:ElementSourcePorts){
|
|
// 失败标记使下次成功查询即使值等价也通知恢复；不把异常缓存成一条虚假元素记录。
|
|
type Bucket={value:ElementLookup;listeners:Set<()=>void>;failed?:boolean};
|
|
const buckets=new Map<Id,Bucket>();let stop:(()=>void)|undefined,previous:EngineSnapshot|undefined;
|
|
/** 先更新全部命中桶再通知，逐个隔离监听器错误，最终重抛首个错误给上游边界。 */
|
|
function changed(commit:Commit,next:EngineSnapshot){
|
|
// The editor snapshots its listeners; this callback can already be unsubscribed.
|
|
if(!previous||!buckets.size)return;
|
|
const before=previous!;previous=next;
|
|
const ids=new Set<Id>();
|
|
if(before.document!==next.document){
|
|
// 显式元素/字段改动走 ID 路由；无 changes 的文档替换无法精确定位，复查全部已订阅元素。
|
|
if(commit.changes.length){for(const change of commit.changes)if(change.kind==='field')ids.add(change.target.id);else if(change.kind==='element')ids.add(change.id);}
|
|
else for(const id of buckets.keys())ids.add(id);
|
|
}
|
|
// loading 的已知范围改变会把 absent/unloaded 变成可查询；原先非 present 的桶必须复查。
|
|
const coverageChanged=before.runtime.loading!==next.runtime.loading||!!next.runtime.loading&&(before.runtime.loadingRead?.document??before.runtime.acknowledged)!==(next.runtime.loadingRead?.document??next.runtime.acknowledged);
|
|
if(coverageChanged)for(const[id,bucket]of buckets)if(bucket.value.status!=='present')ids.add(id);
|
|
const notifications=new Set<()=>void>();
|
|
for(const id of ids){
|
|
const bucket=buckets.get(id);if(!bucket)continue;
|
|
try{const value=ports.lookup(next,id);if(!bucket.failed&&equal(bucket.value,value))continue;bucket.value=value;bucket.failed=false;}catch{bucket.failed=true;/* Snapshot readers retry/raise through their own boundary. */}
|
|
for(const notify of bucket.listeners)notifications.add(notify);
|
|
}
|
|
// All bucket values advance before observers run; a failing observer cannot starve others.
|
|
let failed=false,error:unknown;
|
|
for(const notify of notifications)try{notify();}catch(caught){if(!failed){failed=true;error=caught;}}
|
|
if(failed)throw error;
|
|
}
|
|
/** 创建单元素外部源；单纯创建源不注册上游，只有 subscribe 才建立观察关系。 */
|
|
return Object.freeze({forElement(id:Id):ExternalSource<ElementLookup>{
|
|
let value:ElementLookup|undefined;
|
|
return Object.freeze({
|
|
/** 每次从当前根查询并按值保留稳定引用，不能只依赖通知时机猜测最新数据。 */
|
|
getSnapshot(){const next=ports.lookup(ports.getSnapshot(),id);if(!value||!equal(value,next))value=next;return value;},
|
|
/** 每个订阅包装独立 listener，即使调用方传同一个函数也能分别释放。 */
|
|
subscribe(notify:()=>void){
|
|
let bucket=buckets.get(id);if(!bucket){bucket={value:ports.lookup(ports.getSnapshot(),id),listeners:new Set()};buckets.set(id,bucket);}
|
|
// Distinct subscriptions may intentionally share the same callback identity.
|
|
const listener=()=>notify();bucket.listeners.add(listener);
|
|
// 首次连上游抛错时撤销这次 listener/空桶，避免半注册泄漏。
|
|
if(!stop){previous=ports.getSnapshot();try{stop=ports.subscribe(changed);}catch(error){bucket.listeners.delete(listener);if(!bucket.listeners.size)buckets.delete(id);if(!buckets.size)previous=undefined;throw error;}}
|
|
// 最后一个桶清空才退订上游，释放 previous 根引用；其它元素仍有监听时继续共享。
|
|
let active=true;return()=>{if(!active)return;active=false;bucket!.listeners.delete(listener);if(!bucket!.listeners.size)buckets.delete(id);if(!buckets.size){stop?.();stop=undefined;previous=undefined;}};
|
|
},
|
|
});
|
|
}});
|
|
}
|