forked from
tokono.ma/diffuse
A music player that connects to your cloud/distributed storage.
feat: add support to audio engine for streaming
+241
-25
+202
-10
src/components/engine/audio/element.js
+202
-10
src/components/engine/audio/element.js
···
4
4
import { computed, signal, untracked } from "~/common/signal.js";
5
5
6
6
/**
7
-
* @import {Actions, Audio, AudioState, AudioStateReadOnly, LoadingState} from "./types.d.ts"
7
+
* @import {Actions, AudioUrl, AudioState, AudioStateReadOnly, LoadingState} from "./types.d.ts"
8
8
* @import {RenderArg} from "~/common/element.d.ts"
9
9
* @import {SignalReader} from "~/common/signal.d.ts"
10
10
*/
···
34
34
35
35
// SIGNALS
36
36
37
-
#items = signal(/** @type {Audio[]} */ ([]));
37
+
#items = signal(/** @type {AudioUrl[]} */ ([]));
38
38
#volume = signal(0.5);
39
+
40
+
/** @type {Map<string, ReadableStream>} Streams pending MediaSource setup */
41
+
#streams = new Map();
42
+
43
+
/** @type {Map<string, string>} MediaSource object URLs created from streams, keyed by item ID */
44
+
#mediaSourceUrls = new Map();
39
45
40
46
// STATE
41
47
···
232
238
supply(args) {
233
239
const existingMap = new Map(this.#items.value.map((a) => [a.id, a]));
234
240
235
-
const hasNewIds = args.audio.some((a) => !existingMap.has(a.id));
236
-
const hasPreloadChanges = args.audio.some(
241
+
// Start loading new streams
242
+
for (const item of args.audio) {
243
+
if (
244
+
"stream" in item &&
245
+
!existingMap.has(item.id) &&
246
+
!this.#streams.has(item.id)
247
+
) {
248
+
this.#streams.set(item.id, item.stream);
249
+
this.#resolveStream(
250
+
item.id,
251
+
item.stream,
252
+
item.mimeType ?? "",
253
+
item.seek,
254
+
item.duration,
255
+
);
256
+
}
257
+
}
258
+
259
+
// Stop streams that are no longer needed
260
+
const newIds = new Set(args.audio.map((a) => a.id));
261
+
262
+
for (const [id, objectUrl] of this.#mediaSourceUrls) {
263
+
if (!newIds.has(id)) {
264
+
URL.revokeObjectURL(objectUrl);
265
+
this.#mediaSourceUrls.delete(id);
266
+
}
267
+
}
268
+
269
+
for (const id of this.#streams.keys()) {
270
+
if (!newIds.has(id)) this.#streams.delete(id);
271
+
}
272
+
273
+
/** @type {AudioUrl[]} Remove `stream` field, replace it with `url` */
274
+
const resolvedAudio = args.audio.map((a) => {
275
+
const url = "stream" in a ? this.#mediaSourceUrls.get(a.id) : a.url;
276
+
277
+
if (!url) {
278
+
throw new Error("Stream did not produce a media source url");
279
+
}
280
+
281
+
return {
282
+
id: a.id,
283
+
isPreload: a.isPreload,
284
+
mimeType: a.mimeType,
285
+
progress: a.progress,
286
+
url,
287
+
};
288
+
});
289
+
290
+
const hasNewIds = resolvedAudio.some((a) => !existingMap.has(a.id));
291
+
const hasPreloadChanges = resolvedAudio.some(
237
292
(a) => existingMap.get(a.id)?.isPreload !== a.isPreload,
238
293
);
239
294
240
295
if (hasNewIds || hasPreloadChanges) {
241
-
this.#items.value = args.audio;
296
+
this.#items.value = resolvedAudio;
242
297
}
243
298
244
299
if (args.play) this.play(args.play);
245
300
}
246
301
302
+
// STREAMS
303
+
304
+
/**
305
+
* @param {string} id
306
+
* @param {ReadableStream} stream
307
+
* @param {string} mimeType
308
+
* @param {((timeSeconds: number) => Promise<ReadableStream>) | undefined} seekFn
309
+
* @param {number | undefined} duration
310
+
*/
311
+
async #resolveStream(id, stream, mimeType, seekFn, duration) {
312
+
const mediaSource = new MediaSource();
313
+
const objectUrl = URL.createObjectURL(mediaSource);
314
+
315
+
this.#mediaSourceUrls.set(id, objectUrl);
316
+
this.#streams.delete(id);
317
+
318
+
// Yield so the render triggered by supply() can complete, ensuring the
319
+
// audio element is in the DOM before we set its src.
320
+
await Promise.resolve();
321
+
322
+
if (!this.#mediaSourceUrls.has(id)) {
323
+
// Item was removed while waiting
324
+
URL.revokeObjectURL(objectUrl);
325
+
return;
326
+
}
327
+
328
+
const itemEl = this.#itemElement(id);
329
+
if (!itemEl) {
330
+
URL.revokeObjectURL(objectUrl);
331
+
this.#mediaSourceUrls.delete(id);
332
+
return;
333
+
}
334
+
335
+
// MediaSource must be attached via audio.src directly;
336
+
// <source> elements do not trigger sourceopen.
337
+
itemEl.audio.src = objectUrl;
338
+
339
+
await new Promise((resolve) => {
340
+
mediaSource.addEventListener("sourceopen", resolve, { once: true });
341
+
});
342
+
343
+
if (duration !== undefined) mediaSource.duration = duration;
344
+
345
+
const sourceBuffer = mediaSource.addSourceBuffer(mimeType);
346
+
347
+
// 'reader' is always the current active reader; the seeking handler
348
+
// closes over this variable so it always cancels the right one.
349
+
let reader = stream.getReader();
350
+
let seekPending = false;
351
+
let seekTarget = 0;
352
+
353
+
const onSeeking = () => {
354
+
if (!seekFn) return;
355
+
const audio = itemEl.audio;
356
+
const target = audio.currentTime;
357
+
358
+
// Only intervene if the target is outside what's already buffered.
359
+
for (let i = 0; i < audio.buffered.length; i++) {
360
+
if (
361
+
audio.buffered.start(i) <= target && target <= audio.buffered.end(i)
362
+
) {
363
+
return; // Browser can handle it with buffered data.
364
+
}
365
+
}
366
+
367
+
seekPending = true;
368
+
seekTarget = target;
369
+
reader.cancel().catch(() => {});
370
+
};
371
+
372
+
itemEl.audio.addEventListener("seeking", onSeeking);
373
+
374
+
try {
375
+
while (true) {
376
+
if (!this.#mediaSourceUrls.has(id)) {
377
+
await reader.cancel();
378
+
break;
379
+
}
380
+
381
+
let done, value;
382
+
383
+
try {
384
+
({ done, value } = await reader.read());
385
+
} catch {
386
+
done = true;
387
+
}
388
+
389
+
if (!this.#mediaSourceUrls.has(id)) break;
390
+
391
+
if (seekPending) {
392
+
seekPending = false;
393
+
394
+
// Clear all buffered data before feeding from the new position.
395
+
if (sourceBuffer.updating) {
396
+
await new Promise((r) =>
397
+
sourceBuffer.addEventListener("updateend", r, { once: true })
398
+
);
399
+
}
400
+
await new Promise((r) => {
401
+
sourceBuffer.addEventListener("updateend", r, { once: true });
402
+
sourceBuffer.remove(0, Infinity);
403
+
});
404
+
405
+
if (!seekFn) throw new Error("seekFn is undefined");
406
+
reader = (await seekFn(seekTarget)).getReader();
407
+
408
+
continue;
409
+
}
410
+
411
+
if (done) {
412
+
if (mediaSource.readyState === "open") mediaSource.endOfStream();
413
+
break;
414
+
}
415
+
416
+
if (sourceBuffer.updating) {
417
+
await new Promise((r) =>
418
+
sourceBuffer.addEventListener("updateend", r, { once: true })
419
+
);
420
+
}
421
+
422
+
sourceBuffer.appendBuffer(value);
423
+
await new Promise((r) =>
424
+
sourceBuffer.addEventListener("updateend", r, { once: true })
425
+
);
426
+
}
427
+
} catch (err) {
428
+
console.error("[audio engine] Stream error:", err);
429
+
if (mediaSource.readyState === "open") mediaSource.endOfStream("decode");
430
+
} finally {
431
+
itemEl.audio.removeEventListener("seeking", onSeeking);
432
+
}
433
+
}
434
+
247
435
// RENDER
248
436
249
437
/**
···
274
462
initial-progress="${ip}"
275
463
mime-type="${audio.mimeType ? audio.mimeType : nothing}"
276
464
preload="${audio.isPreload ? `preload` : nothing}"
277
-
url="${audio.url}"
465
+
url="${audio.url ?? nothing}"
278
466
>
279
467
<audio
280
468
crossorigin="anonymous"
281
469
muted="true"
282
470
preload="auto"
283
471
>
284
-
<source
285
-
src="${audio.url}"
286
-
${audio.mimeType ? 'type="' + audio.mimeType + '"' : ""}
287
-
/>
472
+
${audio.url
473
+
? html`
474
+
<source
475
+
src="${audio.url}"
476
+
${audio.mimeType ? 'type="' + audio.mimeType + '"' : ""}
477
+
/>
478
+
`
479
+
: nothing}
288
480
</audio>
289
481
</de-audio-item>
290
482
`,
+15
-2
src/components/engine/audio/types.d.ts
+15
-2
src/components/engine/audio/types.d.ts
···
11
11
) => void;
12
12
};
13
13
14
-
export type Audio = {
14
+
export type Audio = AudioUrl | AudioStream;
15
+
16
+
export type AudioUrl = AudioBase & { url: string };
17
+
18
+
export type AudioStream = AudioBase & {
19
+
stream: ReadableStream;
20
+
21
+
/** Total duration in seconds. */
22
+
duration?: number;
23
+
24
+
/** Return a new stream starting at the given time. Required for seeking on stream items. */
25
+
seek?: (timeSeconds: number) => Promise<ReadableStream>;
26
+
};
27
+
28
+
export type AudioBase = {
15
29
id: string;
16
30
isPreload: boolean;
17
31
mimeType?: string;
18
32
// NOTE: Initial progress
19
33
progress?: number;
20
-
url: string;
21
34
};
22
35
23
36
export type AudioState = {
+14
-3
src/components/input/types.d.ts
+14
-3
src/components/input/types.d.ts
···
35
35
36
36
export type InputSchemeProvider = { SCHEME: string };
37
37
38
-
export type ResolvedUri = undefined | {
38
+
export type ResolvedUri = undefined | ResolveUriAsUrl | ResolveUriAsStream;
39
+
40
+
export type ResolveUriAsUrl = {
41
+
expiresAt: number;
42
+
url: string;
43
+
};
44
+
45
+
export type ResolveUriAsStream = {
46
+
expiresAt: number;
47
+
mimeType: string;
39
48
stream: ReadableStream;
40
-
expiresAt: number;
41
-
} | { url: string; expiresAt: number };
49
+
50
+
/** Total duration in seconds. */
51
+
duration?: number;
52
+
};
42
53
43
54
export type Source = { label: string; uri: string };
+10
-10
src/components/orchestrator/queue-audio/element.js
+10
-10
src/components/orchestrator/queue-audio/element.js
···
74
74
75
75
const activeItem = queue.now();
76
76
const nextItem = queue.future()[0] ?? null;
77
-
78
77
const tracks = this.output?.tracks.collection();
78
+
79
79
const activeTrack = activeItem
80
80
? tracks?.find((t) => t.id === activeItem.id)
81
81
: undefined;
82
+
82
83
const nextTrack = nextItem
83
84
? tracks?.find((t) => t.id === nextItem.id)
84
85
: undefined;
···
92
93
? await input.resolve({ method: "GET", uri: activeTrack.uri })
93
94
: undefined;
94
95
95
-
if (resolvedUri && "stream" in resolvedUri) {
96
-
throw new Error("Streams are not supported yet.");
97
-
}
98
-
99
-
const url = resolvedUri?.url;
100
-
101
96
// Check if we still need to render
102
97
if (queue.now?.()?.id !== activeItem?.id) return;
103
98
104
99
// Supply active track immediately
105
100
// TODO: Take URL expiration timestamp into account
106
-
const activeAudio = activeItem && url
107
-
? [{ id: activeItem.id, isPreload: false, url }]
101
+
// TODO: Add support for seeking streams
102
+
// (requires a lot of code, decoding audio frames, etc.)
103
+
const activeAudio = activeItem && resolvedUri
104
+
? [{ id: activeItem.id, isPreload: false, ...resolvedUri }]
108
105
: [];
106
+
109
107
audio.supply({
110
108
audio: activeAudio,
111
109
play: activeItem && isPlaying ? { audioId: activeItem.id } : undefined,
···
120
118
method: "GET",
121
119
uri: nextTrack.uri,
122
120
});
121
+
123
122
const nextUrl = resolvedNextUri && !("stream" in resolvedNextUri)
124
123
? resolvedNextUri.url
125
124
: undefined;
125
+
126
126
if (!nextUrl) return;
127
127
128
128
audio.supply({
···
143
143
const aud = now ? this.audio.state(now.id) : undefined;
144
144
145
145
if (aud?.hasEnded() && (await this.isLeader())) {
146
-
// TODO: Not sure yet if this is the best way to approach this.
146
+
// NOTE: Not sure yet if this is the best way to approach this.
147
147
// The idea is that scrobblers would more easily pick this up,
148
148
// as opposed to just resetting the audio.
149
149
if (this.repeatShuffle?.repeat()) {