forked from
tokono.ma/diffuse
A music player that connects to your cloud/distributed storage.
chore: include custom rpc for performance
+190
-199
-1
deno.jsonc
-1
deno.jsonc
···
22
22
"@fcrozatier/htmlcrunch": "jsr:@fcrozatier/htmlcrunch@^1.5.1",
23
23
"@fry69/deep-diff": "jsr:@fry69/deep-diff@^0.1.10",
24
24
"@js-temporal/polyfill": "npm:@js-temporal/polyfill@^0.5.1",
25
-
"@kunkun/kkrpc": "jsr:@kunkun/kkrpc@0.6.0",
26
25
"@mary/ds-queue": "jsr:@mary/ds-queue@^0.1.3",
27
26
"@okikio/transferables": "jsr:@okikio/transferables@^1.0.2",
28
27
"@orama/orama": "npm:@orama/orama@^3.1.18",
+15
-24
src/common/element.js
+15
-24
src/common/element.js
···
1
1
import QS from "query-string";
2
-
import { decodeMessage, encodeMessage, RPCChannel } from "@kunkun/kkrpc";
3
2
import { html, render } from "lit-html";
4
3
5
4
import { effect, signal } from "@common/signal.js";
6
5
import { rpc, workerLink, workerProxy, workerTunnel } from "./worker.js";
7
-
import { BrowserPostMessageIo } from "./worker/rpc.js";
6
+
import { RpcChannel } from "./worker/rpc-channel.js";
8
7
9
8
/**
10
9
* @import {BroadcastingStatus, WorkerOpts} from "./element.d.ts"
···
233
232
},
234
233
);
235
234
236
-
const decoded = await decodeMessage(msg);
237
235
const data = {
238
-
data: Array.isArray(decoded.args) ? decoded.args[0] : decoded.args,
236
+
data: Array.isArray(msg.args) ? msg.args[0] : msg.args,
239
237
ports: Object.fromEntries(ports.map(([k, v]) => {
240
238
return [k, v.port];
241
239
})),
242
240
};
243
241
244
-
const encoded = encodeMessage(
245
-
{
246
-
...decoded,
247
-
args: Array.isArray(decoded.args)
248
-
? [data, ...decoded.args.slice(1)]
249
-
: decoded.args,
250
-
},
251
-
{},
252
-
true,
253
-
ports.map(([_k, v]) => v.port),
254
-
);
255
-
256
242
this.#disposables.push(() => {
257
243
ports.forEach(([_k, v]) => v.disconnect());
258
244
});
259
245
260
246
return {
261
-
data: encoded,
247
+
data: {
248
+
...msg,
249
+
args: Array.isArray(msg.args)
250
+
? [data, ...msg.args.slice(1)]
251
+
: msg.args,
252
+
},
262
253
transfer: ports.map(([_k, v]) => v.port),
263
254
};
264
255
};
···
333
324
channel.addEventListener(
334
325
"message",
335
326
async (event) => {
336
-
if (event.data?.includes('"method":"leader:')) {
327
+
if (event.data?.method?.startsWith("leader:")) {
337
328
const status = await this.#status.promise;
338
329
if (status.leader) {
339
-
const json = event.data.replace('"method":"leader:', '"method":"');
340
-
msg.port1.postMessage(json);
330
+
msg.port1.postMessage({
331
+
...event.data,
332
+
method: event.data.method.slice("leader:".length),
333
+
});
341
334
}
342
335
} else {
343
336
msg.port1.postMessage(event.data);
···
358
351
return !!state.pending?.length;
359
352
}
360
353
361
-
const io = new BrowserPostMessageIo(() => msg.port2);
362
-
363
-
/** @type {undefined | RPCChannel<{}, ProxiedActions<Actions>>} */
364
-
const proxyChannel = new RPCChannel(io, { enableTransfer: false });
354
+
/** @type {RpcChannel<{}, Actions>} */
355
+
const proxyChannel = new RpcChannel(msg.port2);
365
356
366
357
/** @type {ProxiedActions<Actions>} */
367
358
const proxy = proxyChannel.getAPI();
+5
-30
src/common/worker.js
+5
-30
src/common/worker.js
···
1
-
import { RPCChannel, transfer } from "@kunkun/kkrpc";
2
1
import { getTransferables } from "@okikio/transferables";
3
2
import { debounceMicrotask } from "@vicary/debounce-microtask";
4
3
import { xxh32 } from "xxh32";
5
4
6
-
import { BrowserPostMessageIo } from "./worker/rpc.js";
5
+
import { RpcChannel } from "./worker/rpc-channel.js";
7
6
8
7
export { getTransferables } from "@okikio/transferables";
9
-
export { transfer } from "@kunkun/kkrpc";
10
8
11
9
/**
12
-
* @import {Track} from "@definitions/types.d.ts"
13
10
* @import {Announcement, MessengerRealm, ProxiedActions, Tunnel} from "./worker.d.ts"
14
11
*/
15
12
···
54
51
}
55
52
56
53
/**
57
-
* @param {Uint8Array} data
58
-
* @returns {Track[]}
59
-
*/
60
-
export function tracksIn(data) {
61
-
return JSON.parse(new TextDecoder().decode(data));
62
-
}
63
-
64
-
/**
65
-
* @param {Track[]} tracks
66
-
*/
67
-
export function tracksOut(tracks) {
68
-
const buffer = new TextEncoder().encode(JSON.stringify(tracks));
69
-
return transfer(buffer, [buffer]);
70
-
}
71
-
72
-
/**
73
54
* @param {Worker | SharedWorker} worker
74
55
*/
75
56
export function workerLink(worker) {
···
93
74
/** @returns {ProxiedActions<Actions>} */
94
75
function ensureAPI() {
95
76
if (!int_api) {
96
-
const io = new BrowserPostMessageIo(workerLinkCreator);
97
-
98
-
/** @type {undefined | RPCChannel<{}, ProxiedActions<Actions>>} */
99
-
const rpc = new RPCChannel(io, { enableTransfer: false });
100
-
101
-
int_api = rpc.getAPI();
77
+
/** @type {RpcChannel<Actions, Actions>} */
78
+
const channel = new RpcChannel(workerLinkCreator());
79
+
int_api = channel.getAPI();
102
80
}
103
81
104
82
return int_api;
···
213
191
* @param {Actions} actions
214
192
*/
215
193
export function rpc(context, actions) {
216
-
const io = new BrowserPostMessageIo(() => context);
217
-
218
-
/** @type {undefined | RPCChannel<Actions, {}>} */
219
-
return new RPCChannel(io, { enableTransfer: false, expose: actions });
194
+
return new RpcChannel(context, { expose: actions });
220
195
}
221
196
222
197
////////////////////////////////////////////
+170
src/common/worker/rpc-channel.js
+170
src/common/worker/rpc-channel.js
···
1
+
/**
2
+
* @import {MessengerRealm, ProxiedActions} from "../worker.d.ts"
3
+
*/
4
+
5
+
/**
6
+
* Lightweight RPC channel over postMessage using structured clone.
7
+
*
8
+
* Protocol:
9
+
* - Request: { __rpc: true, id, method, args, type: "request" }
10
+
* - Response: { __rpc: true, id, type: "response", result } or { ..., error }
11
+
*
12
+
* @template {Record<string, (...args: any[]) => any>} LocalAPI
13
+
* @template {Record<string, (...args: any[]) => any>} RemoteAPI
14
+
*/
15
+
export class RpcChannel {
16
+
/** @type {LocalAPI | undefined} */
17
+
#actions;
18
+
19
+
/** @type {Record<string, { resolve: (v: any) => void, reject: (e: any) => void }>} */
20
+
#pending = {};
21
+
22
+
#port;
23
+
24
+
/** @type {(event: MessageEvent) => void} */
25
+
#listener;
26
+
27
+
/**
28
+
* @param {MessagePort | Worker | MessengerRealm} port
29
+
* @param {{ expose?: LocalAPI }} [options]
30
+
*/
31
+
constructor(port, options) {
32
+
this.#port = port;
33
+
this.#actions = options?.expose;
34
+
35
+
this.#listener = (/** @type {MessageEvent} */ event) => {
36
+
const msg = event.data;
37
+
if (!msg || msg.__rpc !== true) return;
38
+
39
+
if (msg.type === "request") {
40
+
this.#handleRequest(msg);
41
+
} else if (msg.type === "response") {
42
+
this.#handleResponse(msg);
43
+
}
44
+
};
45
+
46
+
port.addEventListener(
47
+
"message",
48
+
/** @type {EventListener} */ (this.#listener),
49
+
);
50
+
}
51
+
52
+
/**
53
+
* @param {LocalAPI} actions
54
+
*/
55
+
expose(actions) {
56
+
this.#actions = actions;
57
+
}
58
+
59
+
/**
60
+
* @template {keyof RemoteAPI & string} M
61
+
* @param {M} method
62
+
* @param {any[]} args
63
+
* @returns {Promise<any>}
64
+
*/
65
+
callMethod(method, args) {
66
+
return new Promise((resolve, reject) => {
67
+
const id = crypto.randomUUID();
68
+
this.#pending[id] = { resolve, reject };
69
+
this.#port.postMessage({
70
+
__rpc: true,
71
+
id,
72
+
method,
73
+
args,
74
+
type: "request",
75
+
});
76
+
});
77
+
}
78
+
79
+
/**
80
+
* @returns {ProxiedActions<RemoteAPI>}
81
+
*/
82
+
getAPI() {
83
+
return /** @type {ProxiedActions<RemoteAPI>} */ (
84
+
new Proxy(/** @type {any} */ ({}), {
85
+
get: (_target, /** @type {string} */ prop) => {
86
+
return (/** @type {any[]} */ ...args) => this.callMethod(prop, args);
87
+
},
88
+
})
89
+
);
90
+
}
91
+
92
+
destroy() {
93
+
this.#port.removeEventListener(
94
+
"message",
95
+
/** @type {EventListener} */ (this.#listener),
96
+
);
97
+
98
+
if ("close" in this.#port && typeof this.#port.close === "function") {
99
+
this.#port.close();
100
+
}
101
+
102
+
if (
103
+
"terminate" in this.#port && typeof this.#port.terminate === "function"
104
+
) {
105
+
/** @type {any} */ (this.#port).terminate();
106
+
}
107
+
108
+
// Reject all pending requests
109
+
for (const [id, { reject }] of Object.entries(this.#pending)) {
110
+
reject(new Error("RPC channel destroyed"));
111
+
delete this.#pending[id];
112
+
}
113
+
}
114
+
115
+
/**
116
+
* @param {{ id: string, method: string, args: any[] }} msg
117
+
*/
118
+
#handleRequest(msg) {
119
+
const { id, method, args } = msg;
120
+
const fn = this.#actions?.[/** @type {keyof LocalAPI} */ (method)];
121
+
122
+
if (typeof fn !== "function") {
123
+
this.#port.postMessage({
124
+
__rpc: true,
125
+
id,
126
+
type: "response",
127
+
error: `Method "${method}" not found`,
128
+
});
129
+
return;
130
+
}
131
+
132
+
try {
133
+
Promise.resolve(fn(...args)).then(
134
+
(result) => {
135
+
this.#port.postMessage({ __rpc: true, id, type: "response", result });
136
+
},
137
+
(err) => {
138
+
this.#port.postMessage({
139
+
__rpc: true,
140
+
id,
141
+
type: "response",
142
+
error: err?.message ?? String(err),
143
+
});
144
+
},
145
+
);
146
+
} catch (err) {
147
+
this.#port.postMessage({
148
+
__rpc: true,
149
+
id,
150
+
type: "response",
151
+
error: /** @type {Error} */ (err)?.message ?? String(err),
152
+
});
153
+
}
154
+
}
155
+
156
+
/**
157
+
* @param {{ id: string, result?: any, error?: string }} msg
158
+
*/
159
+
#handleResponse(msg) {
160
+
const pending = this.#pending[msg.id];
161
+
if (!pending) return;
162
+
delete this.#pending[msg.id];
163
+
164
+
if (msg.error !== undefined) {
165
+
pending.reject(new Error(msg.error));
166
+
} else {
167
+
pending.resolve(msg.result);
168
+
}
169
+
}
170
+
}
-144
src/common/worker/rpc.js
-144
src/common/worker/rpc.js
···
1
-
/**
2
-
* @import { IoCapabilities, IoInterface, IoMessage, WireEnvelope } from "@kunkun/kkrpc"
3
-
*
4
-
* @import { MessengerRealm } from "../worker.d.ts"
5
-
*/
6
-
7
-
const DESTROY_SIGNAL = "__DESTROY__";
8
-
9
-
/**
10
-
* @implements {IoInterface}
11
-
*/
12
-
export class BrowserPostMessageIo {
13
-
name = "browser-postmessage-io";
14
-
15
-
/** @type {Array<string | IoMessage>} */
16
-
#messageQueue = [];
17
-
18
-
/** @type {((value: string | IoMessage | null) => void) | null} */
19
-
#resolveRead = null;
20
-
21
-
/** */
22
-
#realm;
23
-
24
-
/** @type {IoCapabilities} */
25
-
capabilities = {
26
-
structuredClone: true,
27
-
transfer: true,
28
-
};
29
-
30
-
/**
31
-
* @param {() => MessengerRealm} realmCreator
32
-
*/
33
-
constructor(realmCreator) {
34
-
/** @type {undefined | MessengerRealm} */
35
-
const realm = realmCreator();
36
-
realm.addEventListener("message", this.#handleMessage.bind(this));
37
-
38
-
this.#realm = () => {
39
-
return realm;
40
-
};
41
-
}
42
-
43
-
/**
44
-
* @param {MessageEvent} event
45
-
*/
46
-
#handleMessage(event) {
47
-
const raw = event.data;
48
-
const message = this.#normalizeIncoming(raw);
49
-
50
-
// Handle destroy signal
51
-
if (message === DESTROY_SIGNAL) {
52
-
this.destroy();
53
-
return;
54
-
}
55
-
56
-
if (this.#resolveRead) {
57
-
this.#resolveRead(message);
58
-
this.#resolveRead = null;
59
-
} else {
60
-
this.#messageQueue.push(message);
61
-
}
62
-
}
63
-
64
-
/**
65
-
* @param {any} message
66
-
* @returns {string | IoMessage}
67
-
*/
68
-
#normalizeIncoming(message) {
69
-
if (typeof message === "string") {
70
-
return message;
71
-
}
72
-
73
-
if (message && typeof message === "object" && message.version === 2) {
74
-
const envelope = /** @type {WireEnvelope} */ (message);
75
-
return {
76
-
data: envelope,
77
-
transfers: (/** @type {unknown[] | undefined} */ (envelope
78
-
.__transferredValues)) ?? [],
79
-
};
80
-
}
81
-
82
-
return /** @type {string} */ (message);
83
-
}
84
-
85
-
/** @returns {Promise<string | IoMessage | null>} */
86
-
read() {
87
-
// If there are queued messages, return the first one
88
-
if (this.#messageQueue.length > 0) {
89
-
return Promise.resolve(this.#messageQueue.shift() ?? null);
90
-
}
91
-
92
-
// Otherwise, wait for the next message
93
-
return new Promise((resolve) => {
94
-
this.#resolveRead = resolve;
95
-
});
96
-
}
97
-
98
-
/**
99
-
* @param {string | IoMessage} message
100
-
*/
101
-
write(message) {
102
-
if (typeof message === "string") {
103
-
this.#realm().postMessage(message);
104
-
return Promise.resolve();
105
-
}
106
-
107
-
if (message.transfers && message.transfers.length > 0) {
108
-
const msg = { ...message };
109
-
110
-
if (typeof msg.data === "object" && msg.data.payload.args) {
111
-
if (msg.data.payload.args[0] instanceof HTMLElement) {
112
-
msg.data.payload.args[0] = undefined;
113
-
}
114
-
}
115
-
116
-
console.log(message)
117
-
118
-
this.#realm().postMessage(
119
-
message.data,
120
-
/** @type {Transferable[]} */ (message.transfers),
121
-
);
122
-
} else {
123
-
this.#realm().postMessage(message.data);
124
-
}
125
-
126
-
return Promise.resolve();
127
-
}
128
-
129
-
destroy() {
130
-
const realm = this.#realm();
131
-
132
-
realm.postMessage(DESTROY_SIGNAL);
133
-
134
-
if (
135
-
"terminate" in realm && typeof realm.terminate === "function"
136
-
) {
137
-
realm.terminate();
138
-
}
139
-
}
140
-
141
-
signalDestroy() {
142
-
this.#realm().postMessage(DESTROY_SIGNAL);
143
-
}
144
-
}