web: subscribe console on route changes
This commit is contained in:
@@ -560,10 +560,11 @@ Deno.test("Worker Console page is routed by runtime_id and worker_id through bac
|
||||
"reload token advancement should not synchronously read and write the rune state",
|
||||
);
|
||||
assert(
|
||||
consolePage.includes("advanceReloadToken();") &&
|
||||
consolePage.includes("void loadConsoleData(target);") &&
|
||||
consolePage.includes("const token = advanceReloadToken();") &&
|
||||
consolePage.includes("worker = targetWorker;") &&
|
||||
consolePage.includes("if (!targetWorker) void loadWorker(target, token);") &&
|
||||
!consolePage.includes("void refreshConsole();\n });\n\n $effect"),
|
||||
"target-change effect should load data without depending on manual refresh state reads",
|
||||
"target-change effect should install route data and guard fallback loading with the new target token",
|
||||
);
|
||||
assert(
|
||||
consolePage.includes(
|
||||
@@ -725,10 +726,18 @@ Deno.test("Workspace Worker list and Console share the multiplexed connection",
|
||||
new URL("./../multiplexer.ts", import.meta.url),
|
||||
);
|
||||
assert(
|
||||
consolePage.includes("workspaceMultiplexer(workspaceId)") &&
|
||||
consolePage.includes("workspaceMultiplexer(target.workspaceId)") &&
|
||||
sidebarStore.includes("workspaceMultiplexer(workspaceId)") &&
|
||||
multiplexer.includes("const multiplexers = new Map") &&
|
||||
multiplexer.includes("frame: 'worker_protocol'"),
|
||||
"Sidebar and Console should share one Workspace multiplexer and route Worker methods through a subscription lane",
|
||||
);
|
||||
assert(
|
||||
multiplexer.includes("this.#socket?.readyState === WebSocket.OPEN") &&
|
||||
multiplexer.includes("this.#sendSubscribe(subscription)") &&
|
||||
consolePage.includes("const targetWorker = data.worker") &&
|
||||
consolePage.includes("worker = targetWorker") &&
|
||||
consolePage.includes("const consoleTarget = $derived({ workspaceId, runtimeId, workerId })"),
|
||||
"A reused Console route should subscribe immediately on the live Workspace socket and install the new route Worker",
|
||||
);
|
||||
});
|
||||
|
||||
@@ -54,15 +54,21 @@ export class WorkspaceMultiplexer {
|
||||
listener: Listener,
|
||||
): WorkspaceMultiplexerSubscription {
|
||||
const clientId = crypto.randomUUID();
|
||||
this.#subscriptions.set(clientId, {
|
||||
const subscription: ActiveSubscription = {
|
||||
clientId,
|
||||
selector,
|
||||
listener,
|
||||
requestId: null,
|
||||
subscriptionId: null,
|
||||
});
|
||||
};
|
||||
this.#subscriptions.set(clientId, subscription);
|
||||
this.#closed = false;
|
||||
this.#ensureConnected();
|
||||
if (this.#socket?.readyState === WebSocket.OPEN) {
|
||||
subscription.listener.onStatus?.('open');
|
||||
this.#sendSubscribe(subscription);
|
||||
} else {
|
||||
this.#ensureConnected();
|
||||
}
|
||||
return {
|
||||
close: () => this.#remove(clientId),
|
||||
sendWorkerMethod: (method) => this.#sendWorkerMethod(clientId, method),
|
||||
|
||||
+16
-11
@@ -140,11 +140,12 @@
|
||||
let reloadToken = $state(0);
|
||||
|
||||
type ConsoleTarget = {
|
||||
workspaceId: string;
|
||||
runtimeId: string;
|
||||
workerId: string;
|
||||
};
|
||||
|
||||
const consoleTarget = $derived({ runtimeId, workerId });
|
||||
const consoleTarget = $derived({ workspaceId, runtimeId, workerId });
|
||||
|
||||
const lines = $derived(consoleProjection.lines);
|
||||
const timelineLayout = $derived(
|
||||
@@ -179,17 +180,20 @@
|
||||
return response.json() as Promise<T>;
|
||||
}
|
||||
|
||||
async function loadWorker(target: ConsoleTarget) {
|
||||
async function loadWorker(target: ConsoleTarget, token: number) {
|
||||
workerError = null;
|
||||
try {
|
||||
const payload = await getJson<Worker>(
|
||||
workerApiPath(
|
||||
workspaceApiPath(
|
||||
target.workspaceId,
|
||||
`/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}`,
|
||||
),
|
||||
);
|
||||
if (token !== reloadToken) return;
|
||||
worker = payload;
|
||||
liveWorkerState = payload.state;
|
||||
} catch (error) {
|
||||
if (token !== reloadToken) return;
|
||||
workerError =
|
||||
error instanceof Error ? error.message : String(error);
|
||||
worker = null;
|
||||
@@ -197,10 +201,6 @@
|
||||
}
|
||||
}
|
||||
|
||||
async function loadConsoleData(target: ConsoleTarget) {
|
||||
if (!worker) await loadWorker(target);
|
||||
}
|
||||
|
||||
function advanceReloadToken(): number {
|
||||
nextReloadToken += 1;
|
||||
reloadToken = nextReloadToken;
|
||||
@@ -515,7 +515,7 @@
|
||||
return;
|
||||
}
|
||||
protocolState = "connecting";
|
||||
const subscription = workspaceMultiplexer(workspaceId).subscribe(
|
||||
const subscription = workspaceMultiplexer(target.workspaceId).subscribe(
|
||||
{
|
||||
topic: "worker_protocol",
|
||||
worker_id: target.workerId,
|
||||
@@ -1083,11 +1083,16 @@
|
||||
|
||||
$effect(() => {
|
||||
const target = consoleTarget;
|
||||
const targetWorker = data.worker;
|
||||
const targetWorkerError = data.workerError;
|
||||
resetObservedEvents();
|
||||
liveWorkerState = null;
|
||||
worker = targetWorker;
|
||||
workerError = targetWorkerError;
|
||||
liveWorkerState = targetWorker?.state ?? null;
|
||||
streamDiagnostics = [];
|
||||
advanceReloadToken();
|
||||
void loadConsoleData(target);
|
||||
protocolState = "connecting";
|
||||
const token = advanceReloadToken();
|
||||
if (!targetWorker) void loadWorker(target, token);
|
||||
});
|
||||
|
||||
$effect(() => connectProtocolTransport(worker, reloadToken, consoleTarget));
|
||||
|
||||
Reference in New Issue
Block a user