forked from
tokono.ma/diffuse
A music player that connects to your cloud/distributed storage.
chore: all work must be performed in workers
+376
-127
+4
-3
src/common/element.d.ts
+4
-3
src/common/element.d.ts
···
2
2
| { leader: true; initialLeader: boolean }
3
3
| { leader: false };
4
4
5
-
export type FnParams<Fn> = Fn extends (...args: infer P) => any ? P : never;
6
-
export type FnReturn<Fn> = Fn extends (...args: any[]) => infer P ? P : never;
7
-
8
5
export type HtmlTagFunction = (
9
6
strings: string[] | ArrayLike<string>,
10
7
...values: unknown[]
11
8
) => string;
9
+
10
+
export type ProvisionedWorkers<T extends string> = {
11
+
[K in T]: Worker | SharedWorker;
12
+
};
12
13
13
14
export type RenderArg<State = undefined> = {
14
15
html: HtmlTagFunction;
+105
-17
src/common/element.js
+105
-17
src/common/element.js
···
2
2
import { html, render } from "lit-html";
3
3
4
4
import { effect, signal } from "@common/signal.js";
5
-
import { rpc, workerLink, workerProxy } from "./worker.js";
5
+
import {
6
+
getTransferables,
7
+
rpc,
8
+
transfer,
9
+
workerLink,
10
+
workerTunnel,
11
+
} from "./worker.js";
6
12
import { BrowserPostMessageIo } from "./worker/rpc.js";
7
13
import { RPCChannel } from "@kunkun/kkrpc";
8
14
9
15
/**
10
-
* @import {BroadcastingStatus, FnParams, FnReturn} from "./element.d.ts"
11
-
* @import {ProxiedActions} from "./worker.d.ts";
16
+
* @import {BroadcastingStatus, FnParams, FnReturn, ProvisionedWorkers} from "./element.d.ts"
17
+
* @import {ProxiedActions, Tunnel} from "./worker.d.ts";
12
18
* @import {Signal} from "./signal.d.ts"
13
19
*/
14
20
···
92
98
/** @type {undefined | Worker | SharedWorker} */
93
99
#worker;
94
100
95
-
worker() {
101
+
createWorker() {
96
102
const NAME = this.constructor.prototype.constructor.NAME;
97
103
const WORKER_URL = this.constructor.prototype.constructor.WORKER_URL;
98
104
···
121
127
return worker;
122
128
}
123
129
124
-
/**
125
-
* @param {Worker | SharedWorker} [worker]
126
-
*/
127
-
workerLink(worker) {
128
-
let w;
129
-
130
-
if (worker) {
131
-
w = worker;
132
-
} else {
133
-
this.#worker ??= this.worker();
134
-
w = this.#worker;
135
-
}
130
+
worker() {
131
+
this.#worker ??= this.createWorker();
132
+
return this.#worker;
133
+
}
136
134
137
-
return workerLink(w);
135
+
workerLink() {
136
+
const worker = this.worker();
137
+
return workerLink(worker);
138
138
}
139
139
}
140
140
···
326
326
}
327
327
328
328
/**
329
+
* @template {string} A
330
+
* @template {ProvisionedWorkers<A>} B
331
+
* @template {Record<string, any>} C
332
+
* @template R
333
+
* @param {Promise<B> | undefined} provisions
334
+
* @param {(args: C & { ports: { [K in keyof B]: MessagePort } }) => R} fn
335
+
* @param {C} fnArgs
336
+
* @returns {Promise<R>}
337
+
*/
338
+
export async function callWorkerWithProvisions(provisions, fn, fnArgs) {
339
+
const workers = await provisions;
340
+
if (!workers) throw new Error("Workers not defined");
341
+
342
+
/** @type {Array<[keyof B, Tunnel]>} */
343
+
const tunnels = Object.keys(workers).map(
344
+
(value) => {
345
+
const key = /** @type {keyof B} */ (value);
346
+
const worker = workers[key];
347
+
return [key, workerTunnel(worker)];
348
+
},
349
+
);
350
+
351
+
const ports = /** @type {{ [K in keyof B]: MessagePort }} */ (
352
+
Object.fromEntries(
353
+
tunnels.map(([key, tunnel]) => {
354
+
return [key, tunnel.port];
355
+
}),
356
+
)
357
+
);
358
+
359
+
const args = {
360
+
...fnArgs,
361
+
ports,
362
+
};
363
+
364
+
const result = await fn(transfer(
365
+
args,
366
+
tunnels.map(([_key, tunnel]) => {
367
+
return tunnel.port;
368
+
}),
369
+
));
370
+
371
+
tunnels.forEach(([_key, tunnel]) => {
372
+
tunnel.disconnect();
373
+
});
374
+
375
+
return result;
376
+
}
377
+
378
+
/**
329
379
* Component DOM selector.
330
380
*
331
381
* Basically `document.querySelector` but returns the element
···
370
420
371
421
return element;
372
422
}
423
+
424
+
/**
425
+
* @template {Record<string, DiffuseElement>} T
426
+
* @param {T} elements
427
+
* @returns {Promise<{ [K in keyof T]: Worker | SharedWorker }>}
428
+
*/
429
+
export async function provisionWorkers(elements) {
430
+
await whenElementsDefined(elements);
431
+
432
+
const entries = Object.entries(elements).map(([key, element]) => {
433
+
return [key, element.createWorker()];
434
+
});
435
+
436
+
return Object.fromEntries(entries);
437
+
}
438
+
439
+
/**
440
+
* @param {ProvisionedWorkers<any> | undefined} workers
441
+
*/
442
+
export function terminateWorkers(workers) {
443
+
if (!workers) return;
444
+
445
+
Object.values(workers).forEach((worker) => {
446
+
if (worker instanceof Worker) worker.terminate();
447
+
});
448
+
}
449
+
450
+
/**
451
+
* @template {Record<string, DiffuseElement>} T
452
+
* @param {T} elements
453
+
*/
454
+
export async function whenElementsDefined(elements) {
455
+
await Promise.all(
456
+
Object.values(elements).map((element) =>
457
+
customElements.whenDefined(element.localName)
458
+
),
459
+
);
460
+
}
+6
src/common/worker.d.ts
+6
src/common/worker.d.ts
+11
-6
src/common/worker.js
+11
-6
src/common/worker.js
···
5
5
6
6
import { BrowserPostMessageIo } from "./worker/rpc.js";
7
7
8
+
export { getTransferables } from "@okikio/transferables";
8
9
export { transfer } from "@kunkun/kkrpc";
9
10
10
11
/**
11
-
* @import {Announcement, MessengerRealm, ProxiedActions} from "./worker.d.ts"
12
+
* @import {Announcement, MessengerRealm, ProxiedActions, Tunnel} from "./worker.d.ts"
12
13
*/
13
14
14
15
////////////////////////////////////////////
···
64
65
}
65
66
66
67
/**
67
-
* @param {MessagePort | Worker} workerLink
68
+
* @param {MessagePort | Worker | SharedWorker} workerOrLink
69
+
* @returns {Tunnel}
68
70
*/
69
-
export function workerTunnel(workerLink) {
71
+
export function workerTunnel(workerOrLink) {
72
+
const link = workerOrLink instanceof SharedWorker
73
+
? workerLink(workerOrLink)
74
+
: workerOrLink;
70
75
const channel = new MessageChannel();
71
76
72
77
channel.port1.addEventListener("message", (event) => {
73
-
workerLink.postMessage(event.data);
78
+
link.postMessage(event.data);
74
79
});
75
80
76
81
/**
···
81
86
channel.port1.postMessage(msgEvent.data);
82
87
};
83
88
84
-
workerLink.addEventListener("message", workerListener);
89
+
link.addEventListener("message", workerListener);
85
90
86
91
channel.port1.start();
87
92
channel.port2.start();
88
93
89
94
return {
90
95
disconnect: () => {
91
-
workerLink.removeEventListener("message", workerListener);
96
+
link.removeEventListener("message", workerListener);
92
97
channel.port1.close();
93
98
channel.port2.close();
94
99
},
+1
-1
src/components/engine/queue/element.js
+1
-1
src/components/engine/queue/element.js
+23
-53
src/components/orchestrator/process-tracks/element.js
+23
-53
src/components/orchestrator/process-tracks/element.js
···
1
-
import { DiffuseElement, query } from "@common/element.js";
1
+
import {
2
+
callWorkerWithProvisions,
3
+
DiffuseElement,
4
+
provisionWorkers,
5
+
query,
6
+
terminateWorkers,
7
+
} from "@common/element.js";
2
8
import { signal, untracked } from "@common/signal.js";
3
9
import {
4
10
transfer,
···
9
15
10
16
/**
11
17
* @import {Track} from "@definitions/types.d.ts"
18
+
* @import {ProvisionedWorkers} from "@common/element.d.ts"
12
19
* @import {ProxiedActions} from "@common/worker.d.ts"
13
20
* @import {InputElement} from "@components/input/types.d.ts"
14
21
* @import {OutputElement} from "@components/output/types.d.ts"
···
26
33
* from the assigned output element.
27
34
*/
28
35
class ProcessTracksOrchestrator extends DiffuseElement {
29
-
#process;
30
-
31
-
/** @type {Promise<{ input: Worker | SharedWorker; metadataProcessor: Worker | SharedWorker }> | undefined} */
32
-
#workers = undefined;
33
-
34
36
static NAME = "diffuse/orchestrator/process-tracks";
35
37
static WORKER_URL = "components/orchestrator/process-tracks/worker.js";
36
38
37
-
constructor() {
38
-
super();
39
+
/** @type {ProxiedActions<Actions>} */
40
+
#proxy;
39
41
40
-
/** @type {ProxiedActions<Actions>} */
41
-
const p = workerProxy(this.workerLink);
42
+
/** @type {Promise<ProvisionedWorkers<"input" | "metadataProcessor">> | undefined} */
43
+
#workers = undefined;
42
44
43
-
this.#process = p.process;
45
+
constructor() {
46
+
super();
47
+
this.#proxy = workerProxy(this.workerLink);
44
48
}
45
49
46
50
// SIGNALS
···
73
77
this.output = output;
74
78
this.metadataProcessor = metadataProcessor;
75
79
76
-
// Create new workers specially for track processing
77
-
this.#workers = Promise.all([
78
-
customElements.whenDefined(input.localName),
79
-
customElements.whenDefined(metadataProcessor.localName),
80
-
]).then(() => {
81
-
return {
82
-
input: input.worker(),
83
-
metadataProcessor: metadataProcessor.worker(),
84
-
};
85
-
});
80
+
// Create new workers
81
+
this.#workers = provisionWorkers({ input, metadataProcessor });
86
82
87
83
// Wait until defined
88
84
await customElements.whenDefined(output.localName);
···
101
97
*/
102
98
async disconnectedCallback() {
103
99
super.disconnectedCallback();
104
-
105
-
const workers = await this.#workers;
106
-
107
-
if (workers?.input instanceof Worker) workers.input.terminate();
108
-
if (workers?.metadataProcessor instanceof Worker) {
109
-
workers.metadataProcessor.terminate();
110
-
}
100
+
terminateWorkers(await this.#workers);
111
101
}
112
102
113
103
// ACTIONS
114
104
115
105
async process() {
116
-
const workers = await this.#workers;
117
-
118
-
if (!workers) return;
119
106
if (!this.output) return;
120
107
121
108
// Start
···
123
110
console.log("🪵 Processing initiated");
124
111
125
112
const cachedTracks = this.output.tracks.collection();
126
-
127
-
// Establish channel between external workers and our processing worker
128
-
const ports = {
129
-
input: workerTunnel(workerLink(workers.input)),
130
-
metadataProcessor: workerTunnel(workerLink(workers.metadataProcessor)),
131
-
};
132
-
133
-
// Send everything to worker
134
-
const result = await this.#process(transfer({
135
-
ports: {
136
-
input: ports.input.port,
137
-
metadataProcessor: ports.metadataProcessor.port,
138
-
},
139
-
tracks: cachedTracks,
140
-
}, [
141
-
ports.input.port,
142
-
ports.metadataProcessor.port,
143
-
]));
113
+
const result = await callWorkerWithProvisions(
114
+
this.#workers,
115
+
this.#proxy.process,
116
+
{ tracks: cachedTracks },
117
+
);
144
118
145
119
// Save if collection changed
146
120
if (result) await this.output.tracks.save(result);
147
-
148
-
// Close external channels
149
-
ports.input.disconnect();
150
-
ports.metadataProcessor.disconnect();
151
121
152
122
// Fin
153
123
console.log("🪵 Processing completed");
-2
src/components/orchestrator/process-tracks/types.d.ts
-2
src/components/orchestrator/process-tracks/types.d.ts
+55
-23
src/components/orchestrator/queue-tracks/element.js
+55
-23
src/components/orchestrator/queue-tracks/element.js
···
1
-
import { DiffuseElement, query } from "@common/element.js";
1
+
import {
2
+
callWorkerWithProvisions,
3
+
DiffuseElement,
4
+
provisionWorkers,
5
+
query,
6
+
terminateWorkers,
7
+
whenElementsDefined,
8
+
} from "@common/element.js";
2
9
import { untracked } from "@common/signal.js";
10
+
import { workerProxy } from "@common/worker.js";
3
11
4
12
/**
5
13
* @import {Track} from "@definitions/types.d.ts"
14
+
* @import {ProvisionedWorkers} from "@common/element.d.ts"
15
+
* @import {ProxiedActions} from "@common/worker.d.ts"
6
16
* @import {InputElement} from "@components/input/types.d.ts"
7
17
* @import {OutputElement} from "@components/output/types.d.ts"
18
+
*
19
+
* @import {Actions} from "./types.d.ts"
8
20
*/
9
21
10
22
////////////////////////////////////////////
···
17
29
* or the tracks collection changes.
18
30
*/
19
31
class QueueTracksOrchestrator extends DiffuseElement {
32
+
static NAME = "diffuse/orchestrator/queue-tracks";
33
+
static WORKER_URL = "components/orchestrator/queue-tracks/worker.js";
34
+
35
+
/** @type {ProxiedActions<Actions>} */
36
+
#proxy;
37
+
38
+
/** @type {Promise<ProvisionedWorkers<"input" | "queue">> | undefined} */
39
+
#workers = undefined;
40
+
41
+
constructor() {
42
+
super();
43
+
this.#proxy = workerProxy(this.workerLink);
44
+
}
45
+
20
46
/**
21
47
* @override
22
48
*/
···
24
50
super.connectedCallback();
25
51
26
52
/** @type {InputElement} */
27
-
this.input = query(this, "input-selector");
53
+
const input = query(this, "input-selector");
28
54
29
55
/** @type {OutputElement<Track[]>} */
30
-
this.output = query(this, "output-selector");
56
+
const output = query(this, "output-selector");
31
57
32
58
/** @type {import("@components/engine/queue/element.js").CLASS} */
33
-
this.queue = query(this, "queue-engine-selector");
59
+
const queue = query(this, "queue-engine-selector");
60
+
61
+
// Assign to self
62
+
this.input = input;
63
+
this.output = output;
64
+
this.queue = queue;
65
+
66
+
// Create new workers
67
+
this.#workers = whenElementsDefined({ input, queue }).then(() => ({
68
+
input: input.createWorker(),
69
+
queue: queue.worker(),
70
+
}));
34
71
35
72
// When defined
36
73
await customElements.whenDefined(this.input.localName);
···
38
75
39
76
// Watch tracks collection
40
77
this.effect(() => {
41
-
if (!this.output) return;
42
-
43
-
const tracks = this.output.tracks.collection().filter((t) =>
78
+
const tracks = output.tracks.collection().filter((t) =>
44
79
t.kind !== "placeholder"
45
80
);
46
81
···
48
83
});
49
84
}
50
85
86
+
/**
87
+
* @override
88
+
*/
89
+
async disconnectedCallback() {
90
+
super.disconnectedCallback();
91
+
terminateWorkers(await this.#workers);
92
+
}
93
+
51
94
// 🌊
52
95
53
96
/**
54
97
* @param {Track[]} cachedTracks
55
98
*/
56
99
async poolAvailable(cachedTracks) {
57
-
if (!this.input) return;
58
-
if (!this.output) return;
59
-
if (!this.queue) return;
60
-
61
-
const groups = await this.input.groupConsult(cachedTracks);
62
-
63
-
/** @type {Track[]} */
64
-
let availableTracks = [];
65
-
66
-
Object.values(groups).forEach((value) => {
67
-
if (value.available === false) return;
68
-
availableTracks = availableTracks.concat(value.tracks);
69
-
}, []);
70
-
71
-
// Set pool
72
-
await this.queue.pool(availableTracks);
100
+
return await callWorkerWithProvisions(
101
+
this.#workers,
102
+
this.#proxy.poolAvailable,
103
+
{ tracks: cachedTracks },
104
+
);
73
105
}
74
106
}
75
107
+8
src/components/orchestrator/queue-tracks/types.d.ts
+8
src/components/orchestrator/queue-tracks/types.d.ts
+52
src/components/orchestrator/queue-tracks/worker.js
+52
src/components/orchestrator/queue-tracks/worker.js
···
1
+
import { ostiary, rpc, workerProxy } from "@common/worker.js";
2
+
3
+
/**
4
+
* @import {Track} from "@definitions/types.d.ts"
5
+
* @import {ProxiedActions} from "@common/worker.d.ts"
6
+
* @import {InputActions} from "@components/input/types.d.ts"
7
+
* @import {Actions as QueueEngineActions} from "@components/engine/queue/types.d.ts"
8
+
* @import {Actions} from "./types.d.ts"
9
+
*/
10
+
11
+
////////////////////////////////////////////
12
+
// ACTIONS
13
+
////////////////////////////////////////////
14
+
15
+
/**
16
+
* @type {Actions["poolAvailable"]}
17
+
*/
18
+
export async function poolAvailable(args) {
19
+
const { ports } = args;
20
+
const cachedTracks = args.tracks;
21
+
22
+
/** @type {ProxiedActions<InputActions>} */
23
+
const input = workerProxy(() => ports.input);
24
+
25
+
/** @type {ProxiedActions<QueueEngineActions>} */
26
+
const queue = workerProxy(() => ports.queue);
27
+
28
+
ports.input.start();
29
+
ports.queue.start();
30
+
31
+
// Consult input
32
+
const groups = await input.groupConsult(cachedTracks);
33
+
34
+
/** @type {Track[]} */
35
+
let availableTracks = [];
36
+
37
+
Object.values(groups).forEach((value) => {
38
+
if (value.available === false) return;
39
+
availableTracks = availableTracks.concat(value.tracks);
40
+
}, []);
41
+
42
+
// Set pool
43
+
await queue.pool(availableTracks);
44
+
}
45
+
46
+
////////////////////////////////////////////
47
+
// ⚡️
48
+
////////////////////////////////////////////
49
+
50
+
ostiary((context) => {
51
+
rpc(context, { poolAvailable });
52
+
});
+51
-22
src/components/orchestrator/search-tracks/element.js
+51
-22
src/components/orchestrator/search-tracks/element.js
···
1
-
import { DiffuseElement, query } from "@common/element.js";
1
+
import {
2
+
callWorkerWithProvisions,
3
+
DiffuseElement,
4
+
provisionWorkers,
5
+
query,
6
+
terminateWorkers,
7
+
} from "@common/element.js";
8
+
import { transfer, workerProxy, workerTunnel } from "@common/worker.js";
2
9
3
10
/**
4
11
* @import {Track} from "@definitions/types.d.ts"
12
+
* @import {ProvisionedWorkers} from "@common/element.d.ts"
13
+
* @import {ProxiedActions} from "@common/worker.d.ts"
5
14
* @import {InputElement} from "@components/input/types.d.ts"
6
15
* @import {OutputElement} from "@components/output/types.d.ts"
16
+
*
17
+
* @import {Actions} from "./types.d.ts"
7
18
*/
8
19
9
20
////////////////////////////////////////////
···
16
27
* or the tracks collection changes.
17
28
*/
18
29
class SearchTracksOrchestrator extends DiffuseElement {
30
+
static NAME = "diffuse/orchestrator/search-tracks";
31
+
static WORKER_URL = "components/orchestrator/search-tracks/worker.js";
32
+
33
+
/** @type {ProxiedActions<Actions>} */
34
+
#proxy;
35
+
36
+
/** @type {Promise<ProvisionedWorkers<"input" | "search">> | undefined} */
37
+
#workers = undefined;
38
+
39
+
constructor() {
40
+
super();
41
+
this.#proxy = workerProxy(this.workerLink);
42
+
}
43
+
19
44
/**
20
45
* @override
21
46
*/
···
23
48
super.connectedCallback();
24
49
25
50
/** @type {InputElement} */
26
-
this.input = query(this, "input-selector");
51
+
const input = query(this, "input-selector");
27
52
28
53
/** @type {OutputElement<Track[]>} */
29
-
this.output = query(this, "output-selector");
54
+
const output = query(this, "output-selector");
30
55
31
56
/** @type {import("@components/processor/search/element.js").CLASS} */
32
-
this.search = query(this, "search-processor-selector");
57
+
const search = query(this, "search-processor-selector");
58
+
59
+
// Assign to self
60
+
this.input = input;
61
+
this.output = output;
62
+
this.search = search;
63
+
64
+
// Create new workers
65
+
this.#workers = provisionWorkers({ input, search });
33
66
34
67
// When defined
35
68
await customElements.whenDefined(this.output.localName);
36
69
37
70
// Watch tracks collection
38
71
this.effect(() => {
39
-
if (!this.output) return;
40
-
41
-
const tracks = this.output.tracks.collection().filter((t) =>
72
+
const tracks = output.tracks.collection().filter((t) =>
42
73
t.kind !== "placeholder"
43
74
);
44
75
···
46
77
});
47
78
}
48
79
80
+
/**
81
+
* @override
82
+
*/
83
+
async disconnectedCallback() {
84
+
super.disconnectedCallback();
85
+
terminateWorkers(await this.#workers);
86
+
}
87
+
49
88
// 🚛
50
89
51
90
/**
52
91
* @param {Track[]} cachedTracks
53
92
*/
54
93
async supplyAvailable(cachedTracks) {
55
-
if (!this.input) return;
56
-
if (!this.search) return;
57
-
58
-
const groups = await this.input.groupConsult(cachedTracks);
59
-
60
-
/** @type {Track[]} */
61
-
let availableTracks = [];
62
-
63
-
Object.values(groups).forEach((value) => {
64
-
if (value.available === false) return;
65
-
availableTracks = availableTracks.concat(value.tracks);
66
-
}, []);
67
-
68
-
// Set pool
69
-
await this.search.supply(availableTracks);
94
+
return await callWorkerWithProvisions(
95
+
this.#workers,
96
+
this.#proxy.supplyAvailable,
97
+
{ tracks: cachedTracks },
98
+
);
70
99
}
71
100
}
72
101
+8
src/components/orchestrator/search-tracks/types.d.ts
+8
src/components/orchestrator/search-tracks/types.d.ts
+52
src/components/orchestrator/search-tracks/worker.js
+52
src/components/orchestrator/search-tracks/worker.js
···
1
+
import { ostiary, rpc, workerProxy } from "@common/worker.js";
2
+
3
+
/**
4
+
* @import {Track} from "@definitions/types.d.ts"
5
+
* @import {ProxiedActions} from "@common/worker.d.ts"
6
+
* @import {InputActions} from "@components/input/types.d.ts"
7
+
* @import {Actions as SearchProcessorActions} from "@components/processor/search/types.d.ts"
8
+
* @import {Actions} from "./types.d.ts"
9
+
*/
10
+
11
+
////////////////////////////////////////////
12
+
// ACTIONS
13
+
////////////////////////////////////////////
14
+
15
+
/**
16
+
* @type {Actions["supplyAvailable"]}
17
+
*/
18
+
export async function supplyAvailable(args) {
19
+
const { ports } = args;
20
+
const cachedTracks = args.tracks;
21
+
22
+
/** @type {ProxiedActions<InputActions>} */
23
+
const input = workerProxy(() => ports.input);
24
+
25
+
/** @type {ProxiedActions<SearchProcessorActions>} */
26
+
const search = workerProxy(() => ports.search);
27
+
28
+
ports.input.start();
29
+
ports.search.start();
30
+
31
+
// Consult input
32
+
const groups = await input.groupConsult(cachedTracks);
33
+
34
+
/** @type {Track[]} */
35
+
let availableTracks = [];
36
+
37
+
Object.values(groups).forEach((value) => {
38
+
if (value.available === false) return;
39
+
availableTracks = availableTracks.concat(value.tracks);
40
+
}, []);
41
+
42
+
// Set pool
43
+
await search.supply(availableTracks);
44
+
}
45
+
46
+
////////////////////////////////////////////
47
+
// ⚡️
48
+
////////////////////////////////////////////
49
+
50
+
ostiary((context) => {
51
+
rpc(context, { supplyAvailable });
52
+
});