diff options
Diffstat (limited to 'packages/meshbay-client/src')
| -rw-r--r-- | packages/meshbay-client/src/cast-relay.js | 184 |
1 files changed, 148 insertions, 36 deletions
diff --git a/packages/meshbay-client/src/cast-relay.js b/packages/meshbay-client/src/cast-relay.js index ed20b337..e51d52eb 100644 --- a/packages/meshbay-client/src/cast-relay.js +++ b/packages/meshbay-client/src/cast-relay.js @@ -56,6 +56,9 @@ const util = require('node:util'); const debug = util.debuglog('cast-relay'); const RING_CAP = 64; +// A fragment is a whole segment, megabytes at a film's bitrate: 64 of them +// can be hundreds of megabytes, so the backlog is bounded in bytes too. +const RING_MAX_BYTES = 32 * 1024 * 1024; // What the receiver is ever told a photo is. The page re-encodes every photo // to JPEG, so this list is a guard, not a menu. const FILE_TYPES = new Set([ @@ -66,7 +69,16 @@ const FILE_TYPES = new Set([ // A track is a few megabytes, a long lossless one a few hundred; this only // stops a page from filling the machine's memory. const FILE_MAX_BYTES = 1024 * 1024 * 1024; -const BACKPRESSURE_HIGH = 8 * 1024 * 1024; +// How far a receiver may fall behind before fragments are dropped for it. +// The page runs ahead of the television by its whole read-ahead (up to 48 MB, +// video-player.js) plus the receiver's start-up, and a fragment is a few +// seconds of picture: the 8 MB this used to be dropped one every few seconds +// from a receiver reading at playback speed, and each froze the picture for +// its length. Node holds what waits in the response's own buffer. +const CLIENT_LEAD_MAX = 512 * 1024 * 1024; +// A receiver that connects before the first moof waits for the whole header. +const HEADER_WAIT_MS = 10000; +const FTYP = 0x66747970; const MOOF = 0x6d6f6f66; const PORT_BASE = 19550; const PORT_COUNT = 4; @@ -97,6 +109,10 @@ class BoxAccumulator { constructor() { this._buf = Buffer.alloc(0); this._synced = false; + this._preambleSeen = false; + // Called once, with every byte before the first moof: the stream's + // header (ftyp, moov), however many pushes it came in. + this.onPreamble = null; } push(data) { @@ -106,7 +122,11 @@ class BoxAccumulator { if (!this._synced) { const idx = this._findMoof(); if (idx === -1) return fragments; - debug(`[cast-relay] box sync: found first moof at byte offset ${idx}, discarded ${idx} bytes`); + debug(`[cast-relay] box sync: found first moof at byte offset ${idx}`); + if (!this._preambleSeen) { + this._preambleSeen = true; + if (this.onPreamble) this.onPreamble(Buffer.from(this._buf.subarray(0, idx))); + } this._buf = this._buf.subarray(idx); this._synced = true; } @@ -149,6 +169,7 @@ class BoxAccumulator { reset() { this._buf = Buffer.alloc(0); this._synced = false; + this._preambleSeen = false; } _findMoof() { @@ -164,6 +185,29 @@ class BoxAccumulator { } } +/** + * The header is what comes before the first moof, and only that. + * + * The page hands over the first chunk it decrypted, which is a slice of the + * stream and not a box: measured, it was by turns the 28-byte ftyp alone, the + * header exactly, or the header and the start of the first fragment, + * depending on when the node read ffmpeg's output. Served as the header, the + * first left the receiver without a moov and the third broke the box + * structure, so the same film cast one try in three. The pushed stream + * carries the whole header; this trims the page's chunk for the case where + * nothing came before the first moof. + */ +function headerOnly(init) { + let at = 0; + while (at + 8 <= init.length) { + if (init.readUInt32BE(at + 4) === MOOF) return init.subarray(0, at); + const size = init.readUInt32BE(at); + if (size < 8) break; + at += size; + } + return init; +} + class CastRelay { constructor() { this._server = null; @@ -171,12 +215,16 @@ class CastRelay { this._token = null; this._lanIP = null; this._initSegment = null; + this._headerReady = false; + this._waiting = new Set(); this._ring = []; + this._ringBytes = 0; this._clients = new Set(); this._accum = new BoxAccumulator(); this._fragCount = 0; this._chunkCount = 0; this._bytesSent = 0; + this._dropped = 0; this._subtitle = null; this._subtitleVersion = 0; this._file = null; @@ -303,21 +351,20 @@ class CastRelay { } async start({ codec, initSegment, subtitle }) { - if (this._server) await this.stop(); + // Everything the new stream needs is set before the first await: the + // page's pushes arrive while this waits for its port, and they belong to + // the new stream's accumulator, header first. + const closing = this._teardown(); this._token = crypto.randomBytes(16).toString('hex'); this._lanIP = lanAddress(); - this._initSegment = initSegment ? Buffer.from(initSegment) : null; - this._ring = []; - this._clients = new Set(); - this._accum = new BoxAccumulator(); - this._fragCount = 0; - this._chunkCount = 0; - this._bytesSent = 0; - this._subtitle = null; - this._file = null; - this._incoming = null; + this._initSegment = initSegment ? headerOnly(Buffer.from(initSegment)) : null; + // Nothing to wait for without a first chunk: no header is coming. + this._headerReady = !initSegment; + this._accum.onPreamble = (bytes) => this._preamble(bytes); this.setSubtitle(subtitle); + const token = this._token; + await closing; const server = http.createServer((req, res) => this._handle(req, res)); @@ -333,19 +380,23 @@ class CastRelay { server.on('error', onError); server.listen(port, this._lanIP, () => { server.removeListener('error', onError); - this._port = port; resolve(); }); }); - bound = true; + bound = port; } catch (err) { if (err.code !== 'EADDRINUSE') throw err; debug(`[cast-relay] port ${port} busy, trying next`); } } if (!bound) throw new Error('All cast relay ports are in use'); + if (this._token !== token) { + server.close(); + throw new Error('Cast relay stopped while starting'); + } this._server = server; + this._port = bound; debug(`[cast-relay] started on ${this.url}`); debug(`[cast-relay] init segment: ${this._initSegment ? this._initSegment.length + ' bytes' : 'none'}`); return { @@ -356,6 +407,18 @@ class CastRelay { }; } + /** + * Every byte the page pushed before the first moof. Kept as the header when + * it starts with an ftyp; the page's own first chunk, trimmed, otherwise. + */ + _preamble(bytes) { + if (bytes.length >= 8 && bytes.readUInt32BE(4) === FTYP) this._initSegment = bytes; + debug(`[cast-relay] header complete: ${this._initSegment ? this._initSegment.length : 0} bytes`); + this._headerReady = true; + for (const res of this._waiting) this._admit(res); + this._waiting.clear(); + } + pushSegment(data) { const buf = Buffer.isBuffer(data) ? data : Buffer.from(data); this._chunkCount++; @@ -368,14 +431,22 @@ class CastRelay { debug(`[cast-relay] fragment #${this._fragCount}: ${frag.length} bytes (from ${this._chunkCount} chunks), ${this._clients.size} client(s)`); } - if (this._ring.length >= RING_CAP) { - this._ring.shift(); - } this._ring.push(frag); + this._ringBytes += frag.length; + while (this._ring.length > RING_CAP + || (this._ringBytes > RING_MAX_BYTES && this._ring.length > 1)) { + this._ringBytes -= this._ring.shift().length; + } for (const res of this._clients) { - if (res.writableLength > BACKPRESSURE_HIGH) { - debug(`[cast-relay] backpressure: dropping fragment for slow client`); + if (res.writableLength > CLIENT_LEAD_MAX) { + // The picture on the television freezes for this fragment's length. + this._dropped++; + if (this._dropped <= 3 || this._dropped % 20 === 0) { + debug(`[cast-relay] dropped a fragment of ${frag.length} bytes for a` + + ` receiver ${(res.writableLength / 1048576).toFixed(0)} MB behind,` + + ` ${this._dropped} dropped so far`); + } continue; } res.write(frag); @@ -386,35 +457,55 @@ class CastRelay { finish() { debug('[cast-relay] finishing stream'); + for (const res of this._waiting) this._admit(res); + this._waiting.clear(); for (const res of this._clients) { try { res.end(); } catch { /* already closed */ } } } async stop() { - debug(`[cast-relay] stopping — ${this._chunkCount} chunks, ${this._fragCount} fragments, ${(this._bytesSent / 1048576).toFixed(1)} MB sent`); - for (const res of this._clients) { - try { res.end(); } catch { /* already closed */ } - } - this._clients.clear(); + await this._teardown(); + } - if (this._server) { - const srv = this._server; - this._server = null; - await new Promise((resolve) => srv.close(resolve)); + /** + * Everything is reset before anything is awaited, and the promise only says + * when the old port is closed. The page calls stop() without waiting and + * start() next: a stop that cleared its state after the close cleared the + * new stream's instead (its token, port and header) and the receiver was + * handed `:null/stream.mp4?t=null`. The close used to wait, too, for the + * receiver to let go of the old stream, which it does only once loaded with + * the new one; its connections are cut instead. + */ + _teardown() { + debug(`[cast-relay] stopping: ${this._chunkCount} chunks, ${this._fragCount} fragments, ${(this._bytesSent / 1048576).toFixed(1)} MB sent, ${this._dropped} dropped`); + for (const res of [...this._clients, ...this._waiting]) { + try { res.end(); } catch { /* already closed */ } } + this._clients = new Set(); + this._waiting = new Set(); + const srv = this._server; + this._server = null; this._port = null; this._token = null; this._initSegment = null; + this._headerReady = false; this._subtitle = null; this._file = null; this._incoming = null; this._ring = []; - this._accum.reset(); + this._ringBytes = 0; + this._accum = new BoxAccumulator(); this._fragCount = 0; this._chunkCount = 0; this._bytesSent = 0; + this._dropped = 0; + + if (!srv) return Promise.resolve(); + const closed = new Promise((resolve) => srv.close(resolve)); + srv.closeAllConnections(); + return closed; } _handle(req, res) { @@ -469,7 +560,6 @@ class CastRelay { return; } - let sent = 0; res.writeHead(200, { // A receiver that has been given a side-loaded track reads the media // through the same CORS-checked path as the track, so the headers go on @@ -482,6 +572,31 @@ class CastRelay { 'Connection': 'keep-alive', }); + res.flushHeaders(); + req.on('close', () => { + this._clients.delete(res); + this._waiting.delete(res); + debug(`[cast-relay] client disconnected, ${this._clients.size} remaining`); + }); + + if (this._headerReady) { + this._admit(res); + return; + } + // Before the first moof the header may be a piece of itself: the + // receiver waits for all of it, or for the page's chunk at worst. + this._waiting.add(res); + const token = this._token; + setTimeout(() => { + if (this._token !== token || !this._waiting.delete(res)) return; + debug('[cast-relay] no moof yet, serving the first chunk as the header'); + this._admit(res); + }, HEADER_WAIT_MS).unref(); + } + + /** The header, the backlog, then live fragments from here on. */ + _admit(res) { + let sent = 0; if (this._initSegment) { res.write(this._initSegment); sent += this._initSegment.length; @@ -492,13 +607,9 @@ class CastRelay { sent += frag.length; } - debug(`[cast-relay] client connected from ${req.socket.remoteAddress} — sent init + ${this._ring.length} fragments (${(sent / 1024).toFixed(0)} KB)`); + debug(`[cast-relay] client connected from ${res.socket && res.socket.remoteAddress}: sent init + ${this._ring.length} fragments (${(sent / 1024).toFixed(0)} KB)`); this._clients.add(res); - req.on('close', () => { - this._clients.delete(res); - debug(`[cast-relay] client disconnected, ${this._clients.size} remaining`); - }); } _handleSubtitle(res) { @@ -573,3 +684,4 @@ class CastRelay { } module.exports = CastRelay; +module.exports.headerOnly = headerOnly; |