1196 lines
55 KiB
JavaScript
1196 lines
55 KiB
JavaScript
// The room client: a participant publishes its camera and microphone through the relay
|
|
// and subscribes to everyone else; the room's clock (a room key) publishes the lifecycle.
|
|
//
|
|
// Two ways in:
|
|
// - a room service (joule-moq-room): the page makes its own Ed25519 key, asks the
|
|
// service for a grant for it (a publisher's or a viewer's), and presents the grant with
|
|
// a proof by that key (WAI live-rooms §10.5); the service's clock authors the lifecycle.
|
|
// The service's base URL is the `service` parameter, or `data-service` on #room-app;
|
|
// the page enters when `enter("publisher" | "viewer")` is called (the #join and #watch
|
|
// buttons). The relay is the WebSocket at `/moq` on the page's own origin unless `ws`
|
|
// names another, and WebTransport is not tried;
|
|
// - a test room (`fixture`): every key from the fixture, and a page with `role=clock`
|
|
// authors the lifecycle, driven by the test.
|
|
//
|
|
// Query parameters:
|
|
// service=<url> the room service, as above
|
|
// fixture=<url> the test room (descriptor, grants, composition descriptor, keys)
|
|
// who=<name> this participant, by name; `role=clock` for the room's clock
|
|
// wt=<url> the relay's WebTransport URL (https://127.0.0.1:<port>/moq)
|
|
// ws=<url> its WebSocket URL (ws://127.0.0.1:<port>/moq)
|
|
// hashes=<hex,...> serverCertificateHashes for a relay with a development certificate
|
|
// transport=<t> `websocket` forces the WebSocket binding; otherwise WebTransport is
|
|
// tried first and the binding is the fallback
|
|
// fetch=1 a late joiner: earlier groups of every media track by FETCH
|
|
// (the "Seek back" control, #seek, fetches a participant's last seconds from the relay's
|
|
// DVR window, verifies every object, plays the video from its first key frame in a view of
|
|
// its own, and returns to live; `roomTest.seekBack(seconds)` does the same)
|
|
// audioFrameMs=<n> the Opus frame duration, 20 by default
|
|
//
|
|
// Everything a test reads is on `window.roomTest`.
|
|
|
|
import initMoq, * as moq from "../pkg/joule_moq_web.js";
|
|
import initWai, * as wai from "../pkg/wai_live_web.js";
|
|
import { connect } from "../moq/transport.js";
|
|
import { MoqSession, RequestRefused, withTimeout } from "../moq/session.js";
|
|
import * as media from "../moq/media.js";
|
|
import { Compositor } from "./compose.js";
|
|
|
|
const q = new URLSearchParams(location.search);
|
|
const $ = (id) => document.getElementById(id);
|
|
const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
|
|
const unhex = (h) => new Uint8Array((h.match(/../g) || []).map((x) => parseInt(x, 16)));
|
|
const enc = new TextEncoder();
|
|
const dec = new TextDecoder();
|
|
|
|
const logLines = [];
|
|
function log(m) {
|
|
const line = `${(performance.now() / 1000).toFixed(3)} ${m}`;
|
|
logLines.push(line);
|
|
if (logLines.length > 2000) logLines.shift();
|
|
const el = $("log");
|
|
if (el) { el.textContent += line + "\n"; el.scrollTop = el.scrollHeight; }
|
|
console.log(line);
|
|
}
|
|
|
|
// Track Aliases this session's PUBLISHes give (§9.8): the publisher's for its own data.
|
|
const ALIAS = { catalog: 1, events: 2, audio: 10, audioReceipts: 11, video: 20, videoReceipts: 21 };
|
|
const RECEIPT_ALIAS = { audio: 1, video: 2 }; // the catalog's wai_receipt_alias
|
|
// Publisher priorities (MOQT §5.1.1, lower first): audio before video, the receipts
|
|
// tracks at their media track's, events before both, the catalog first.
|
|
const PRIORITY = { catalog: 0, events: 16, audio: 64, video: 96 };
|
|
// Subscriber priorities (§5.1.2) a participant gives its requests.
|
|
const SUB_PRIORITY = { catalog: 0, events: 8, audio: 32, video: 64 };
|
|
const VIDEO_TIMEOUT_MS = 2000; // OBJECT_DELIVERY_TIMEOUT on each video object
|
|
const AUDIO_TIMEOUT_MS = 1000;
|
|
const RECEIPT_WAIT_MS = 6000; // an object waits this long for its receipts
|
|
const RECOVER_AFTER_MS = 1000; // then, past this, its group of the receipts track is fetched
|
|
const MISSING_AFTER_MS = 4000; // receipts with no object after this: accounted by inclusion
|
|
|
|
const state = {
|
|
phase: "loading",
|
|
errors: [],
|
|
transport: null,
|
|
published: { audio: 0, video: 0, events: 0, bytes: 0 },
|
|
publishCodec: {},
|
|
subs: {}, // principal -> per-track tallies
|
|
timeline: [],
|
|
captions: [],
|
|
goaways: [],
|
|
sessions: 0,
|
|
mode: null,
|
|
ended: null,
|
|
// Seeking back into the relay's DVR window: "live", "seeking" (fetching and verifying)
|
|
// or "playing", and each run's counts.
|
|
dvr: { mode: "live", runs: [] },
|
|
};
|
|
window.roomTest = { state, logLines };
|
|
|
|
function fail(e) {
|
|
const m = (e && e.stack) || String(e);
|
|
state.errors.push(m);
|
|
log(`error: ${m}`);
|
|
status(`Something went wrong: ${(e && e.message) || e}`);
|
|
}
|
|
|
|
// What the page tells its reader, where it has a place for it.
|
|
function status(text) {
|
|
const el = $("status");
|
|
if (el) el.textContent = text;
|
|
}
|
|
window.addEventListener("error", (e) => fail(e.error || e.message));
|
|
window.addEventListener("unhandledrejection", (e) => fail(e.reason));
|
|
|
|
// -------------------------------------------------------------------------- the events track
|
|
|
|
// One principal's events track (live-events §8.1): an MSF event timeline whose groups
|
|
// are the room-clock minutes. Object 0 of a group is its join set (§8.2): here every
|
|
// earlier record of the chain (any other event MAY be repeated, and a late subscriber
|
|
// then holds the whole chain); each later object is an array of one record.
|
|
class EventsTrack {
|
|
constructor(seed, roomId, t0) {
|
|
this.seed = seed;
|
|
this.roomId = roomId;
|
|
this.t0 = t0;
|
|
this.seq = 0;
|
|
this.prev = new Uint8Array(32);
|
|
this.lastT = 0;
|
|
this.records = [];
|
|
this.group = null;
|
|
this.stream = null;
|
|
this.writer = null;
|
|
this.nextObject = 0;
|
|
this.session = null;
|
|
}
|
|
|
|
now() {
|
|
const t = Math.max(this.lastT, Math.round(Date.now() * 1000 - this.t0));
|
|
this.lastT = t;
|
|
return t;
|
|
}
|
|
|
|
async attach(session) {
|
|
this.session = session;
|
|
this.group = null; // a new session opens a new stream for the current group
|
|
}
|
|
|
|
async emit(kind, payload) {
|
|
const t = this.now();
|
|
const rec = wai.eventSign(this.seed, this.roomId, BigInt(this.seq), this.prev, BigInt(t), kind, payload);
|
|
const v = JSON.parse(wai.eventRead(rec));
|
|
this.prev = unhex(v.digest);
|
|
this.seq++;
|
|
await this.send(rec, t);
|
|
this.records.push(rec);
|
|
return { record: rec, view: v };
|
|
}
|
|
|
|
async send(rec, t) {
|
|
const s = this.session;
|
|
if (!s) return;
|
|
const g = Number(wai.moqGroup(BigInt(t)));
|
|
const msf = (r) => JSON.parse(wai.msfRecord(r, BigInt(this.t0)));
|
|
if (this.group !== g) {
|
|
if (this.stream) await this.stream.close();
|
|
this.group = g;
|
|
this.writer = new moq.SubgroupWriter(BigInt(ALIAS.events), BigInt(g), PRIORITY.events, false);
|
|
this.stream = await s.openStream(0);
|
|
await this.stream.write(this.writer.header());
|
|
await this.stream.write(this.writer.object(0n, enc.encode(JSON.stringify(this.records.map(msf)))));
|
|
this.nextObject = 1;
|
|
}
|
|
await this.stream.write(this.writer.object(BigInt(this.nextObject++), enc.encode(JSON.stringify([msf(rec)]))));
|
|
state.published.events++;
|
|
}
|
|
}
|
|
|
|
// --------------------------------------------------------------------------------- room
|
|
|
|
let fx, me, roomId, t0, room, compositor, session, transportOpts, eventsTrack, service;
|
|
const subscribed = new Map(); // principal -> subscription state
|
|
const signers = {};
|
|
const published = []; // the PUBLISH requests held open
|
|
|
|
function roomNow() {
|
|
return Math.max(0, Math.round(Date.now() * 1000 - t0));
|
|
}
|
|
|
|
// A principal as the page names it: the clock, this page, a fixture's participant, or a
|
|
// publisher by its grant's `order` attribute.
|
|
function nameOf(P) {
|
|
if (fx.clock && P === fx.clock.principal) return fx.participants.length ? "clock" : "room";
|
|
if (me && P === me.principal) return fx.participants.length ? me.name : "you";
|
|
const p = fx.participants.find((x) => x.principal === P);
|
|
if (p) return p.name;
|
|
const g = (state.grants || []).find((x) => x.principal === P);
|
|
const order = g && (g.attrs.find((a) => a[0] === "order") || [])[1];
|
|
return order != null ? `participant ${Number(order) + 1}` : P.slice(0, 8);
|
|
}
|
|
|
|
// The room's grants as the page holds them, read.
|
|
function readGrants() {
|
|
const gv = JSON.parse(room.grants());
|
|
if (gv.outcome !== "read") throw new Error(`the room's grants: ${gv.reason}`);
|
|
state.grants = gv.grants;
|
|
}
|
|
|
|
async function boot() {
|
|
await initMoq();
|
|
await initWai();
|
|
fx = await (await fetch(q.get("fixture"))).json();
|
|
const role = q.get("role") || "participant";
|
|
me = role === "clock" ? { name: "clock", ...fx.clock } : fx.participants.find((p) => p.name === q.get("who"));
|
|
if (!me) throw new Error(`no participant "${q.get("who")}" in the fixture`);
|
|
state.mode = "fixture";
|
|
transportOpts = {
|
|
webtransport: q.get("wt"),
|
|
websocket: q.get("ws"),
|
|
hashes: (q.get("hashes") || "").split(",").filter(Boolean).map(unhex),
|
|
force: q.get("transport") || "",
|
|
};
|
|
window.roomTest.role = role;
|
|
await begin(role);
|
|
}
|
|
|
|
// The room service's answer, or why there is none.
|
|
async function api(path, body) {
|
|
const r = await fetch(`${service}${path}`, body ? { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), cache: "no-store" } : { cache: "no-store" });
|
|
let j = null;
|
|
try { j = await r.json(); } catch { /* not JSON */ }
|
|
return { status: r.status, body: j };
|
|
}
|
|
|
|
const ROLE_OF = { publisher: "participant", viewer: "viewer" };
|
|
|
|
// Enter the room the service keeps: as a publisher (camera and microphone) or a viewer.
|
|
async function enter(role) {
|
|
if (state.mode) throw new Error("this page has entered already");
|
|
state.mode = "service";
|
|
state.phase = "entering";
|
|
await initMoq();
|
|
await initWai();
|
|
let r = await api("/room");
|
|
if (r.status !== 200) throw new Error(`the room service answered ${r.status}`);
|
|
let stream = null;
|
|
if (role === "publisher") {
|
|
// The camera and microphone first: a room place is asked for only once they are on.
|
|
status("Asking for your camera and microphone…");
|
|
const d = JSON.parse(wai.roomDescriptor(unhex(r.body.descriptor)));
|
|
stream = await navigator.mediaDevices.getUserMedia(constraintsOf(d));
|
|
}
|
|
// The key this page signs with: made here, and never sent anywhere.
|
|
const seed = crypto.getRandomValues(new Uint8Array(32));
|
|
const principal = hex(moq.publicKeyOf(seed));
|
|
status("Asking the room for a grant…");
|
|
r = await api("/grant", { principal, role });
|
|
if (r.status === 409 && role === "publisher") {
|
|
for (const t of stream.getTracks()) t.stop();
|
|
state.mode = null;
|
|
state.phase = "full";
|
|
status(r.body && r.body.error === "full" ? `The room has its ${r.body.max_publishers} publishers. You can watch, or try again later.` : "The room is busy. You can watch, or try again later.");
|
|
return state.phase;
|
|
}
|
|
if (r.status === 429) throw new Error(`too many grants from this address; try again in ${r.body.retry_after_s} s`);
|
|
if (r.status !== 200) throw new Error(`the room service answered ${r.status}${r.body && r.body.error ? ` (${r.body.error})` : ""}`);
|
|
const v = r.body.room;
|
|
const d = JSON.parse(wai.roomDescriptor(unhex(v.descriptor)));
|
|
fx = { room: v.room, t0_us: v.t0_us, descriptor: v.descriptor, compose: v.compose, grants: v.grants, clock: { principal: d.clock }, participants: [] };
|
|
me = { name: "you", seed: hex(seed), principal, grant: r.body.grant, grant_digest: r.body.digest, role };
|
|
state.grant = { role, exp: r.body.exp, digest: r.body.digest };
|
|
const ws = q.get("ws") || `${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/moq`;
|
|
transportOpts = { webtransport: null, websocket: ws, hashes: [], force: "websocket" };
|
|
window.roomTest.role = ROLE_OF[role];
|
|
await begin(ROLE_OF[role], stream);
|
|
pollRoom();
|
|
return state.phase;
|
|
}
|
|
|
|
function constraintsOf(d) {
|
|
const vp = d.profiles.find((p) => p.media === 2);
|
|
const ap = d.profiles.find((p) => p.media === 1);
|
|
const vl = vp.layers[0];
|
|
return {
|
|
audio: { channelCount: ap.channels, sampleRate: ap.sample_rate, echoCancellation: (ap.processing & 1) !== 0, noiseSuppression: (ap.processing & 2) !== 0, autoGainControl: (ap.processing & 4) !== 0 },
|
|
video: { width: vl.width, height: vl.height, frameRate: vl.fps },
|
|
};
|
|
}
|
|
|
|
// The room, the composition and the session, once the page knows who it is.
|
|
async function begin(role, stream) {
|
|
roomId = unhex(fx.room);
|
|
t0 = fx.t0_us;
|
|
const who = $("who");
|
|
if (who) who.textContent = `${me.name} · ${me.principal.slice(0, 12)}`;
|
|
room = new wai.Room(unhex(fx.descriptor), JSON.stringify(fx.grants));
|
|
readGrants();
|
|
state.descriptor = JSON.parse(wai.roomDescriptor(unhex(fx.descriptor)));
|
|
eventsTrack = new EventsTrack(unhex(me.seed), roomId, t0);
|
|
if (role !== "clock") {
|
|
compositor = new Compositor(wai, $("canvas"), unhex(fx.compose), { log });
|
|
await compositor.startAudio(new URL("./worklet.js", import.meta.url));
|
|
setInterval(() => { try { compositor.draw(); } catch (e) { fail(e); } }, Math.round(1000 / compositor.desc.params.fps));
|
|
setInterval(sweepMissing, 1000);
|
|
}
|
|
status("Connecting to the relay…");
|
|
await openSession(transportOpts);
|
|
if (role === "clock") {
|
|
window.roomTest.clockEmit = clockEmit;
|
|
state.phase = "ready";
|
|
} else if (role === "viewer") {
|
|
await subscribeEvents(fx.clock.principal);
|
|
state.phase = "watching";
|
|
status("Watching. Everything you receive is verified before it is shown.");
|
|
} else {
|
|
await startPublishing(stream);
|
|
await subscribeEvents(fx.clock.principal);
|
|
state.phase = "publishing";
|
|
status("You are in the room. Everything you receive is verified before it is shown.");
|
|
}
|
|
log(`ready as ${role} ${me.name}`);
|
|
}
|
|
|
|
async function openSession(opts) {
|
|
const t = await connect({ ...opts, log });
|
|
// A participant presents its room grant with a proof by its key (live-rooms §10.5.2);
|
|
// the clock, a room key with no grant, presents its own token.
|
|
const id = me.grant ? { grant: { record: unhex(me.grant), seed: unhex(me.seed) } } : { token: unhex(me.token) };
|
|
const s = await MoqSession.connect(t, moq, { ...id, implementation: "joule-moq-web/room", log });
|
|
state.transport = { kind: t.kind, protocol: t.protocol };
|
|
state.sessions++;
|
|
s.onGoaway = (g) => onGoaway(s, g);
|
|
session = s;
|
|
window.roomTest.session = s;
|
|
s.closed.then((i) => onClosed(s, i));
|
|
// The events track is published on every session.
|
|
const ns = [fx.room, me.principal, "events"];
|
|
published.push(await s.publish(ns, "0", ALIAS.events, { publisher_priority: PRIORITY.events }));
|
|
await eventsTrack.attach(s);
|
|
return s;
|
|
}
|
|
|
|
// The session ended: by this page, by a move after GOAWAY, or by the relay (its session
|
|
// limit, the grant's expiry, a refusal).
|
|
function onClosed(s, info) {
|
|
if (s !== session || state.goaways.some((g) => !g.moved)) return;
|
|
state.ended = { code: info && info.code, reason: info && info.reason };
|
|
if (state.phase === "left") return;
|
|
state.phase = "ended";
|
|
log(`the session ended: ${JSON.stringify(state.ended)}`);
|
|
if (state.mode === "service") status(`The session ended${info && info.reason ? `: ${info.reason}` : ""}. Reload the page to enter again.`);
|
|
}
|
|
|
|
// Leave: the session closes, and the relay ends this page's tracks, which the room's clock
|
|
// records as the publisher leaving.
|
|
async function leave() {
|
|
state.phase = "left";
|
|
if (state.capture && state.capture.stream) for (const t of state.capture.stream.getTracks()) t.stop();
|
|
if (session) session.close(0, "left");
|
|
status("You left the room. Reload the page to enter again.");
|
|
}
|
|
|
|
// A new room may start while this page watches an empty one: the page says so.
|
|
function pollRoom() {
|
|
setInterval(async () => {
|
|
if (state.phase !== "watching" && state.phase !== "publishing") return;
|
|
try {
|
|
const r = await api("/room");
|
|
if (r.status === 200 && r.body.room !== fx.room) {
|
|
state.roomMoved = r.body.room;
|
|
if (state.phase === "watching") status("A new room has started. Reload the page to watch it.");
|
|
}
|
|
} catch { /* the next poll */ }
|
|
}, 20000);
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------- the clock
|
|
|
|
async function clockEmit(desc) {
|
|
const lp = wai.lifecyclePayload(JSON.stringify(desc));
|
|
const r = await eventsTrack.emit(lp[0], lp.subarray(1));
|
|
return r.view;
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------ publishing
|
|
|
|
async function pickVideoCodec(profile) {
|
|
const l = profile.layers[0];
|
|
for (const codec of profile.codecs) {
|
|
const cfg = { codec, width: l.width, height: l.height, bitrate: l.max_kbps * 1000, framerate: l.fps, latencyMode: "realtime" };
|
|
try {
|
|
const s = await VideoEncoder.isConfigSupported(cfg);
|
|
if (s.supported) return cfg;
|
|
} catch { /* not this one */ }
|
|
}
|
|
throw new Error(`no video codec of the profile is supported: ${profile.codecs.join(", ")}`);
|
|
}
|
|
|
|
async function startPublishing(given) {
|
|
const d = state.descriptor;
|
|
const vp = d.profiles.find((p) => p.media === 2);
|
|
const ap = d.profiles.find((p) => p.media === 1);
|
|
const vcfg = await pickVideoCodec(vp);
|
|
const frameMs = Number(q.get("audioFrameMs") || 20);
|
|
const acfg = {
|
|
codec: "opus",
|
|
sampleRate: ap.sample_rate,
|
|
numberOfChannels: ap.channels,
|
|
bitrate: 32000,
|
|
opus: { frameDuration: frameMs * 1000, usedtx: (ap.processing & 8) !== 0, complexity: 5 },
|
|
};
|
|
const as = await AudioEncoder.isConfigSupported(acfg);
|
|
if (!as.supported) throw new Error("this browser's AudioEncoder does not take Opus at 48 kHz");
|
|
state.publishCodec = { video: vcfg.codec, audio: "opus", audioFrameMs: frameMs };
|
|
const vl = vp.layers[0];
|
|
const catalog = {
|
|
version: 1,
|
|
tracks: [
|
|
{
|
|
namespace: `${fx.room}-${me.principal}-video`, name: "0", packaging: "wai", isLive: true, role: "video",
|
|
codec: vcfg.codec, width: vl.width, height: vl.height, framerate: vl.fps, bitrate: vl.max_kbps * 1000, renderGroup: 1,
|
|
wai_capabilities: [media.CAPABILITY_OF(vcfg.codec)], wai_receipts: true, wai_receipt_alias: RECEIPT_ALIAS.video, wai_receipts_track: "0_receipts",
|
|
},
|
|
{ namespace: `${fx.room}-${me.principal}-video`, name: "0_receipts", packaging: "wai-receipts", isLive: true },
|
|
{
|
|
namespace: `${fx.room}-${me.principal}-audio`, name: "0", packaging: "wai", isLive: true, role: "audio",
|
|
codec: "opus", samplerate: ap.sample_rate, channelConfig: String(ap.channels), bitrate: 32000, renderGroup: 1,
|
|
wai_capabilities: ["wai.audio.opus"], wai_receipts: true, wai_receipt_alias: RECEIPT_ALIAS.audio, wai_receipts_track: "0_receipts",
|
|
},
|
|
{ namespace: `${fx.room}-${me.principal}-audio`, name: "0_receipts", packaging: "wai-receipts", isLive: true },
|
|
{ namespace: `${fx.room}-${me.principal}-events`, name: "0", packaging: "eventtimeline", eventType: "urn:wai:event:1", mimetype: "application/json", isLive: true },
|
|
],
|
|
};
|
|
state.catalog = catalog;
|
|
signers.video = new moq.TrackSigner(unhex(me.seed), BigInt(RECEIPT_ALIAS.video), BigInt(ALIAS.video), BigInt(ALIAS.videoReceipts), PRIORITY.video, me.name, BigInt(VIDEO_TIMEOUT_MS));
|
|
signers.audio = new moq.TrackSigner(unhex(me.seed), BigInt(RECEIPT_ALIAS.audio), BigInt(ALIAS.audio), BigInt(ALIAS.audioReceipts), PRIORITY.audio, me.name, BigInt(AUDIO_TIMEOUT_MS));
|
|
await publishTracks(session);
|
|
|
|
const stream = given || await navigator.mediaDevices.getUserMedia(constraintsOf(d));
|
|
state.capture = { video: stream.getVideoTracks()[0].getSettings(), audio: stream.getAudioTracks()[0].getSettings() };
|
|
Object.defineProperty(state.capture, "stream", { value: stream, enumerable: false });
|
|
let vgroup = 0, agroup = 0, vframes = 0;
|
|
const venc = new VideoEncoder({
|
|
output: (chunk) => {
|
|
const g = vgroup++;
|
|
const b = new Uint8Array(chunk.byteLength);
|
|
chunk.copyTo(b);
|
|
const payload = vcfg.codec.startsWith("vp09") ? media.ivfOne(b, vl.width, vl.height, chunk.timestamp) : media.av1One(b, vl.width, vl.height, vl.fps, 1);
|
|
const env = wai.pack(media.manifestFor(media.CAPABILITY_OF(vcfg.codec), { width: vl.width, height: vl.height, timestamp_us: chunk.timestamp, key: chunk.type === "key" }), payload);
|
|
sendGroup("video", g, env);
|
|
},
|
|
error: fail,
|
|
});
|
|
venc.configure(vcfg);
|
|
const aenc = new AudioEncoder({
|
|
output: (chunk) => {
|
|
const g = agroup++;
|
|
const b = new Uint8Array(chunk.byteLength);
|
|
chunk.copyTo(b);
|
|
const samples = Math.round(((chunk.duration || frameMs * 1000) / 1e6) * ap.sample_rate);
|
|
const env = wai.pack(media.manifestFor("wai.audio.opus", { sample_rate: ap.sample_rate, channels: ap.channels, timestamp_us: chunk.timestamp, samples }), media.oggOpusOne(b, { channels: ap.channels, inputRate: ap.sample_rate, samples }));
|
|
sendGroup("audio", g, env);
|
|
},
|
|
error: fail,
|
|
});
|
|
aenc.configure(acfg);
|
|
// Frames at the profile's layer size: a capture that is another size is scaled first.
|
|
const scaler = new OffscreenCanvas(vl.width, vl.height);
|
|
const sctx = scaler.getContext("2d");
|
|
pump(new MediaStreamTrackProcessor({ track: stream.getVideoTracks()[0] }).readable, (frame) => {
|
|
// A key frame every two seconds, so a subscriber that joins can start within two.
|
|
const key = vframes % (vl.fps * 2) === 0;
|
|
if (venc.encodeQueueSize < 4) {
|
|
vframes++;
|
|
let f = frame;
|
|
if (frame.displayWidth !== vl.width || frame.displayHeight !== vl.height) {
|
|
sctx.drawImage(frame, 0, 0, vl.width, vl.height);
|
|
f = new VideoFrame(scaler, { timestamp: frame.timestamp });
|
|
}
|
|
venc.encode(f, { keyFrame: key });
|
|
// The participant's own tile shows what it sends (a self view, no audio of its own).
|
|
if (compositor) compositor.setVideo(me.principal, f.clone());
|
|
if (f !== frame) f.close();
|
|
}
|
|
frame.close();
|
|
});
|
|
pump(new MediaStreamTrackProcessor({ track: stream.getAudioTracks()[0] }).readable, (ad) => {
|
|
if (ad.sampleRate !== acfg.sampleRate || ad.numberOfChannels !== acfg.numberOfChannels) {
|
|
if (!state.audioMismatch) log(`the microphone gives ${ad.numberOfChannels} channel(s) at ${ad.sampleRate} Hz; the profile asks ${acfg.numberOfChannels} at ${acfg.sampleRate}`);
|
|
state.audioMismatch = { sampleRate: ad.sampleRate, channels: ad.numberOfChannels };
|
|
ad.close();
|
|
return;
|
|
}
|
|
aenc.encode(ad);
|
|
ad.close();
|
|
});
|
|
}
|
|
|
|
async function publishTracks(s) {
|
|
const P = me.principal;
|
|
const pub = async (ns, name, alias, props) => published.push(await s.publish(ns, name, alias, props));
|
|
await pub([fx.room, P], "catalog", ALIAS.catalog, { publisher_priority: PRIORITY.catalog });
|
|
for (const m of ["audio", "video"]) {
|
|
await pub([fx.room, P, m], "0_receipts", ALIAS[`${m}Receipts`], { publisher_priority: PRIORITY[m] });
|
|
await pub([fx.room, P, m], "0", ALIAS[m], { publisher_priority: PRIORITY[m], object_delivery_timeout: m === "video" ? VIDEO_TIMEOUT_MS : AUDIO_TIMEOUT_MS });
|
|
signers[m].setAliases(BigInt(ALIAS[m]), BigInt(ALIAS[`${m}Receipts`]));
|
|
}
|
|
// The catalog: group 0, one object (MSF §5); a session that moves sends it again.
|
|
const w = new moq.SubgroupWriter(BigInt(ALIAS.catalog), 0n, PRIORITY.catalog, true);
|
|
const b = enc.encode(JSON.stringify(state.catalog));
|
|
const obj = w.object(0n, b);
|
|
const h = w.header();
|
|
const all = new Uint8Array(h.length + obj.length);
|
|
all.set(h, 0);
|
|
all.set(obj, h.length);
|
|
await s.sendStream(all, 1000);
|
|
w.free();
|
|
}
|
|
|
|
async function pump(readable, each) {
|
|
const r = readable.getReader();
|
|
for (;;) {
|
|
const { value, done } = await r.read();
|
|
if (done) return;
|
|
try { each(value); } catch (e) { fail(e); }
|
|
}
|
|
}
|
|
|
|
// Group g of a media track: its receipts first (no later than the group's first object,
|
|
// moq-streaming-format §4.1), then the object, each on a stream of its own.
|
|
async function sendGroup(m, g, envelope) {
|
|
const s = session;
|
|
try {
|
|
const now = new Date().toISOString();
|
|
const sg = signers[m].sign(BigInt(g), envelope, now, now);
|
|
const r = sg.receiptsStream();
|
|
const md = sg.mediaStream();
|
|
sg.free();
|
|
// sendOrder: higher goes first; audio before video, receipts before their object.
|
|
const base = m === "audio" ? 2_000_000 : 1_000_000;
|
|
await s.sendStream(r, base + 1);
|
|
await s.sendStream(md, base);
|
|
state.published[m]++;
|
|
state.published.bytes += md.length + r.length;
|
|
} catch (e) {
|
|
if (s.goaway || s.closedInfo) return; // the session is moving; the next group goes on the new one
|
|
fail(e);
|
|
}
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------ subscribing
|
|
|
|
function tally(principal) {
|
|
if (!state.subs[principal]) {
|
|
const t = () => ({ received: 0, verified: 0, refused: 0, no_receipt: 0, receipts: 0, receipts_refused: 0, decoded: 0, decode_errors: 0, reset: 0, dropped_accounted: 0, dropped_unaccounted: 0, fetched: 0, fetched_verified: 0, fetched_decoded: 0, first_live_group: null, first_fetched_group: null, last_fetched_group: null, chain: {} });
|
|
state.subs[principal] = { audio: t(), video: t(), catalog: false, events: 0 };
|
|
}
|
|
return state.subs[principal];
|
|
}
|
|
|
|
async function subscribeEvents(principal) {
|
|
const ns = [fx.room, principal, "events"];
|
|
const onRecords = (payload) => {
|
|
let arr;
|
|
try { arr = JSON.parse(dec.decode(payload)); } catch { return log(`events of ${principal.slice(0, 8)}: an object that is not JSON`); }
|
|
for (const rec of arr) {
|
|
let bytes;
|
|
try { bytes = wai.msfRecordRead(JSON.stringify(rec), BigInt(t0)); } catch (e) { log(`an event record refused: ${e.message || e}`); continue; }
|
|
records = records.then(() => onRecord(bytes)).catch(fail);
|
|
}
|
|
};
|
|
const handler = {
|
|
stream: () => ({ object: (j, p) => onRecords(p) }),
|
|
fill: () => ({ object: (j, p) => onRecords(p) }),
|
|
};
|
|
await session.subscribe(ns, "0", {
|
|
subscriber_priority: SUB_PRIORITY.events, group_order: 1, forward: true, rendezvous_timeout: 10000,
|
|
location_filter: { type: "next" }, fill: { location_filter: { type: "relative", start_group: 1 } },
|
|
}, handler);
|
|
}
|
|
|
|
// Records are taken one at a time, in the order they arrive: a join under a grant the page
|
|
// does not hold waits while the page asks the service for the room's grants again.
|
|
let records = Promise.resolve();
|
|
|
|
async function onRecord(bytes) {
|
|
const v = JSON.parse(room.offer(bytes));
|
|
if (!v.ok) return log(`an event refused: ${v.reason}`);
|
|
if (!v.new) return;
|
|
if (service && v.kind_name === "join" && v.payload && v.payload.detail && !state.grants.some((g) => g.digest === v.payload.detail.grant)) {
|
|
await refreshGrants();
|
|
}
|
|
const who = nameOf(v.principal);
|
|
const entry = { t: v.t, who, kind: v.kind_name, payload: v.payload, digest: v.digest };
|
|
state.timeline.push(entry);
|
|
renderTimeline(entry);
|
|
if (v.kind_name === "caption") {
|
|
state.captions.push({ t: v.t, who, text: v.payload.text, origin: v.payload.origin_label, model: v.payload.model, lang: v.payload.lang });
|
|
renderCaption();
|
|
}
|
|
if (v.kind >= 0x10) onLifecycle();
|
|
}
|
|
|
|
async function refreshGrants() {
|
|
const r = await api("/room");
|
|
if (r.status !== 200 || r.body.room !== fx.room) return log("the room's grants could not be read again");
|
|
fx.grants = r.body.grants;
|
|
room.setGrants(JSON.stringify(fx.grants));
|
|
readGrants();
|
|
log(`the room's grants read again: ${fx.grants.length}`);
|
|
}
|
|
|
|
function onLifecycle() {
|
|
const st = JSON.parse(room.state());
|
|
state.room = st;
|
|
if (compositor) compositor.relayout(room, roomNow());
|
|
for (const own of st.own) {
|
|
if (own.principal === me.principal || subscribed.has(own.principal)) continue;
|
|
subscribeParticipant(own.principal).catch(fail);
|
|
}
|
|
}
|
|
|
|
async function subscribeParticipant(P) {
|
|
if (subscribed.has(P)) return;
|
|
const sub = { principal: P, tracks: {}, catalog: null };
|
|
subscribed.set(P, sub);
|
|
tally(P);
|
|
await subscribeEvents(P);
|
|
let resolveCatalog;
|
|
const got = new Promise((r) => { resolveCatalog = r; });
|
|
const onCatalog = (p) => {
|
|
try { const c = JSON.parse(dec.decode(p)); if (!sub.catalog) { sub.catalog = c; tally(P).catalog = true; resolveCatalog(c); } } catch (e) { fail(e); }
|
|
};
|
|
await session.subscribe([fx.room, P], "catalog", {
|
|
subscriber_priority: SUB_PRIORITY.catalog, group_order: 1, forward: true, rendezvous_timeout: 10000,
|
|
location_filter: { type: "next" }, fill: { location_filter: { type: "relative", start_group: 1 } },
|
|
}, { stream: () => ({ object: (j, p) => onCatalog(p) }), fill: () => ({ object: (j, p) => onCatalog(p) }) });
|
|
const catalog = await withTimeout(got, 10000, `the catalog of ${P.slice(0, 8)}`);
|
|
for (const tr of catalog.tracks.filter((x) => x.packaging === "wai")) {
|
|
await subscribeMedia(sub, tr);
|
|
}
|
|
}
|
|
|
|
function mediaOf(tr) {
|
|
return tr.role === "audio" ? "audio" : "video";
|
|
}
|
|
|
|
async function subscribeMedia(sub, tr) {
|
|
const P = sub.principal;
|
|
const m = mediaOf(tr);
|
|
const T = tally(P)[m];
|
|
const ns = [fx.room, P, m];
|
|
const verifier = new moq.TrackVerifier(unhex(P), BigInt(tr.wai_receipt_alias));
|
|
const st = { verifier, pending: new Map(), seen: new Set(), receiptsAt: new Map(), decoder: null, waitingKey: true, tr, m };
|
|
sub.tracks[m] = st;
|
|
st.decoder = makeDecoder(P, tr, T, st);
|
|
|
|
const onReceipts = (group, payload) => {
|
|
const r = JSON.parse(verifier.receipts(BigInt(group), payload, BigInt(Date.now())));
|
|
if (!r.ok) { T.receipts_refused++; log(`receipts of ${m} group ${group} of ${P.slice(0, 8)} refused: ${r.reason}`); return; }
|
|
T.receipts++;
|
|
T.chain[r.chain] = (T.chain[r.chain] || 0) + 1;
|
|
if (!st.seen.has(group)) st.receiptsAt.set(group, performance.now());
|
|
const held = st.pending.get(group);
|
|
if (held) { st.pending.delete(group); onObject(group, held.payload, held.fetched); }
|
|
};
|
|
const onObject = (group, payload, fetched) => {
|
|
const r = JSON.parse(verifier.object(BigInt(group), 0n, payload));
|
|
if (!r.ok && r.reason === "no_receipt") {
|
|
if (!st.pending.has(group)) { st.pending.set(group, { payload, at: performance.now(), fetched }); T.no_receipt++; }
|
|
return;
|
|
}
|
|
st.receiptsAt.delete(group);
|
|
if (!r.ok) { T.refused++; log(`${m} group ${group} of ${P.slice(0, 8)} refused: ${r.reason}`); return; }
|
|
if (fetched) T.fetched_verified++; else T.verified++;
|
|
decodeEnvelope(P, st, T, group, payload, fetched);
|
|
};
|
|
st.onObject = onObject;
|
|
st.onReceipts = onReceipts;
|
|
st.ns = ns;
|
|
|
|
const rh = {
|
|
stream: (h) => ({ object: (j, p) => onReceipts(h.group_id, p) }),
|
|
fill: () => ({ object: (j, p) => onReceipts(j.group_id, p) }),
|
|
};
|
|
const mh = {
|
|
stream: (h) => ({
|
|
object: (j, p) => {
|
|
if (st.seen.has(h.group_id)) return;
|
|
st.seen.add(h.group_id);
|
|
T.received++;
|
|
if (T.first_live_group == null) T.first_live_group = h.group_id;
|
|
if (T.last_live_group == null || h.group_id > T.last_live_group) T.last_live_group = h.group_id;
|
|
onObject(h.group_id, p, false);
|
|
},
|
|
reset: () => { T.reset++; },
|
|
}),
|
|
fill: () => ({ object: () => {} }),
|
|
};
|
|
// The receipts track first and with no delivery timeout (moq-streaming-format §4.1),
|
|
// at the media track's subscriber priority.
|
|
await session.subscribe(ns, tr.wai_receipts_track, { subscriber_priority: SUB_PRIORITY[m], group_order: 1, forward: true, rendezvous_timeout: 10000, location_filter: { type: "next" } }, rh);
|
|
await session.subscribe(ns, tr.name, { subscriber_priority: SUB_PRIORITY[m], group_order: 1, forward: true, rendezvous_timeout: 10000, location_filter: { type: "next" } }, mh);
|
|
log(`subscribed to ${m} of ${P.slice(0, 8)} (${tr.codec}, receipt alias ${tr.wai_receipt_alias})`);
|
|
|
|
if (q.get("fetch") === "1") lateJoin(sub, st, T, ns, tr).catch(fail);
|
|
}
|
|
|
|
// A late joiner (MOQT §3.2): every earlier group of the track and of its receipts track,
|
|
// by FETCH, from group 0 to the Largest Object; each object verified against its receipt
|
|
// like any other, and the video decoded from its first key frame.
|
|
async function lateJoin(sub, st, T, ns, tr) {
|
|
const params = { group_order: 1, subscriber_priority: SUB_PRIORITY[st.m] + 1, location_filter: { type: "absolute", start: [0, 0] } };
|
|
// A FETCH is refused before any object is sent, so one tried again counts nothing twice:
|
|
// a relay asked for groups still arriving may pass the request to the publisher, which
|
|
// takes no requests; a moment later its cache holds them.
|
|
const fetchAgain = async (name, onObject) => {
|
|
for (let i = 0; ; i++) {
|
|
try { return await session.fetch(ns, name, params, onObject); } catch (e) {
|
|
if (i >= 3 || !(e instanceof RequestRefused)) throw e;
|
|
log(`FETCH of ${st.m} ${name} of ${sub.principal.slice(0, 8)}: ${e.message}; again`);
|
|
T.fetch_retries = (T.fetch_retries || 0) + 1;
|
|
await new Promise((r) => setTimeout(r, 750));
|
|
}
|
|
}
|
|
};
|
|
const rec = [];
|
|
const r1 = await fetchAgain(tr.wai_receipts_track, (j, p) => rec.push([j.group_id, p]));
|
|
for (const [g, p] of rec) {
|
|
const r = JSON.parse(st.verifier.receipts(BigInt(g), p, BigInt(Date.now())));
|
|
if (r.ok) { T.receipts++; T.chain[r.chain] = (T.chain[r.chain] || 0) + 1; } else T.receipts_refused++;
|
|
}
|
|
const objs = [];
|
|
const r2 = await fetchAgain(tr.name, (j, p) => objs.push([j.group_id, p]));
|
|
for (const [g, p] of objs) {
|
|
T.fetched++;
|
|
if (T.first_fetched_group == null) T.first_fetched_group = g;
|
|
T.last_fetched_group = g;
|
|
st.seen.add(g);
|
|
st.onObject(g, p, true);
|
|
}
|
|
log(`FETCH of ${st.m} of ${sub.principal.slice(0, 8)}: ${rec.length} receipts (${r1.how}), ${objs.length} objects (${r2.how})`);
|
|
}
|
|
|
|
function makeDecoder(P, tr, T, st) {
|
|
if (tr.role === "audio") {
|
|
const d = new AudioDecoder({
|
|
output: (ad) => { T.decoded++; if (compositor) compositor.addAudio(P, ad); else ad.close(); },
|
|
error: (e) => { T.decode_errors++; log(`audio decoder of ${P.slice(0, 8)}: ${e.message || e}`); },
|
|
});
|
|
d.configure({ codec: "opus", sampleRate: tr.samplerate, numberOfChannels: Number(tr.channelConfig || 1) });
|
|
return d;
|
|
}
|
|
const d = new VideoDecoder({
|
|
output: (vf) => {
|
|
T.decoded++;
|
|
if (compositor) compositor.setVideo(P, vf); else vf.close();
|
|
},
|
|
error: (e) => { T.decode_errors++; log(`video decoder of ${P.slice(0, 8)}: ${e.message || e}`); st.waitingKey = true; },
|
|
});
|
|
d.configure({ codec: tr.codec, codedWidth: tr.width, codedHeight: tr.height, optimizeForLatency: true });
|
|
return d;
|
|
}
|
|
|
|
function decodeEnvelope(P, st, T, group, envelope, fetched) {
|
|
let e;
|
|
try { e = wai.unpack(envelope); } catch (err) { T.decode_errors++; return log(`an envelope that does not read: ${err.message || err}`); }
|
|
const manifest = JSON.parse(e.manifest);
|
|
const cap = e.capability;
|
|
const payload = e.payload();
|
|
e.free();
|
|
if (!st.tr.wai_capabilities.includes(cap)) {
|
|
// A capability the catalog did not declare is a protocol error (§4), not a fallback.
|
|
T.decode_errors++;
|
|
return log(`a capability the catalog does not declare: ${cap}`);
|
|
}
|
|
// A codec that met an error is closed for good (WebCodecs): a new one is made, and the
|
|
// video waits for its next key frame.
|
|
if (st.decoder.state === "closed") {
|
|
st.decoder = makeDecoder(P, st.tr, T, st);
|
|
st.waitingKey = true;
|
|
T.decoder_restarts = (T.decoder_restarts || 0) + 1;
|
|
log(`${st.m} decoder of ${P.slice(0, 8)}: made again after an error`);
|
|
}
|
|
const d = st.decoder;
|
|
if (d.state !== "configured") return;
|
|
try {
|
|
if (cap === "wai.audio.opus") {
|
|
const packets = media.oggOpusPackets(payload);
|
|
// Fetched audio is earlier than the live stream: decoded on a decoder of its own and
|
|
// counted, never mixed.
|
|
if (fetched) return decodeFetchedAudio(st, T, packets, manifest);
|
|
for (const pk of packets) {
|
|
d.decode(new EncodedAudioChunk({ type: "key", timestamp: manifest.target.timestamp_us, data: pk }));
|
|
}
|
|
return;
|
|
}
|
|
const units = cap === "wai.video.vp9" ? media.ivfFrames(payload).map((f) => f.data) : media.av1Units(payload);
|
|
const key = !!manifest.target.key;
|
|
// A fetched group older than the live stream is decoded on its own decoder.
|
|
if (fetched) return decodeFetched(P, st, T, units, key, manifest);
|
|
if (st.waitingKey && !key) return;
|
|
// Each group rides a stream of its own and waits for its receipts, so under load a
|
|
// group can be released after a later one: decoded then, it would corrupt the
|
|
// decoder's references. It is dropped, and counted.
|
|
if (st.lastVideoGroup != null && group <= st.lastVideoGroup) {
|
|
T.late_dropped = (T.late_dropped || 0) + 1;
|
|
return;
|
|
}
|
|
st.lastVideoGroup = group;
|
|
st.waitingKey = false;
|
|
for (const u of units) d.decode(new EncodedVideoChunk({ type: key ? "key" : "delta", timestamp: manifest.target.timestamp_us, data: u }));
|
|
} catch (err) {
|
|
T.decode_errors++;
|
|
log(`decode of ${st.m} group ${group} of ${P.slice(0, 8)}: ${err.message || err}`);
|
|
}
|
|
}
|
|
|
|
function decodeFetchedAudio(st, T, packets, manifest) {
|
|
if (!st.fetchDecoder) {
|
|
st.fetchDecoder = new AudioDecoder({ output: (ad) => { T.fetched_decoded++; ad.close(); }, error: (e) => { T.decode_errors++; log(`fetched audio decoder: ${e.message || e}`); } });
|
|
st.fetchDecoder.configure({ codec: "opus", sampleRate: st.tr.samplerate, numberOfChannels: Number(st.tr.channelConfig || 1) });
|
|
}
|
|
for (const pk of packets) st.fetchDecoder.decode(new EncodedAudioChunk({ type: "key", timestamp: manifest.target.timestamp_us, data: pk }));
|
|
}
|
|
|
|
// The earlier groups a late joiner fetched, decoded in order on a decoder of their own from
|
|
// the first key frame, so the live decoder's state is not disturbed.
|
|
function decodeFetched(P, st, T, units, key, manifest) {
|
|
if (!st.fetchDecoder) {
|
|
if (!key) return;
|
|
st.fetchDecoder = new VideoDecoder({ output: (vf) => { T.fetched_decoded++; vf.close(); }, error: (e) => { T.decode_errors++; log(`fetched video decoder: ${e.message || e}`); } });
|
|
st.fetchDecoder.configure({ codec: st.tr.codec, codedWidth: st.tr.width, codedHeight: st.tr.height });
|
|
}
|
|
for (const u of units) st.fetchDecoder.decode(new EncodedVideoChunk({ type: key ? "key" : "delta", timestamp: manifest.target.timestamp_us, data: u }));
|
|
}
|
|
|
|
// Objects that waited past RECEIPT_WAIT_MS for their receipts, and receipts whose object
|
|
// never came: the latter accounted for by inclusion (jwp-receipts §4.4).
|
|
function sweepMissing() {
|
|
const now = performance.now();
|
|
for (const sub of subscribed.values()) {
|
|
for (const st of Object.values(sub.tracks)) {
|
|
const T = tally(sub.principal)[st.m];
|
|
for (const [g, p] of st.pending) {
|
|
if (now - p.at > RECEIPT_WAIT_MS) { st.pending.delete(g); T.refused++; log(`${st.m} group ${g}: its receipts did not arrive`); continue; }
|
|
// A receipts-track object the subscription did not deliver (one that reached the
|
|
// relay just before the subscription did) is recovered by FETCH of its group
|
|
// (moq-streaming-format §4.1).
|
|
if (now - p.at > RECOVER_AFTER_MS && !p.recovering) {
|
|
p.recovering = true;
|
|
T.receipt_fetches = (T.receipt_fetches || 0) + 1;
|
|
session.fetch(st.ns, st.tr.wai_receipts_track, { group_order: 1, location_filter: { type: "absolute_group_end", start: [g, 0], end_group_delta: 0 } }, (j, payload) => st.onReceipts(j.group_id, payload))
|
|
.then(() => { if (!st.pending.has(g)) T.receipts_recovered = (T.receipts_recovered || 0) + 1; })
|
|
.catch((e) => log(`${st.m} group ${g}: the FETCH of its receipts: ${e.message || e}`));
|
|
}
|
|
}
|
|
for (const [g, at] of st.receiptsAt) {
|
|
if (st.seen.has(g)) { st.receiptsAt.delete(g); continue; }
|
|
// Receipts of a group before the media subscription's first are of no object this
|
|
// subscription asked for (the receipts track was subscribed first): not a drop.
|
|
if (T.first_live_group == null || g < T.first_live_group) { if (now - at > MISSING_AFTER_MS) st.receiptsAt.delete(g); continue; }
|
|
if (now - at < MISSING_AFTER_MS) continue;
|
|
st.receiptsAt.delete(g);
|
|
const r = JSON.parse(st.verifier.inclusion(BigInt(g), 0n));
|
|
if (r.ok) T.dropped_accounted++; else T.dropped_unaccounted++;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------- seeking back
|
|
|
|
// Seeking back (WAI live-recording §8, live-rooms §10.6): a participant's last seconds, by
|
|
// FETCH below the live edge of its tracks and of their receipts tracks, which the relay
|
|
// answers from its DVR window. Every object is verified against its receipt exactly as a
|
|
// live one is (the same verifier, which admits the earlier groups as history), the video
|
|
// is played from its first key frame in a view of its own at the pace it was captured, the
|
|
// audio decoded and counted, and the page then returns to live. What the relay no longer
|
|
// holds, or withholds, comes as an End of Unknown Range, counted and never played.
|
|
const AUDIO_FRAME_MS = Number(q.get("audioFrameMs") || 20);
|
|
let dvrStop = null;
|
|
|
|
function renderDvr() {
|
|
const d = state.dvr;
|
|
const run = d.runs[d.runs.length - 1];
|
|
const canvas = $("dvr-canvas");
|
|
if (canvas) canvas.hidden = d.mode !== "playing";
|
|
if ($("live")) $("live").hidden = d.mode === "live";
|
|
if ($("seek")) $("seek").disabled = d.mode !== "live" || !subscribedVideo();
|
|
const el = $("dvr-status");
|
|
if (!el) return;
|
|
if (!run) { el.textContent = ""; return; }
|
|
const v = run.video, a = run.audio;
|
|
const what = `${run.who}, ${run.seconds} s back: video ${v.verified} of ${v.fetched} verified, audio ${a.verified} of ${a.fetched}${v.unknown + a.unknown ? `; ${v.unknown + a.unknown} range(s) the relay no longer holds` : ""}`;
|
|
el.textContent = d.mode === "seeking" ? `Fetching ${what}` : d.mode === "playing" ? `Playing ${what}; ${v.played} frames shown` : `Back to live after ${what}; ${v.played} frames shown`;
|
|
}
|
|
|
|
function subscribedVideo() {
|
|
return [...subscribed.values()].find((s) => s.tracks.video && tally(s.principal).video.last_live_group != null);
|
|
}
|
|
|
|
// FETCH groups `from` to `to` of a track and of its receipts track, the receipts first,
|
|
// each verified; returns the verified objects in group order.
|
|
async function fetchVerified(st, m, from, to, c) {
|
|
const params = { group_order: 1, subscriber_priority: SUB_PRIORITY[m] + 2, location_filter: { type: "absolute_group_end", start: [from, 0], end_group_delta: to - from } };
|
|
const rec = [];
|
|
const r1 = await session.fetch(st.ns, st.tr.wai_receipts_track, params, (j, p) => rec.push([j.group_id, p]));
|
|
for (const [g, p] of rec) {
|
|
const r = JSON.parse(st.verifier.receipts(BigInt(g), p, BigInt(Date.now())));
|
|
if (!r.ok) c.receipts_refused++;
|
|
}
|
|
const objs = [];
|
|
const r2 = await session.fetch(st.ns, st.tr.name, params, (j, p) => objs.push([j.group_id, j.object_id, p]));
|
|
c.unknown += r1.ranges.length + r2.ranges.length;
|
|
const ok = [];
|
|
for (const [g, o, p] of objs) {
|
|
c.fetched++;
|
|
const r = JSON.parse(st.verifier.object(BigInt(g), BigInt(o), p));
|
|
if (r.ok) { c.verified++; ok.push([g, p]); } else { c.refused++; log(`seek back: ${m} group ${g} refused: ${r.reason}`); }
|
|
}
|
|
return ok;
|
|
}
|
|
|
|
async function seekBack(seconds = 10) {
|
|
if (state.dvr.mode !== "live") return state.dvr.runs[state.dvr.runs.length - 1];
|
|
const sub = subscribedVideo();
|
|
if (!sub) throw new Error("no participant's video to seek back into");
|
|
const counts = () => ({ from: null, to: null, fetched: 0, verified: 0, refused: 0, receipts_refused: 0, unknown: 0, decoded: 0, played: 0 });
|
|
const run = { principal: sub.principal, who: nameOf(sub.principal), seconds, video: counts(), audio: counts(), started_at: performance.now(), playing_at: null, live_at: null, error: null };
|
|
state.dvr.runs.push(run);
|
|
state.dvr.mode = "seeking";
|
|
renderDvr();
|
|
try {
|
|
const got = {};
|
|
for (const m of ["video", "audio"]) {
|
|
const st = sub.tracks[m];
|
|
const T = tally(sub.principal)[m];
|
|
if (!st || T.last_live_group == null) continue;
|
|
const perSecond = m === "video" ? Number(st.tr.framerate || 30) : 1000 / AUDIO_FRAME_MS;
|
|
const to = T.last_live_group;
|
|
const from = Math.max(0, to - Math.round(seconds * perSecond));
|
|
run[m].from = from;
|
|
run[m].to = to;
|
|
got[m] = await fetchVerified(st, m, from, to, run[m]);
|
|
}
|
|
log(`seek back ${seconds} s into ${run.who}: video groups ${run.video.from}..${run.video.to}, ${run.video.verified} of ${run.video.fetched} verified; audio ${run.audio.verified} of ${run.audio.fetched}; ${run.video.unknown + run.audio.unknown} unknown range(s)`);
|
|
state.dvr.mode = "playing";
|
|
run.playing_at = performance.now();
|
|
renderDvr();
|
|
await playBack(sub, got, run);
|
|
} catch (e) {
|
|
run.error = String(e.message || e);
|
|
log(`seek back: ${run.error}`);
|
|
} finally {
|
|
backToLive();
|
|
}
|
|
return run;
|
|
}
|
|
|
|
// The fetched video, decoded and drawn at the pace of its timestamps from its first key
|
|
// frame; the audio decoded alongside and counted.
|
|
async function playBack(sub, got, run) {
|
|
const canvas = $("dvr-canvas");
|
|
const ctx = canvas && canvas.getContext("2d");
|
|
const vst = sub.tracks.video;
|
|
const vdec = new VideoDecoder({
|
|
output: (vf) => {
|
|
run.video.decoded++;
|
|
if (ctx && state.dvr.mode === "playing") { ctx.drawImage(vf, 0, 0, canvas.width, canvas.height); run.video.played++; renderDvr(); }
|
|
vf.close();
|
|
},
|
|
error: (e) => log(`seek back: video decoder: ${e.message || e}`),
|
|
});
|
|
vdec.configure({ codec: vst.tr.codec, codedWidth: vst.tr.width, codedHeight: vst.tr.height });
|
|
const ast = sub.tracks.audio;
|
|
let adec = null;
|
|
if (ast && got.audio) {
|
|
adec = new AudioDecoder({ output: (ad) => { run.audio.decoded++; ad.close(); }, error: (e) => log(`seek back: audio decoder: ${e.message || e}`) });
|
|
adec.configure({ codec: "opus", sampleRate: ast.tr.samplerate, numberOfChannels: Number(ast.tr.channelConfig || 1) });
|
|
for (const [, env] of got.audio) {
|
|
try {
|
|
const e = wai.unpack(env);
|
|
const manifest = JSON.parse(e.manifest);
|
|
const payload = e.payload();
|
|
e.free();
|
|
for (const pk of media.oggOpusPackets(payload)) adec.decode(new EncodedAudioChunk({ type: "key", timestamp: manifest.target.timestamp_us, data: pk }));
|
|
} catch (err) { log(`seek back: audio: ${err.message || err}`); }
|
|
}
|
|
}
|
|
const frames = [];
|
|
for (const [, env] of got.video || []) {
|
|
try {
|
|
const e = wai.unpack(env);
|
|
const manifest = JSON.parse(e.manifest);
|
|
const cap = e.capability;
|
|
const payload = e.payload();
|
|
e.free();
|
|
const units = cap === "wai.video.vp9" ? media.ivfFrames(payload).map((f) => f.data) : media.av1Units(payload);
|
|
frames.push({ key: !!manifest.target.key, ts: manifest.target.timestamp_us, units });
|
|
} catch (err) { log(`seek back: video: ${err.message || err}`); }
|
|
}
|
|
const first = frames.findIndex((f) => f.key);
|
|
let stopped = false;
|
|
dvrStop = () => { stopped = true; };
|
|
if (first >= 0) {
|
|
const t0 = performance.now();
|
|
const ts0 = frames[first].ts;
|
|
for (const f of frames.slice(first)) {
|
|
if (stopped) break;
|
|
const due = t0 + (f.ts - ts0) / 1000;
|
|
const wait = due - performance.now();
|
|
if (wait > 0) await new Promise((r) => setTimeout(r, Math.min(wait, 1000)));
|
|
for (const u of f.units) vdec.decode(new EncodedVideoChunk({ type: f.key ? "key" : "delta", timestamp: f.ts, data: u }));
|
|
}
|
|
}
|
|
try { await vdec.flush(); } catch { /* closed */ }
|
|
if (adec) { try { await adec.flush(); } catch { /* closed */ } adec.close(); }
|
|
vdec.close();
|
|
}
|
|
|
|
function backToLive() {
|
|
if (dvrStop) { dvrStop(); dvrStop = null; }
|
|
const run = state.dvr.runs[state.dvr.runs.length - 1];
|
|
if (run && run.live_at == null) run.live_at = performance.now();
|
|
state.dvr.mode = "live";
|
|
renderDvr();
|
|
}
|
|
|
|
window.roomTest.seekBack = (seconds) => seekBack(seconds);
|
|
if ($("seek")) {
|
|
$("seek").addEventListener("click", () => {
|
|
const s = Math.max(1, Math.min(900, Number(($("seek-s") || {}).value || 10)));
|
|
seekBack(s).catch(fail);
|
|
});
|
|
}
|
|
if ($("live")) $("live").addEventListener("click", () => { if (dvrStop) dvrStop(); });
|
|
setInterval(() => { if (state.dvr.mode === "live") renderDvr(); }, 1000);
|
|
|
|
// ------------------------------------------------------------------------------- GOAWAY
|
|
|
|
// GOAWAY (MOQT §9.2): no new request on the old session; a new session to the URI it
|
|
// names (or the same one), the tracks published again and every subscription made again
|
|
// there, then the old session closed by this end, before the relay's timeout.
|
|
async function onGoaway(old, g) {
|
|
state.goaways.push({ uri: g.uri, timeout_ms: g.timeout_ms, at: performance.now() });
|
|
try {
|
|
const opts = { ...transportOpts };
|
|
if (g.uri) {
|
|
if (g.uri.startsWith("https://")) { opts.webtransport = g.uri; opts.force = "webtransport"; }
|
|
else if (g.uri.startsWith("ws")) { opts.websocket = g.uri; opts.force = "websocket"; }
|
|
}
|
|
transportOpts = opts;
|
|
const s = await openSession(opts);
|
|
if (window.roomTest.role !== "clock") await publishTracks(s);
|
|
// This end's tracks now go to the new session, so the old one closes, by this end and
|
|
// before the relay's timeout; then every subscription is made again there.
|
|
old.close(0, "moved after GOAWAY");
|
|
state.goaways[state.goaways.length - 1].moved = { transport: s.t.kind, at: performance.now() };
|
|
log(`moved to ${g.uri || "the same relay"} after GOAWAY`);
|
|
if (window.roomTest.role !== "clock") {
|
|
const others = [...subscribed.keys()];
|
|
subscribed.clear();
|
|
await subscribeEvents(fx.clock.principal);
|
|
for (const P of others) await subscribeParticipant(P);
|
|
log(`subscriptions made again on the new session: ${others.length} participant(s)`);
|
|
}
|
|
} catch (e) {
|
|
fail(e);
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------- the views
|
|
|
|
function renderTimeline(e) {
|
|
const li = document.createElement("li");
|
|
li.dataset.kind = e.kind;
|
|
const p = e.payload || {};
|
|
const what = e.kind === "caption" ? `“${p.text}” (${p.origin_label})` : e.kind === "chat" ? p.text : p.detail ? `${(p.subject || "").slice(0, 8)} ${JSON.stringify(p.detail)}` : "";
|
|
li.textContent = `t=${(e.t / 1e6).toFixed(3)}s ${e.who} ${e.kind} ${what}`;
|
|
$("timeline").appendChild(li);
|
|
}
|
|
|
|
function renderCaption() {
|
|
const c = state.captions[state.captions.length - 1];
|
|
const el = $("caption");
|
|
el.innerHTML = "";
|
|
const label = document.createElement("span");
|
|
label.className = `origin origin-${c.origin}`;
|
|
label.textContent = c.origin === "human" ? "Human captioner" : c.origin === "machine" ? "Machine-generated" : "Machine-generated, reviewed";
|
|
const text = document.createElement("span");
|
|
text.className = "text";
|
|
text.textContent = c.text;
|
|
el.append(label, text);
|
|
el.dataset.origin = c.origin;
|
|
}
|
|
|
|
// A participant's own events: captions and chats, signed with its key.
|
|
window.roomTest.sendCaption = async ({ text, origin = 1, model = null, lang = "en", duration_ms = 3000 }) => {
|
|
const p = wai.captionPayload(duration_ms, origin, model || undefined, lang, text);
|
|
const r = await eventsTrack.emit(0x0a, p);
|
|
await (records = records.then(() => onRecord(r.record)));
|
|
return r.view;
|
|
};
|
|
window.roomTest.sendChat = async (text) => {
|
|
const r = await eventsTrack.emit(0x03, wai.chatPayload(text));
|
|
await (records = records.then(() => onRecord(r.record)));
|
|
return r.view;
|
|
};
|
|
window.roomTest.snapshot = () => ({
|
|
phase: state.phase,
|
|
mode: state.mode,
|
|
me: me && { principal: me.principal, role: me.role || window.roomTest.role },
|
|
room: fx && fx.room,
|
|
grant: state.grant,
|
|
ended: state.ended,
|
|
errors: state.errors,
|
|
transport: state.transport,
|
|
sessions: state.sessions,
|
|
published: state.published,
|
|
codec: state.publishCodec,
|
|
capture: state.capture,
|
|
subs: state.subs,
|
|
dvr: state.dvr,
|
|
captions: state.captions,
|
|
timeline: state.timeline.map((e) => ({ t: e.t, who: e.who, kind: e.kind })),
|
|
goaways: state.goaways,
|
|
roomState: state.room,
|
|
frame: compositor && compositor.frame,
|
|
layouts: compositor && compositor.layouts,
|
|
drawn: compositor && Object.fromEntries(compositor.drawn),
|
|
draws: compositor && compositor.draws,
|
|
samples: compositor && compositor.samples(),
|
|
mixed: compositor && compositor.mixed,
|
|
worklet: compositor && compositor.workletStats,
|
|
audioState: compositor && compositor.ac && compositor.ac.state,
|
|
session: session && { counters: session.counters, transport: session.t.stats ? session.t.stats() : {} },
|
|
verifiers: Object.fromEntries([...subscribed.entries()].map(([P, s]) => [P, Object.fromEntries(Object.entries(s.tracks).map(([m, st]) => [m, JSON.parse(st.verifier.stats())]))])),
|
|
});
|
|
// A participant's grant presented as a bare token, with no proof: the relay must refuse
|
|
// the session (live-rooms §10.5.2, UNAUTHORIZED). Resolves with how the session ended.
|
|
window.roomTest.tryBareGrant = async () => {
|
|
const t = await connect({ ...transportOpts, log: () => {} });
|
|
let setup = "exchanged";
|
|
try {
|
|
await MoqSession.connect(t, moq, { token: unhex(me.grant), implementation: "joule-moq-web/room-bare", log: () => {} });
|
|
} catch (e) {
|
|
setup = String(e.message || e);
|
|
}
|
|
const closed = await Promise.race([t.closed, new Promise((r) => setTimeout(() => r(null), 8000))]);
|
|
return { transport: t.kind, setup, closed };
|
|
};
|
|
|
|
// This page's grant presented with a proof made by another key: refused the session too
|
|
// (§10.5.4 step 4, pop_wrong_key).
|
|
window.roomTest.tryWrongKey = async () => {
|
|
const t = await connect({ ...transportOpts, log: () => {} });
|
|
let setup = "exchanged";
|
|
try {
|
|
await MoqSession.connect(t, moq, { grant: { record: unhex(me.grant), seed: unhex(me.seed), proofBy: crypto.getRandomValues(new Uint8Array(32)) }, implementation: "joule-moq-web/room-wrong-key", log: () => {} });
|
|
} catch (e) {
|
|
setup = String(e.message || e);
|
|
}
|
|
const closed = await Promise.race([t.closed, new Promise((r) => setTimeout(() => r(null), 8000))]);
|
|
return { transport: t.kind, setup, closed };
|
|
};
|
|
|
|
window.roomTest.roomJson = () => room.roomJson();
|
|
window.roomTest.compose = () => fx && fx.compose;
|
|
window.roomTest.frameAt = (t) => JSON.parse(room.frame(unhex(fx.compose), BigInt(t)));
|
|
window.roomTest.now = () => roomNow();
|
|
|
|
// A human caption, typed by this participant.
|
|
async function sendTypedCaption(e) {
|
|
e.preventDefault();
|
|
const el = $("caption-text");
|
|
const text = el.value.trim();
|
|
if (!text || state.phase !== "publishing") return;
|
|
el.value = "";
|
|
try { await window.roomTest.sendCaption({ text, origin: 1, lang: document.documentElement.lang || "en" }); } catch (err) { fail(err); }
|
|
}
|
|
|
|
service = q.get("service") || ($("room-app") && $("room-app").dataset.service) || "";
|
|
if (service) {
|
|
window.roomTest.enter = (role) => enter(role);
|
|
window.roomTest.leave = leave;
|
|
const buttons = ["join", "watch"].map($).filter(Boolean);
|
|
const go = (role) => async () => {
|
|
for (const b of buttons) b.disabled = true;
|
|
try {
|
|
const phase = await enter(role);
|
|
if (phase === "full") for (const b of buttons) b.disabled = b.id === "join";
|
|
else if ($("leave")) $("leave").hidden = false;
|
|
if (phase === "publishing" && $("caption-form")) $("caption-form").hidden = false;
|
|
} catch (e) {
|
|
fail(e);
|
|
state.mode = state.mode === "service" && state.phase === "entering" ? null : state.mode;
|
|
for (const b of buttons) b.disabled = false;
|
|
}
|
|
};
|
|
if ($("join")) $("join").addEventListener("click", go("publisher"));
|
|
if ($("watch")) $("watch").addEventListener("click", go("viewer"));
|
|
if ($("leave")) $("leave").addEventListener("click", () => { $("leave").disabled = true; if ($("caption-form")) $("caption-form").hidden = true; leave(); });
|
|
if ($("caption-form")) $("caption-form").addEventListener("submit", sendTypedCaption);
|
|
state.phase = "idle";
|
|
} else {
|
|
boot().catch(fail);
|
|
}
|