74 lines
5.5 KiB
TypeScript
74 lines
5.5 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. */
|
|
/** buckets 按 elementId 共享上游查询结果/观察器，每个 forElement 另有即时读取缓存。
|
|
* 首次 subscribe 才接上游，全部桶退订才释放 previous 根引用；创建源或直接读不持有上游租约。
|
|
* 依赖不可变快照与完整 Commit.changes 合同，不从队列/网络结果猜测元素是否已改变。 */
|
|
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){
|
|
// 显式变化只路由 field.target.id / element.id，不按 canvas 变更自动扩散到全体子元素。
|
|
// 文档替换且无 changes 时复查全部已订阅桶；路由准确性依赖发布者提供完整变更及 lookup 的依赖合同。
|
|
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 桶。
|
|
// 不将任意 runtime 更新都当元素变化；failed 桶也只有进入本次候选 IDs 才会在此重新查询。
|
|
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({
|
|
/** 每次查询当前根，用 equal 保持等价 ElementLookup 的局部引用；读取不改共享桶通知基线。
|
|
* 不深复制查询值，lookup 必须返回适合不可变使用的结果；异常直接抛给调用方读取边界。 */
|
|
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;}}
|
|
// 最后一个桶清空才 stop 上游并释放 previous；同回调的多次订阅因 listener 包装仍分别退订。
|
|
// stop 异常不被捕获，后续状态重置不保证执行；宿主的取消订阅端口应不抛错。
|
|
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;}};
|
|
},
|
|
});
|
|
}});
|
|
}
|