syndicate-lang
/
syndicate-js
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Repair route with no steps

This commit is contained in:
Tony Garnock-Jones 2023-02-11 22:01:19 +01:00
parent 7a09be7141
commit a51851283f
2 changed files with 33 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/core/src/runtime/dataflow.ts
View File

@ -112,9 +112,13 @@ export abstract class Cell {
return this.__value; return this.__value;
} }
changed() {
this.graph.recordDamage(this);
}
update(newValue: unknown) { update(newValue: unknown) {
if (!this.valuesEqual(this.__value, newValue)) { if (!this.valuesEqual(this.__value, newValue)) {
this.graph.recordDamage(this); this.changed();
this.__value = newValue; this.__value = newValue;
} }
} }

67
packages/ws-relay/src/index.ts
View File

@ -6,6 +6,7 @@ import {
Bytes, Bytes,
Dataflow, Dataflow,
Embedded, Embedded,
IdentitySet,
Observe, Observe,
QuasiValue as Q, QuasiValue as Q,
Ref, Ref,
@ -110,7 +111,7 @@ export function boot(ds: Ref, debug: boolean = false) {
ws.binaryType = 'arraybuffer'; ws.binaryType = 'arraybuffer';
ws.onopen = () => facet.turn(() => succeed(ws)); ws.onopen = () => facet.turn(() => succeed(ws));
ws.onclose = () => facet.turn(() => fail(Symbol.for('closed'))); ws.onclose = () => facet.turn(() => fail(Symbol.for('closed')));
ws.onerror = (e) => facet.turn(() => fail(Symbol.for('websocket-error-event'))); ws.onerror = () => facet.turn(() => fail(Symbol.for('websocket-error-event')));
} catch (e) { } catch (e) {
console.error('Failed opening websocket', addr.url, e); console.error('Failed opening websocket', addr.url, e);
fail(Symbol.for('websocket-exception')); fail(Symbol.for('websocket-exception'));
@ -127,36 +128,26 @@ export function boot(ds: Ref, debug: boolean = false) {
}) }) => { }) }) => {
const route = Q.drop_lit(routePatValue, G.toRoute); const route = Q.drop_lit(routePatValue, G.toRoute);
if (!route) return; if (!route) return;
const candidates: Dataflow.Field<TransportState | null>[] = []; field candidates: IdentitySet<TransportState> = new IdentitySet();
route.transports.forEach(t => { route.transports.forEach(t => {
const addr = T.toWebSocket(t); const addr = T.toWebSocket(t);
if (!addr) return; if (!addr) return;
let counter = 0;
field state: TransportState | null = null;
candidates.push(state);
console.log('tracking', addr.url); console.log('tracking', addr.url);
during G.TransportConnection({ during G.TransportConnection({
"addr": addr, "addr": addr,
"control": $control: Embedded, "control": $control_e: Embedded,
"resolved": $resolved: G.Resolved, "resolved": G.Resolved.accepted($peer_e: Embedded),
}) => { }) => {
const me = counter++; const entry = {
switch (resolved._variant) { addr,
case "accepted": control: control_e.embeddedValue,
state.value = { peer: peer_e.embeddedValue,
addr, };
control: control.embeddedValue, candidates.value.add(entry);
peer: resolved.responderSession, candidates.changed();
};
break;
case "Rejected":
state.value = null;
break;
}
on stop { on stop {
if (counter === me) { candidates.value.delete(entry);
state.value = null; candidates.changed();
}
} }
} }
}); });
@ -164,22 +155,20 @@ export function boot(ds: Ref, debug: boolean = false) {
field rootPeer: Ref | null = null; field rootPeer: Ref | null = null;
dataflow { dataflow {
best.value = null; best.value = null;
for (const c of candidates) { for (const c of candidates.value) {
if (c.value !== null) { best.value = c;
if (best.value === null) best.value = c.value; break;
}
} }
rootPeer.value = best.value?.peer ?? null; rootPeer.value = best.value?.peer ?? null;
} }
resolve(() => rootPeer.value, route.pathSteps, (r) => { resolve(() => rootPeer.value, route.pathSteps, (r) => {
const s = best.value!; console.log('leaf', best.value?.addr?.url);
console.log('leaf', s.addr.url, stringify(r));
assert G.ResolvePath<Ref>({ assert G.ResolvePath<Ref>({
"route": route, "route": route,
"addr": fromJS(s.addr), "addr": fromJS(best.value!.addr),
"control": s.control, "control": best.value!.control,
"resolved": r, "resolved": r()!
}); }) when (r());
}); });
} }
} }
@ -188,13 +177,13 @@ export function boot(ds: Ref, debug: boolean = false) {
function resolve( function resolve(
e: () => Ref | null, e: () => Ref | null,
steps: G.PathStep[], steps: G.PathStep[],
k: (r: G.Resolved) => void, k: (r: () => G.Resolved | null) => void,
) { ) {
if (steps.length === 0) { if (steps.length === 0) {
const peer = e(); k(() => {
k(peer === null const peer = e();
? G.Resolved.Rejected(G.Rejected(Symbol.for('not-connected'))) return peer === null ? null : G.Resolved.accepted(peer);
: G.Resolved.accepted(peer)); });
} else { } else {
const [step, ...more] = steps; const [step, ...more] = steps;
at ds { at ds {
@ -208,7 +197,7 @@ export function boot(ds: Ref, debug: boolean = false) {
resolve(() => resolved.responderSession, more, k); resolve(() => resolved.responderSession, more, k);
break; break;
case "Rejected": case "Rejected":
k(resolved); k(() => resolved);
break; break;
} }
} }