forked from
tokono.ma/diffuse
A music player that connects to your cloud/distributed storage.
feat: BroadcastableDiffuseElement
+207
-114
+4
src/common/element.d.ts
+4
src/common/element.d.ts
+164
-3
src/common/element.js
+164
-3
src/common/element.js
···
1
1
import morphdom from "morphdom";
2
-
import { effect } from "@common/signal.js";
2
+
3
+
import { effect, unbiasedSignal } from "@common/signal.js";
4
+
import { define, use } from "@common/worker.js";
3
5
4
6
/**
5
-
* @import {HtmlTagFunction, MorphOptions} from "./element.d.ts"
7
+
* @import {BroadcastingStatus, HtmlTagFunction, MorphOptions} from "./element.d.ts"
8
+
* @import {Signal} from "./signal.d.ts"
6
9
*/
7
10
8
-
export default class DiffuseElement extends HTMLElement {
11
+
/**
12
+
* Base for custom elements, provides some utility functionality
13
+
* around rendering and managing signals.
14
+
*/
15
+
export class DiffuseElement extends HTMLElement {
9
16
#disposables = /** @type {Array<() => void>} */ ([]);
10
17
11
18
#teardown() {
···
92
99
this.#teardown();
93
100
}
94
101
}
102
+
103
+
/**
104
+
* Broadcastable version of the base class.
105
+
*
106
+
* Share the state of an element across multiple tabs
107
+
* of the same origin and have one instance be the leader.
108
+
*/
109
+
export class BroadcastableDiffuseElement extends DiffuseElement {
110
+
broadcasted = false;
111
+
112
+
/** @type {PromiseWithResolvers<void>} */
113
+
#lock = Promise.withResolvers();
114
+
115
+
/** @type {PromiseWithResolvers<BroadcastingStatus>} */
116
+
#status = Promise.withResolvers();
117
+
118
+
constructor() {
119
+
super();
120
+
121
+
this.broadcast = this.broadcast.bind(this);
122
+
123
+
/** @type {Signal<Promise<BroadcastingStatus>>} */
124
+
this.broadcastingStatus = unbiasedSignal(this.#status.promise);
125
+
}
126
+
127
+
/**
128
+
* @param {string} name
129
+
*/
130
+
broadcast(name) {
131
+
const channel = new BroadcastChannel(name);
132
+
const msg = new MessageChannel();
133
+
134
+
this.broadcasted = true;
135
+
this.name = name;
136
+
137
+
channel.addEventListener(
138
+
"message",
139
+
async (event) => {
140
+
const name = event.data.name?.split(":");
141
+
142
+
if (name[0] === "leader") {
143
+
const status = await this.#status.promise;
144
+
if (status.leader) {
145
+
msg.port1.postMessage({
146
+
...event.data,
147
+
name: name.splice(1).join(":"),
148
+
});
149
+
}
150
+
} else {
151
+
msg.port1.postMessage(event.data);
152
+
}
153
+
},
154
+
);
155
+
156
+
msg.port1.addEventListener(
157
+
"message",
158
+
(event) => channel.postMessage(event.data),
159
+
);
160
+
161
+
msg.port1.start();
162
+
msg.port2.start();
163
+
164
+
async function anyoneWaiting() {
165
+
const state = await navigator.locks.query();
166
+
return !!state.pending?.length;
167
+
}
168
+
169
+
/**
170
+
* @param {string} method
171
+
* @param {Function} fn
172
+
*/
173
+
return (method, fn) => {
174
+
define(method, fn.bind(this), msg.port2);
175
+
176
+
/** @param {any[]} args */
177
+
const leaderOnly = async (...args) => {
178
+
const status = await this.#status.promise;
179
+
return status.leader
180
+
? fn.call(this, ...args)
181
+
: use(`leader:${method}`, msg.port2)(...args);
182
+
};
183
+
184
+
/** @param {any[]} args */
185
+
const replicate = (...args) => {
186
+
anyoneWaiting().then((bool) => {
187
+
if (bool) use(method, msg.port2)(...args);
188
+
});
189
+
return fn.call(this, ...args);
190
+
};
191
+
192
+
return {
193
+
leaderOnly,
194
+
replicate,
195
+
};
196
+
};
197
+
}
198
+
199
+
// LIFECYCLE
200
+
201
+
/**
202
+
* @override
203
+
*/
204
+
connectedCallback() {
205
+
super.connectedCallback();
206
+
207
+
if (!this.broadcasted) return;
208
+
209
+
// Grab a lock if it isn't acquired yet,
210
+
// and hold it until `this.lock.promise` resolves.
211
+
navigator.locks.request(
212
+
`${this.name}/lock`,
213
+
{ ifAvailable: true },
214
+
(lock) => {
215
+
this.#status.resolve(
216
+
lock ? { leader: true, initialLeader: true } : { leader: false },
217
+
);
218
+
if (lock) return this.#lock.promise;
219
+
},
220
+
);
221
+
222
+
// When the lock status is initially determined, log its status.
223
+
// Additionally, wait for lock if needed.
224
+
this.#status.promise.then((status) => {
225
+
if (status.leader) {
226
+
console.log(`🧙 Elected leader for: ${this.name}`);
227
+
} else {
228
+
console.log(`🔮 Watching leader: ${this.name}`);
229
+
}
230
+
231
+
// Wait for leadership
232
+
if (status.leader === false) {
233
+
navigator.locks.request(
234
+
`${this.name}/lock`,
235
+
() => {
236
+
this.#status = Promise.withResolvers();
237
+
this.#status.resolve({ leader: true, initialLeader: false });
238
+
239
+
this.broadcastingStatus(this.#status.promise);
240
+
241
+
return this.#lock.promise;
242
+
},
243
+
);
244
+
}
245
+
});
246
+
}
247
+
248
+
/**
249
+
* @override
250
+
*/
251
+
disconnectedCallback() {
252
+
super.disconnectedCallback();
253
+
this.#lock.resolve();
254
+
}
255
+
}
-11
src/common/lock.js
-11
src/common/lock.js
+2
src/common/signal.js
+2
src/common/signal.js
+4
-2
src/common/worker.js
+4
-2
src/common/worker.js
···
4
4
import { xxh32 } from "xxh32";
5
5
6
6
/**
7
-
* @import {WorkerGlobalScope} from "@mys/m-rpc";
7
+
* @import {MRpcCallOptions, WorkerGlobalScope} from "@mys/m-rpc";
8
8
* @import {Announcement} from "./worker.d.ts"
9
9
*/
10
10
···
111
111
/**
112
112
* @param {string} name
113
113
* @param {MessagePort | Worker | WorkerGlobalScope} [context] Uses `globalThis` by default.
114
+
* @param {MRpcCallOptions} [options]
114
115
*/
115
116
export function use(
116
117
name,
117
118
context = /** @type {WorkerGlobalScope} */ (globalThis),
119
+
options,
118
120
) {
119
-
return useWorkerFn(name, /** @type {any} */ (context));
121
+
return useWorkerFn(name, /** @type {any} */ (context), options);
120
122
}
121
123
122
124
////////////////////////////////////////////
+1
-1
src/component/constituent/blur/browser-list/element.js
+1
-1
src/component/constituent/blur/browser-list/element.js
+31
-94
src/component/engine/audio/element.js
+31
-94
src/component/engine/audio/element.js
···
1
-
import DiffuseElement from "@common/element.js";
1
+
import {
2
+
BroadcastableDiffuseElement,
3
+
DiffuseElement,
4
+
} from "@common/element.js";
2
5
import { signal } from "@common/signal.js";
3
-
import { define, use } from "@common/worker.js";
4
-
import { lock } from "@common/lock.js";
5
6
6
7
/**
7
8
* @import {Actions, Audio, AudioState, Signals, State} from "./types.d.ts"
···
22
23
* @implements {Actions}
23
24
* @implements {Signals}
24
25
*/
25
-
class AudioEngine extends DiffuseElement {
26
+
class AudioEngine extends BroadcastableDiffuseElement {
26
27
// TODO:
27
28
// static observedAttributes = ["volume"];
28
29
···
35
36
36
37
// Setup leader election if shared
37
38
if (isShared) {
38
-
const name = `diffuse/engine/audio/${group}`;
39
-
40
-
const channel = new BroadcastChannel(name);
41
-
const msg = new MessageChannel();
42
-
43
-
channel.onmessage = (event) => msg.port1.postMessage(event.data);
44
-
msg.port1.addEventListener(
45
-
"message",
46
-
(event) => channel.postMessage(event.data),
47
-
);
48
-
49
-
msg.port1.start();
50
-
msg.port2.start();
39
+
const fn = this.broadcast(`diffuse/engine/audio/${group}`);
51
40
52
-
// Port 1 = Incoming, from channel.
53
-
// Port 2 = Outgoing, to channel.
41
+
this.pause = fn("pause", this.pause).leaderOnly;
42
+
this.play = fn("play", this.play).leaderOnly;
43
+
this.reload = fn("reload", this.reload).leaderOnly;
44
+
this.seek = fn("seek", this.seek).leaderOnly;
45
+
this.supply = fn("supply", this.supply).replicate;
54
46
55
-
this.lock = lock();
56
-
57
-
define("pause", this.#pause.bind(this), msg.port2);
58
-
define("play", this.#play.bind(this), msg.port2);
59
-
define("reload", this.#reload.bind(this), msg.port2);
60
-
define("seek", this.#seek.bind(this), msg.port2);
61
-
define("supply", this.#supply.bind(this), msg.port2);
62
-
63
-
/**
64
-
* @param {string} method
65
-
* @param {Function} fn
66
-
*/
67
-
const u = (method, fn) => {
68
-
/** @param {any[]} args */
69
-
return async (...args) => {
70
-
const status = await this.lock?.status.promise;
71
-
return status === "waiting"
72
-
? use(method, msg.port2)(...args)
73
-
: fn.call(this, ...args);
74
-
};
75
-
};
76
-
77
-
this.pause = u("pause", this.#pause);
78
-
this.play = u("play", this.#play);
79
-
this.reload = u("reload", this.#reload);
80
-
this.seek = u("seek", this.#seek);
81
-
this.supply = u("supply", this.#supply);
82
-
} else {
83
-
this.pause = this.#pause;
84
-
this.play = this.#play;
85
-
this.reload = this.#reload;
86
-
this.seek = this.#seek;
87
-
this.supply = this.#supply;
47
+
this.isPlaying = fn("isPlaying", this.isPlaying).replicate;
48
+
this.volume = fn("volume", this.volume).replicate;
88
49
}
89
50
90
51
// TODO: Get volume from previous session if possible
···
119
80
connectedCallback() {
120
81
super.connectedCallback();
121
82
83
+
// Manage playback across tabs if needed
84
+
if (this.broadcasted) {
85
+
this.effect(async () => {
86
+
const status = await this.broadcastingStatus();
87
+
if (status.leader && status.initialLeader === false) {
88
+
// TODO:
89
+
// console.log("🧙 Leadership acquired");
90
+
}
91
+
});
92
+
}
93
+
122
94
// Monitor volume
123
95
// NOTE: Support different volume levels for audio elements?
124
96
this.effect(() => {
···
130
102
},
131
103
);
132
104
});
133
-
134
-
// Setup leader election if shared
135
-
const isShared = this.hasAttribute("group");
136
-
const elementLock = this.lock;
137
-
138
-
if (isShared && elementLock) {
139
-
navigator.locks.request(
140
-
`${name}/lock`,
141
-
{ ifAvailable: true },
142
-
(lock) => {
143
-
elementLock.status.resolve(lock ? "acquired" : "waiting");
144
-
if (lock) return elementLock.promise;
145
-
},
146
-
);
147
-
148
-
elementLock.status.promise.then((status) => {
149
-
const name = `diffuse/engine/audio/${
150
-
this.getAttribute("group") || "main"
151
-
}`;
152
-
153
-
if (status === "acquired") {
154
-
console.log(`🧙 Elected leader for: ${name}`);
155
-
} else {
156
-
console.log(`🔮 Watching leader: ${name}`);
157
-
}
158
-
});
159
-
}
160
-
}
161
-
162
-
/**
163
-
* @override
164
-
*/
165
-
disconnectedCallback() {
166
-
super.disconnectedCallback();
167
-
if (this.lock) this.lock.resolve();
168
105
}
169
106
170
107
// ACTIONS (PRIVATE)
···
172
109
/**
173
110
* @type {Actions["pause"]}
174
111
*/
175
-
#pause({ audioId }) {
112
+
pause({ audioId }) {
176
113
this.withAudioNode(audioId, (audio) => audio.pause());
177
114
}
178
115
179
116
/**
180
117
* @type {Actions["play"]}
181
118
*/
182
-
#play({ audioId, volume }) {
119
+
play({ audioId, volume }) {
183
120
this.withAudioNode(audioId, (audio, item) => {
184
121
audio.volume = volume ?? this.state.volume();
185
122
audio.muted = false;
···
205
142
/**
206
143
* @type {Actions["reload"]}
207
144
*/
208
-
#reload(args) {
145
+
reload(args) {
209
146
this.withAudioNode(args.audioId, (audio, item) => {
210
147
if (audio.readyState === 0 || audio.error?.code === 2) {
211
148
audio.load();
···
218
155
}
219
156
220
157
if (args.play) {
221
-
this.#play({ audioId: args.audioId, volume: audio.volume });
158
+
this.play({ audioId: args.audioId, volume: audio.volume });
222
159
}
223
160
}
224
161
});
···
227
164
/**
228
165
* @type {Actions["seek"]}
229
166
*/
230
-
#seek({ audioId, percentage }) {
167
+
seek({ audioId, percentage }) {
231
168
this.withAudioNode(audioId, (audio) => {
232
169
if (!isNaN(audio.duration)) {
233
170
audio.currentTime = audio.duration * percentage;
···
238
175
/**
239
176
* @type {Actions["supply"]}
240
177
*/
241
-
#supply(args) {
178
+
supply(args) {
242
179
this.#items(args.audio);
243
-
if (args.play) this.#play(args.play);
180
+
if (args.play) this.play(args.play);
244
181
}
245
182
246
183
// RENDER
+1
-1
src/component/engine/queue/element.js
+1
-1
src/component/engine/queue/element.js
-2
src/theme/blur/index.vto
-2
src/theme/blur/index.vto