diff --git a/src/api/clips.js b/src/api/clips.js index edffe30d..84d316ff 100644 --- a/src/api/clips.js +++ b/src/api/clips.js @@ -1,9 +1,8 @@ import { athena as Athena } from '../api'; import localforage from 'localforage'; import { deviceVersionAtLeast } from '../utils'; +import { webrtcConnectionManager } from '../utils/webrtc'; -const CLIP_CHUNK_CONCURRENCY = 3; -const CLIP_RETRY_DELAYS = [500, 1000, 2000, 4000, 8000, 16000]; const activeDownloads = new Map(); const supportRequests = new Map(); const clipStorage = localforage.createInstance({ name: 'connect', storeName: 'clip_cache' }); @@ -20,7 +19,7 @@ async function invalidateClip(dongleId, filename) { const prefix = `clip:${dongleId}/${filename}/`; for (const [key, entry] of activeDownloads.entries()) { if (key.startsWith(prefix)) { - entry.cancelled = true; + entry.controller.abort(); entry.listeners.clear(); activeDownloads.delete(key); } @@ -29,45 +28,6 @@ async function invalidateClip(dongleId, filename) { await Promise.all(keys.filter(key => key.startsWith(prefix)).map(key => clipStorage.removeItem(key))).catch(() => {}); } -async function downloadClip(dongleId, filename, reportProgress, isCancelled) { - const chunks = []; - let loaded = 0; - let size; - let chunkBytes; - let nextOffset = 0; - - while (size === undefined || nextOffset < size) { - if (isCancelled()) throw new Error('Clip download cancelled'); - const offsets = Array.from( - { length: chunkBytes ? Math.min(CLIP_CHUNK_CONCURRENCY, Math.ceil((size - nextOffset) / chunkBytes)) : 1 }, - (_, index) => nextOffset + (index * (chunkBytes || 0)), - ); - let results; - try { - // eslint-disable-next-line no-await-in-loop - results = await Promise.all(offsets.map(offset => getClipChunk(dongleId, filename, offset))); - } catch (error) { - if (isCancelled()) throw new Error('Clip download cancelled'); - throw error; - } - if (isCancelled()) throw new Error('Clip download cancelled'); - for (const result of results) { - if (size === undefined) size = result.size; - if (result.size !== size || result.offset !== nextOffset) throw new Error('Clip changed during download'); - const binary = atob(result.data); - const chunk = Uint8Array.from(binary, character => character.charCodeAt(0)); - if (!chunk.length && nextOffset < size) throw new Error('Clip download returned an empty chunk'); - if (!chunkBytes) chunkBytes = chunk.length; - chunks.push(chunk); - loaded += chunk.length; - nextOffset += chunk.length; - reportProgress(loaded, size); - } - } - if (loaded !== size) throw new Error(`Clip download ended at ${loaded} of ${size} bytes`); - return new Blob(chunks, { type: 'video/mp4' }); -} - async function getClipBlob(dongleId, filename, requestedAt, onProgress) { const key = cacheKey(dongleId, filename, requestedAt); const stored = await clipStorage.getItem(key).catch(() => null); @@ -75,12 +35,12 @@ async function getClipBlob(dongleId, filename, requestedAt, onProgress) { let entry = activeDownloads.get(key); if (!entry) { - entry = { cancelled: false, listeners: new Set(), loaded: 0, total: 0 }; - entry.promise = downloadClip(dongleId, filename, (loaded, total) => { + entry = { controller: new AbortController(), listeners: new Set(), loaded: 0, total: 0 }; + entry.promise = webrtcConnectionManager.downloadClip(dongleId, filename, (loaded, total) => { entry.loaded = loaded; entry.total = total; for (const listener of entry.listeners) listener(loaded, total); - }, () => entry.cancelled).then(async (blob) => { + }, entry.controller.signal).then(async (blob) => { if (activeDownloads.get(key) === entry) await clipStorage.setItem(key, blob).catch(() => {}); return blob; }).finally(() => { @@ -123,16 +83,6 @@ export function deviceSupportsClips(device) { return supportRequests.get(device.dongle_id); } -async function getClipChunk(dongleId, filename, offset, attempt = 0) { - try { - return await call(dongleId, 'getClipChunk', { filename, offset }); - } catch (error) { - if (error.message !== 'Athena request failed' || attempt === CLIP_RETRY_DELAYS.length) throw error; - await new Promise(resolve => setTimeout(resolve, CLIP_RETRY_DELAYS[attempt])); - return getClipChunk(dongleId, filename, offset, attempt + 1); - } -} - export const clipDevice = { async getClipState(dongleId, params) { return call(dongleId, 'getClipState', params); diff --git a/src/utils/webrtc.js b/src/utils/webrtc.js index fbd17a4b..df2937d6 100644 --- a/src/utils/webrtc.js +++ b/src/utils/webrtc.js @@ -10,6 +10,8 @@ const CLOCK_PING_MS = 500; const CONNECTION_DEADLINE_MS = 15000; const ICE_GATHER_DEADLINE_MS = 8000; +const CLIP_BLOB_PART_SIZE = 256 * 1024; +const CLIP_HEADER_SIZE = 17; // Drop mDNS (.local) host candidates from an SDP — the device can't resolve them. function stripMdnsCandidates(sdp) { @@ -40,6 +42,9 @@ export class WebRTCConnection extends EventTarget { this.clockSynced = false; this.connectStartedAt = null; this.transformWorkers = []; + this.clipTransfers = new Map(); + this.clipTransferReady = false; + this.dataMessageQueue = Promise.resolve(); this.videoEnabled = false; this.connectionState = 'new'; this.failReason = null; @@ -138,8 +143,10 @@ export class WebRTCConnection extends EventTarget { // set up data channel this.dc = this.pc.createDataChannel('data', { ordered: true }); + this.dc.binaryType = 'arraybuffer'; this.dc.onopen = () => { this._log('Data channel open'); + this.dispatchEvent(new Event('datachannelopen')); if (this.videoEnabled) { this._sendDc('livestreamVideoEnable', { enabled: true }); this.enableJoystick(true); @@ -150,15 +157,14 @@ export class WebRTCConnection extends EventTarget { this._stopClockSync(); }; this.dc.onmessage = (evt) => { - try { - const msg = JSON.parse(typeof evt.data === 'string' ? evt.data : new TextDecoder().decode(evt.data)); - if (msg.type === 'carState') this.callbacks.onBatteryLevel({ level: Math.round(msg.data.fuelGauge * 100), charging: !!msg.data.charging }); - if (msg.type === 'deviceState') this.callbacks.onIgnition?.(!!msg.data?.started); - if (msg.type === 'disconnect') this.disconnect(msg.data || 'Connection replaced by another device.'); - if (msg.type === 'clockSync' && msg.data?.action === 'pong') this._handleClockPong(msg.data); - } catch (e) { - console.warn('webrtc: ignoring malformed data-channel message', e); - } + this.dataMessageQueue = this.dataMessageQueue.then(async () => { + if (typeof evt.data !== 'string') { + const buffer = evt.data instanceof ArrayBuffer ? evt.data : await evt.data.arrayBuffer(); + this._handleClipChunk(buffer); + } else { + this._handleDataMessage(JSON.parse(evt.data)); + } + }).catch(e => console.warn('webrtc: ignoring malformed data-channel message', e)); }; const offer = await this.pc.createOffer(); @@ -232,6 +238,125 @@ export class WebRTCConnection extends EventTarget { return false; } + async downloadClip(filename, onProgress, signal) { + if (signal?.aborted) throw new Error('Clip download cancelled'); + if (this.dc?.readyState !== 'open') await this._waitForDataChannel(signal); + if (!this.clipTransferReady) await this._waitForEvent('cliptransferready', signal); + + const id = crypto.randomUUID().replaceAll('-', ''); + return new Promise((resolve, reject) => { + const transfer = { chunks: [], pendingChunks: [], pendingBytes: 0, loaded: 0, size: null, onProgress, resolve, reject }; + const abort = () => { + this._sendDc('clipTransferCancel', { id }); + this._failClipTransfer(id, 'Clip download cancelled'); + }; + transfer.abort = abort; + transfer.signal = signal; + signal?.addEventListener('abort', abort, { once: true }); + this.clipTransfers.set(id, transfer); + if (signal?.aborted) abort(); + else if (!this._sendDc('clipTransferStart', { id, filename })) this._failClipTransfer(id, 'Connection lost'); + }); + } + + _waitForDataChannel(signal) { + return this._waitForEvent('datachannelopen', signal); + } + + _waitForEvent(name, signal) { + return new Promise((resolve, reject) => { + const finish = (error) => { + clearTimeout(timeout); + this.removeEventListener(name, onReady); + signal?.removeEventListener('abort', onAbort); + if (error) reject(error); + else resolve(); + }; + const onReady = () => finish(); + const onAbort = () => finish(new Error('Clip download cancelled')); + const timeout = setTimeout(() => finish(new Error(this.failReason || 'Could not connect to device')), CONNECTION_DEADLINE_MS); + this.addEventListener(name, onReady, { once: true }); + signal?.addEventListener('abort', onAbort, { once: true }); + if (signal?.aborted) onAbort(); + }); + } + + _startClipTransfer({ id, size }) { + const transfer = this.clipTransfers.get(id); + if (!transfer) return; + if (!Number.isSafeInteger(size) || size < 0) { + this._failClipTransfer(id, 'Device returned an invalid clip size'); + return; + } + transfer.size = size; + transfer.onProgress?.(0, size); + } + + _handleDataMessage(msg) { + if (msg.type === 'carState') this.callbacks.onBatteryLevel({ level: Math.round(msg.data.fuelGauge * 100), charging: !!msg.data.charging }); + if (msg.type === 'deviceState') this.callbacks.onIgnition?.(!!msg.data?.started); + if (msg.type === 'disconnect') this.disconnect(msg.data || 'Connection replaced by another device.'); + if (msg.type === 'clockSync' && msg.data?.action === 'pong') this._handleClockPong(msg.data); + if (msg.type === 'clipTransferReady') { + this.clipTransferReady = true; + this.dispatchEvent(new Event('cliptransferready')); + } + if (msg.type === 'clipTransferStart') this._startClipTransfer(msg.data); + if (msg.type === 'clipTransferEnd') this._finishClipTransfer(msg.data.id); + if (msg.type === 'clipTransferError') this._failClipTransfer(msg.data.id, msg.data.message); + } + + _handleClipChunk(buffer) { + const bytes = new Uint8Array(buffer); + if (bytes.length <= CLIP_HEADER_SIZE || bytes[0] !== 1) return; + const id = Array.from(bytes.subarray(1, CLIP_HEADER_SIZE), byte => byte.toString(16).padStart(2, '0')).join(''); + const transfer = this.clipTransfers.get(id); + if (!transfer || transfer.size === null) return; + const chunk = bytes.slice(CLIP_HEADER_SIZE); + transfer.pendingChunks.push(chunk); + transfer.pendingBytes += chunk.length; + transfer.loaded += chunk.length; + if (transfer.loaded > transfer.size) { + this._failClipTransfer(id, 'Device sent more data than expected'); + return; + } + if (transfer.pendingBytes >= CLIP_BLOB_PART_SIZE) this._flushClipChunks(transfer); + transfer.onProgress?.(transfer.loaded, transfer.size); + } + + _flushClipChunks(transfer) { + const part = new Uint8Array(transfer.pendingBytes); + let offset = 0; + for (const chunk of transfer.pendingChunks) { + part.set(chunk, offset); + offset += chunk.length; + } + transfer.chunks.push(part); + transfer.pendingChunks = []; + transfer.pendingBytes = 0; + } + + _finishClipTransfer(id) { + const transfer = this.clipTransfers.get(id); + if (!transfer) return; + if (transfer.loaded !== transfer.size) { + this._failClipTransfer(id, `Clip download ended at ${transfer.loaded} of ${transfer.size} bytes`); + return; + } + if (transfer.pendingBytes) this._flushClipChunks(transfer); + this.clipTransfers.delete(id); + transfer.signal?.removeEventListener('abort', transfer.abort); + transfer.resolve(new Blob(transfer.chunks, { type: 'video/mp4' })); + } + + _failClipTransfer(id, message) { + const transfer = this.clipTransfers.get(id); + if (!transfer) return; + this.clipTransfers.delete(id); + transfer.signal?.removeEventListener('abort', transfer.abort); + transfer.reject(new Error(message || 'Clip download failed')); + } + enableVideo(enabled) { this.videoEnabled = enabled; this._sendDc('livestreamVideoEnable', { enabled }); @@ -347,11 +472,14 @@ export class WebRTCConnection extends EventTarget { cleanup() { this._clearConnectionTimeout(); + this.clipTransferReady = false; + this.dataMessageQueue = Promise.resolve(); this.enableJoystick(false); this._stopClockSync(); for (const worker of this.transformWorkers.splice(0)) { worker.terminate(); } + for (const id of Array.from(this.clipTransfers.keys())) this._failClipTransfer(id, 'Connection lost'); if (this.dc) { this.dc.onopen = null; this.dc.onclose = null; @@ -522,6 +650,11 @@ export class WebRTCConnectionManager { this.connection.enableJoystick(enabled); } } + + downloadClip(dongleId, filename, onProgress, signal) { + if (!this._healthy(dongleId)) this._open(dongleId); + return this.connection.downloadClip(filename, onProgress, signal); + } } export const webrtcConnectionManager = new WebRTCConnectionManager();