Add managed desktop daemon runtime support
This commit is contained in:
Mohamed Boudra
@@ -7,7 +7,7 @@ import {
|
||||
} from "./connection-selection";
|
||||
|
||||
function makeDirect(id: string, endpoint: string): HostConnection {
|
||||
return { id, type: "direct", endpoint };
|
||||
return { id, type: "directTcp", endpoint };
|
||||
}
|
||||
|
||||
function makeRelay(
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
|
||||
const managedRuntimeMock = vi.hoisted(() => {
|
||||
let eventHandler: ((payload: {
|
||||
sessionId: string;
|
||||
kind: "open" | "message" | "close" | "error";
|
||||
text?: string | null;
|
||||
binaryBase64?: string | null;
|
||||
code?: number | null;
|
||||
reason?: string | null;
|
||||
error?: string | null;
|
||||
}) => void) | null = null;
|
||||
|
||||
const openLocalTransportSession = vi.fn<(...args: unknown[]) => Promise<string>>();
|
||||
const listenToLocalTransportEvents = vi.fn(async (handler: typeof eventHandler extends ((...args: infer A) => any) | null ? (...args: A) => void : never) => {
|
||||
eventHandler = handler;
|
||||
return () => {
|
||||
eventHandler = null;
|
||||
};
|
||||
});
|
||||
const sendLocalTransportMessage = vi.fn(async () => undefined);
|
||||
const closeLocalTransportSession = vi.fn(async () => undefined);
|
||||
|
||||
return {
|
||||
openLocalTransportSession,
|
||||
listenToLocalTransportEvents,
|
||||
sendLocalTransportMessage,
|
||||
closeLocalTransportSession,
|
||||
emitEvent(payload: {
|
||||
sessionId: string;
|
||||
kind: "open" | "message" | "close" | "error";
|
||||
text?: string | null;
|
||||
binaryBase64?: string | null;
|
||||
code?: number | null;
|
||||
reason?: string | null;
|
||||
error?: string | null;
|
||||
}) {
|
||||
eventHandler?.(payload);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/desktop/managed-runtime/managed-runtime", () => ({
|
||||
openLocalTransportSession: managedRuntimeMock.openLocalTransportSession,
|
||||
listenToLocalTransportEvents: managedRuntimeMock.listenToLocalTransportEvents,
|
||||
sendLocalTransportMessage: managedRuntimeMock.sendLocalTransportMessage,
|
||||
closeLocalTransportSession: managedRuntimeMock.closeLocalTransportSession,
|
||||
}));
|
||||
|
||||
describe("managed-tauri-daemon-transport", () => {
|
||||
beforeEach(() => {
|
||||
managedRuntimeMock.openLocalTransportSession.mockReset();
|
||||
managedRuntimeMock.listenToLocalTransportEvents.mockClear();
|
||||
managedRuntimeMock.sendLocalTransportMessage.mockClear();
|
||||
managedRuntimeMock.closeLocalTransportSession.mockClear();
|
||||
});
|
||||
|
||||
it("emits open after the session resolves even if the rust open event raced earlier", async () => {
|
||||
let resolveSession!: (sessionId: string) => void;
|
||||
managedRuntimeMock.openLocalTransportSession.mockImplementation(
|
||||
() =>
|
||||
new Promise<string>((resolve) => {
|
||||
resolveSession = resolve;
|
||||
})
|
||||
);
|
||||
|
||||
const mod = await import("./managed-tauri-daemon-transport");
|
||||
const transportFactory = mod.createTauriLocalDaemonTransportFactory();
|
||||
expect(transportFactory).not.toBeNull();
|
||||
|
||||
const transport = transportFactory!({
|
||||
url: "paseo+local://socket?path=%2Ftmp%2Fpaseo.sock",
|
||||
});
|
||||
|
||||
const onOpen = vi.fn();
|
||||
transport.onOpen(onOpen);
|
||||
|
||||
managedRuntimeMock.emitEvent({
|
||||
sessionId: "local-session-1",
|
||||
kind: "open",
|
||||
});
|
||||
|
||||
expect(onOpen).not.toHaveBeenCalled();
|
||||
|
||||
resolveSession("local-session-1");
|
||||
await Promise.resolve();
|
||||
|
||||
expect(onOpen).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("cleans up late async setup after the transport is closed", async () => {
|
||||
let resolveSession!: (sessionId: string) => void;
|
||||
let resolveListen!: (cleanup: () => void) => void;
|
||||
const cleanup = vi.fn();
|
||||
|
||||
managedRuntimeMock.openLocalTransportSession.mockImplementation(
|
||||
() =>
|
||||
new Promise<string>((resolve) => {
|
||||
resolveSession = resolve;
|
||||
})
|
||||
);
|
||||
managedRuntimeMock.listenToLocalTransportEvents.mockImplementation(
|
||||
() =>
|
||||
new Promise<() => void>((resolve) => {
|
||||
resolveListen = resolve;
|
||||
})
|
||||
);
|
||||
|
||||
const mod = await import("./managed-tauri-daemon-transport");
|
||||
const transportFactory = mod.createTauriLocalDaemonTransportFactory();
|
||||
expect(transportFactory).not.toBeNull();
|
||||
|
||||
const transport = transportFactory!({
|
||||
url: "paseo+local://socket?path=%2Ftmp%2Fpaseo.sock",
|
||||
});
|
||||
|
||||
transport.close();
|
||||
|
||||
resolveSession("local-session-2");
|
||||
resolveListen(cleanup);
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
expect(managedRuntimeMock.closeLocalTransportSession).toHaveBeenCalledWith("local-session-2");
|
||||
expect(cleanup).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,186 @@
|
||||
import type { DaemonTransport, DaemonTransportFactory } from "@server/client/daemon-client";
|
||||
import {
|
||||
closeLocalTransportSession,
|
||||
listenToLocalTransportEvents,
|
||||
openLocalTransportSession,
|
||||
sendLocalTransportMessage,
|
||||
type LocalTransportTarget,
|
||||
} from "@/desktop/managed-runtime/managed-runtime";
|
||||
|
||||
const LOCAL_TRANSPORT_SCHEME = "paseo+local:";
|
||||
|
||||
function encodeBinaryToBase64(data: Uint8Array | ArrayBuffer): string {
|
||||
const bytes = data instanceof ArrayBuffer ? new Uint8Array(data) : data;
|
||||
let binary = "";
|
||||
for (let index = 0; index < bytes.length; index += 1) {
|
||||
binary += String.fromCharCode(bytes[index]!);
|
||||
}
|
||||
return globalThis.btoa(binary);
|
||||
}
|
||||
|
||||
function decodeBase64ToBytes(base64: string): Uint8Array {
|
||||
const binary = globalThis.atob(base64);
|
||||
const bytes = new Uint8Array(binary.length);
|
||||
for (let index = 0; index < binary.length; index += 1) {
|
||||
bytes[index] = binary.charCodeAt(index);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
export function buildLocalDaemonTransportUrl(target: LocalTransportTarget): string {
|
||||
const url = new URL(`${LOCAL_TRANSPORT_SCHEME}//${target.transportType}`);
|
||||
url.searchParams.set("path", target.transportPath);
|
||||
return url.toString();
|
||||
}
|
||||
|
||||
function parseLocalDaemonTransportUrl(url: string): LocalTransportTarget {
|
||||
const parsed = new URL(url);
|
||||
if (parsed.protocol !== LOCAL_TRANSPORT_SCHEME) {
|
||||
throw new Error(`Unsupported local transport URL: ${url}`);
|
||||
}
|
||||
const transportType = parsed.hostname;
|
||||
const transportPath = parsed.searchParams.get("path")?.trim() ?? "";
|
||||
if ((transportType !== "socket" && transportType !== "pipe") || !transportPath) {
|
||||
throw new Error(`Invalid local transport target: ${url}`);
|
||||
}
|
||||
return {
|
||||
transportType,
|
||||
transportPath,
|
||||
};
|
||||
}
|
||||
|
||||
export function createTauriLocalDaemonTransportFactory(): DaemonTransportFactory | null {
|
||||
return ({ url }) => {
|
||||
const target = parseLocalDaemonTransportUrl(url);
|
||||
let sessionId: string | null = null;
|
||||
let unlisten: (() => void) | null = null;
|
||||
let disposed = false;
|
||||
let didEmitOpen = false;
|
||||
|
||||
const openHandlers = new Set<() => void>();
|
||||
const closeHandlers = new Set<(event?: unknown) => void>();
|
||||
const errorHandlers = new Set<(event?: unknown) => void>();
|
||||
const messageHandlers = new Set<(data: unknown) => void>();
|
||||
|
||||
const emitOpen = () => {
|
||||
if (didEmitOpen || disposed) {
|
||||
return;
|
||||
}
|
||||
didEmitOpen = true;
|
||||
for (const handler of openHandlers) {
|
||||
handler();
|
||||
}
|
||||
};
|
||||
const emitClose = (event?: unknown) => {
|
||||
for (const handler of closeHandlers) {
|
||||
handler(event);
|
||||
}
|
||||
};
|
||||
const emitError = (event?: unknown) => {
|
||||
for (const handler of errorHandlers) {
|
||||
handler(event);
|
||||
}
|
||||
};
|
||||
const emitMessage = (data: unknown) => {
|
||||
for (const handler of messageHandlers) {
|
||||
handler(data);
|
||||
}
|
||||
};
|
||||
|
||||
void listenToLocalTransportEvents((payload) => {
|
||||
if (disposed || !sessionId || payload.sessionId !== sessionId) {
|
||||
return;
|
||||
}
|
||||
if (payload.kind === "open") {
|
||||
emitOpen();
|
||||
return;
|
||||
}
|
||||
if (payload.kind === "message") {
|
||||
if (payload.text) {
|
||||
emitMessage({ data: payload.text });
|
||||
return;
|
||||
}
|
||||
if (payload.binaryBase64) {
|
||||
emitMessage({ data: decodeBase64ToBytes(payload.binaryBase64) });
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (payload.kind === "close") {
|
||||
emitClose(payload);
|
||||
return;
|
||||
}
|
||||
emitError(payload.error ?? "Local daemon transport error");
|
||||
})
|
||||
.then((cleanup) => {
|
||||
if (disposed) {
|
||||
cleanup();
|
||||
return;
|
||||
}
|
||||
unlisten = cleanup;
|
||||
})
|
||||
.catch((error) => {
|
||||
emitError(error);
|
||||
});
|
||||
|
||||
void openLocalTransportSession(target)
|
||||
.then((id) => {
|
||||
if (disposed) {
|
||||
void closeLocalTransportSession(id).catch((error) => emitError(error));
|
||||
return;
|
||||
}
|
||||
sessionId = id;
|
||||
emitOpen();
|
||||
})
|
||||
.catch((error) => {
|
||||
emitError(error);
|
||||
});
|
||||
|
||||
const transport: DaemonTransport = {
|
||||
send: (data) => {
|
||||
if (!sessionId) {
|
||||
return;
|
||||
}
|
||||
if (typeof data === "string") {
|
||||
void sendLocalTransportMessage({ sessionId, text: data }).catch((error) =>
|
||||
emitError(error)
|
||||
);
|
||||
return;
|
||||
}
|
||||
const binaryBase64 = encodeBinaryToBase64(
|
||||
data instanceof ArrayBuffer ? data : new Uint8Array(data)
|
||||
);
|
||||
void sendLocalTransportMessage({ sessionId, binaryBase64 }).catch((error) =>
|
||||
emitError(error)
|
||||
);
|
||||
},
|
||||
close: () => {
|
||||
disposed = true;
|
||||
const currentSessionId = sessionId;
|
||||
sessionId = null;
|
||||
if (currentSessionId) {
|
||||
void closeLocalTransportSession(currentSessionId).catch((error) => emitError(error));
|
||||
}
|
||||
unlisten?.();
|
||||
unlisten = null;
|
||||
},
|
||||
onMessage: (handler) => {
|
||||
messageHandlers.add(handler);
|
||||
return () => messageHandlers.delete(handler);
|
||||
},
|
||||
onOpen: (handler) => {
|
||||
openHandlers.add(handler);
|
||||
return () => openHandlers.delete(handler);
|
||||
},
|
||||
onClose: (handler) => {
|
||||
closeHandlers.add(handler);
|
||||
return () => closeHandlers.delete(handler);
|
||||
},
|
||||
onError: (handler) => {
|
||||
errorHandlers.add(handler);
|
||||
return () => errorHandlers.delete(handler);
|
||||
},
|
||||
};
|
||||
|
||||
return transport;
|
||||
};
|
||||
}
|
||||
@@ -71,12 +71,12 @@ describe("test-daemon-connection probe client identity", () => {
|
||||
|
||||
await mod.measureConnectionLatency({
|
||||
id: "direct:lan:6767",
|
||||
type: "direct",
|
||||
type: "directTcp",
|
||||
endpoint: "lan:6767",
|
||||
});
|
||||
await mod.measureConnectionLatency({
|
||||
id: "direct:lan:6767",
|
||||
type: "direct",
|
||||
type: "directTcp",
|
||||
endpoint: "lan:6767",
|
||||
});
|
||||
|
||||
@@ -85,4 +85,18 @@ describe("test-daemon-connection probe client identity", () => {
|
||||
expect(second?.clientId).toBe("cid_shared_probe_test");
|
||||
expect(clientIdMock.getOrCreateClientId).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("encodes the local socket target into the probe client config", async () => {
|
||||
const mod = await import("./test-daemon-connection");
|
||||
|
||||
await mod.measureConnectionLatency({
|
||||
id: "socket:/tmp/paseo.sock",
|
||||
type: "directSocket",
|
||||
path: "/tmp/paseo.sock",
|
||||
});
|
||||
|
||||
expect(daemonClientMock.createdConfigs[0]?.url).toBe(
|
||||
"paseo+local://socket?path=%2Ftmp%2Fpaseo.sock"
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -3,6 +3,10 @@ import type { DaemonClientConfig } from "@server/client/daemon-client";
|
||||
import type { HostConnection } from "@/contexts/daemon-registry-context";
|
||||
import { getOrCreateClientId } from "./client-id";
|
||||
import { buildDaemonWebSocketUrl, buildRelayWebSocketUrl } from "./daemon-endpoints";
|
||||
import {
|
||||
buildLocalDaemonTransportUrl,
|
||||
createTauriLocalDaemonTransportFactory,
|
||||
} from "./managed-tauri-daemon-transport";
|
||||
import { createTauriWebSocketTransportFactory } from "./tauri-daemon-transport";
|
||||
|
||||
function normalizeNonEmptyString(value: unknown): string | null {
|
||||
@@ -47,14 +51,31 @@ async function buildClientConfig(
|
||||
): Promise<DaemonClientConfig> {
|
||||
const clientId = await getOrCreateClientId();
|
||||
const tauriTransportFactory = createTauriWebSocketTransportFactory();
|
||||
const localTransportFactory = createTauriLocalDaemonTransportFactory();
|
||||
const base = {
|
||||
clientId,
|
||||
clientType: "mobile" as const,
|
||||
suppressSendErrors: true,
|
||||
...(tauriTransportFactory ? { transportFactory: tauriTransportFactory } : {}),
|
||||
...(connection.type === "directSocket" || connection.type === "directPipe"
|
||||
? localTransportFactory
|
||||
? { transportFactory: localTransportFactory }
|
||||
: {}
|
||||
: tauriTransportFactory
|
||||
? { transportFactory: tauriTransportFactory }
|
||||
: {}),
|
||||
};
|
||||
|
||||
if (connection.type === "direct") {
|
||||
if (connection.type === "directSocket" || connection.type === "directPipe") {
|
||||
return {
|
||||
...base,
|
||||
url: buildLocalDaemonTransportUrl({
|
||||
transportType: connection.type === "directSocket" ? "socket" : "pipe",
|
||||
transportPath: connection.path,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
if (connection.type === "directTcp") {
|
||||
return {
|
||||
...base,
|
||||
url: buildDaemonWebSocketUrl(connection.endpoint),
|
||||
|
||||
Reference in New Issue
Block a user