forked from
tokono.ma/diffuse
A music player that connects to your cloud/distributed storage.
chore: refactoring + cleanup : worker tunnel
+56
-92
-27
src/common/worker.d.ts
-27
src/common/worker.d.ts
···
9
9
export type MRpcBaseMsg = { ns: string; name: string; key: number };
10
10
11
11
/** */
12
-
export type PortProvider = () => {
13
-
disconnect(): void;
14
-
port: MessagePort;
15
-
};
16
-
17
-
/** */
18
-
export type PortProviderMethod = { port: PortProvider };
19
-
20
-
/** */
21
12
export type ProxiedActions<
22
13
Actions extends Record<string, (...args: any[]) => any>,
23
14
> = {
···
32
23
> = (...args: Parameters<Action>) => PromisedReturn;
33
24
34
25
/** */
35
-
export type ProxyProvider<
36
-
Actions extends Record<string, (...args: any[]) => any>,
37
-
> = (
38
-
workerLinkCreator: () => MessagePort | Worker,
39
-
) => ProxiedActions<Actions>;
40
-
41
-
/** */
42
-
export type ProxyProviderMethod<
43
-
Actions extends Record<string, (...args: any[]) => any>,
44
-
> = { proxy: ProxyProvider<Actions> };
45
-
46
-
/** */
47
26
export interface MessengerRealm {
48
27
postMessage: MessagePort["postMessage"];
49
28
addEventListener: MessagePort["addEventListener"];
50
29
removeEventListener: MessagePort["removeEventListener"];
51
30
}
52
-
53
-
/** */
54
-
export type WorkerProvider = (group?: string) => Worker | SharedWorker;
55
-
56
-
/** */
57
-
export type WorkerProviderMethod = { worker: WorkerProvider };
+33
-36
src/common/worker.js
+33
-36
src/common/worker.js
···
52
52
}
53
53
54
54
/**
55
-
* @param {() => MessagePort | Worker} workerLinkCreator
56
-
*/
57
-
export function portProvider(workerLinkCreator) {
58
-
return () => {
59
-
const channel = new MessageChannel();
60
-
const workerOrPort = workerLinkCreator();
61
-
62
-
channel.port1.addEventListener("message", (event) => {
63
-
workerOrPort.postMessage(event.data);
64
-
});
65
-
66
-
/**
67
-
* @param {Event} event
68
-
*/
69
-
const workerListener = (event) => {
70
-
const msgEvent = /** @type {MessageEvent} */ (event);
71
-
channel.port1.postMessage(msgEvent.data);
72
-
};
73
-
74
-
workerOrPort.addEventListener("message", workerListener);
75
-
76
-
channel.port1.start();
77
-
channel.port2.start();
78
-
79
-
return {
80
-
disconnect: () => {
81
-
workerOrPort.removeEventListener("message", workerListener);
82
-
channel.port1.close();
83
-
channel.port2.close();
84
-
},
85
-
port: channel.port2,
86
-
};
87
-
};
88
-
}
89
-
90
-
/**
91
55
* @param {Worker | SharedWorker} worker
92
56
*/
93
57
export function workerLink(worker) {
···
97
61
} else {
98
62
return worker;
99
63
}
64
+
}
65
+
66
+
/**
67
+
* @param {MessagePort | Worker} workerLink
68
+
*/
69
+
export function workerTunnel(workerLink) {
70
+
const channel = new MessageChannel();
71
+
72
+
channel.port1.addEventListener("message", (event) => {
73
+
workerLink.postMessage(event.data);
74
+
});
75
+
76
+
/**
77
+
* @param {Event} event
78
+
*/
79
+
const workerListener = (event) => {
80
+
const msgEvent = /** @type {MessageEvent} */ (event);
81
+
channel.port1.postMessage(msgEvent.data);
82
+
};
83
+
84
+
workerLink.addEventListener("message", workerListener);
85
+
86
+
channel.port1.start();
87
+
channel.port2.start();
88
+
89
+
return {
90
+
disconnect: () => {
91
+
workerLink.removeEventListener("message", workerListener);
92
+
channel.port1.close();
93
+
channel.port2.close();
94
+
},
95
+
port: channel.port2,
96
+
};
100
97
}
101
98
102
99
////////////////////////////////////////////
+2
-2
src/components/input/opensubsonic/element.js
+2
-2
src/components/input/opensubsonic/element.js
···
1
1
import { DiffuseElement } from "@common/element.js";
2
-
import { portProvider, workerProxy } from "@common/worker.js";
2
+
import { workerProxy } from "@common/worker.js";
3
3
4
4
/**
5
5
* @import {InputActions} from "@components/input/types.d.ts"
6
-
* @import {PortProviderMethod, ProxiedActions, ProxyProvider, ProxyProviderMethod} from "@common/worker.d.ts"
6
+
* @import {ProxiedActions} from "@common/worker.d.ts"
7
7
*/
8
8
9
9
////////////////////////////////////////////
+2
-2
src/components/input/s3/element.js
+2
-2
src/components/input/s3/element.js
···
1
1
import { DiffuseElement } from "@common/element.js";
2
-
import { portProvider, workerProxy } from "@common/worker.js";
2
+
import { workerProxy } from "@common/worker.js";
3
3
4
4
/**
5
5
* @import {InputActions} from "@components/input/types.d.ts"
6
-
* @import {PortProviderMethod, ProxiedActions, ProxyProvider, ProxyProviderMethod} from "@common/worker.d.ts"
6
+
* @import {ProxiedActions} from "@common/worker.d.ts"
7
7
*/
8
8
9
9
////////////////////////////////////////////
+2
-6
src/components/input/types.d.ts
+2
-6
src/components/input/types.d.ts
···
1
-
import type {
2
-
ProxiedActions,
3
-
ProxyProviderMethod,
4
-
WorkerProviderMethod,
5
-
} from "@common/worker.d.ts";
1
+
import type { ProxiedActions } from "@common/worker.d.ts";
6
2
7
3
import type { Track } from "@definitions/types.d.ts";
8
-
import { DiffuseElement } from "@common/element.js";
4
+
import type { DiffuseElement } from "@common/element.js";
9
5
10
6
/**
11
7
* Consultation.
+13
-15
src/components/orchestrator/process-tracks/element.js
+13
-15
src/components/orchestrator/process-tracks/element.js
···
1
1
import { DiffuseElement, query } from "@common/element.js";
2
2
import { signal, untracked } from "@common/signal.js";
3
3
import {
4
-
portProvider,
5
4
transfer,
6
5
workerLink,
7
6
workerProxy,
7
+
workerTunnel,
8
8
} from "@common/worker.js";
9
9
10
10
/**
···
26
26
* from the assigned output element.
27
27
*/
28
28
class ProcessTracksOrchestrator extends DiffuseElement {
29
+
#process;
30
+
29
31
/** @type {Promise<{ input: Worker | SharedWorker; metadataProcessor: Worker | SharedWorker } | undefined>} */
30
-
#external = Promise.resolve(undefined);
31
-
#process;
32
+
#workers = Promise.resolve(undefined);
32
33
33
34
static NAME = "diffuse/orchestrator/process-tracks";
34
35
static WORKER_URL = "components/orchestrator/process-tracks/worker.js";
···
68
69
this.metadataProcessor = query(this, "metadata-processor-selector");
69
70
70
71
// Create new workers specially for track processing
71
-
this.#external = Promise.all([
72
+
this.#workers = Promise.all([
72
73
customElements.whenDefined(this.input.localName),
73
74
customElements.whenDefined(this.metadataProcessor.localName),
74
75
]).then(() => {
···
101
102
async disconnectedCallback() {
102
103
super.disconnectedCallback();
103
104
104
-
const ext = await this.#external;
105
-
if (!ext) return;
105
+
const workers = await this.#workers;
106
106
107
-
if (ext.input instanceof Worker) ext.input.terminate();
108
-
if (ext.metadataProcessor instanceof Worker) {
109
-
ext.metadataProcessor.terminate();
107
+
if (workers?.input instanceof Worker) workers.input.terminate();
108
+
if (workers?.metadataProcessor instanceof Worker) {
109
+
workers.metadataProcessor.terminate();
110
110
}
111
111
}
112
112
113
113
// ACTIONS
114
114
115
115
async process() {
116
-
const ext = await this.#external;
116
+
const workers = await this.#workers;
117
117
118
-
if (!ext) return;
118
+
if (!workers) return;
119
119
if (!this.output) return;
120
120
121
121
// Start
···
126
126
127
127
// Establish channel between external workers and our processing worker
128
128
const ports = {
129
-
input: portProvider(() => workerLink(ext.input))(),
130
-
metadataProcessor: portProvider(() =>
131
-
workerLink(ext.metadataProcessor)
132
-
)(),
129
+
input: workerTunnel(workerLink(workers.input)),
130
+
metadataProcessor: workerTunnel(workerLink(workers.metadataProcessor)),
133
131
};
134
132
135
133
// Send everything to worker
+1
-1
src/components/output/polymorphic/indexed-db/element.js
+1
-1
src/components/output/polymorphic/indexed-db/element.js
+1
-1
src/components/processor/artwork/element.js
+1
-1
src/components/processor/artwork/element.js
+1
-1
src/components/processor/metadata/element.js
+1
-1
src/components/processor/metadata/element.js
···
2
2
import { workerProxy } from "@common/worker.js";
3
3
4
4
/**
5
-
* @import {PortProviderMethod, ProxiedActions, ProxyProvider, ProxyProviderMethod, WorkerProviderMethod} from "@common/worker.d.ts"
5
+
* @import {ProxiedActions} from "@common/worker.d.ts"
6
6
* @import {Actions} from "./types.d.ts"
7
7
*/
8
8
+1
-1
src/components/processor/search/element.js
+1
-1
src/components/processor/search/element.js