diff --git a/README.md b/README.md index e624b44..c07cf41 100644 --- a/README.md +++ b/README.md @@ -278,16 +278,12 @@ The following features are planned for upcoming releases: - Automatic aspect ratio detection - Picture-in-Picture (PiP) -- Latency retention for asynchronous renditions - CEA-608 closed captions - VOD playback (DVR support) - VOD thumbnail previews - SEI timecodes support - WebTransport protocol -- Nimble Advertizer integration -- Sync mode - Screenshot capture -- Splash/startup image - Extended Player API - OffscreenCanvas rendering - Resume from pause in DVR mode (no auto-jump to live) diff --git a/index.html b/index.html index adeece8..41b7a23 100644 --- a/index.html +++ b/index.html @@ -58,6 +58,7 @@ }, workletLogs: true, latencyAdjustMethod: "fast-forward", // "fast-forward" | "seek" + syncBuffer: 3000, }); window.nimio.on("nimio:play", (inst, contId) => { diff --git a/package-lock.json b/package-lock.json index 1c68751..e23c1f7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11,7 +11,7 @@ "devDependencies": { "@vitest/coverage-v8": "^3.2.4", "jsdom": "^26.1.0", - "prettier": "^3.5.3", + "prettier": "^3.8.1", "rollup-plugin-copy": "^3.5.0", "tseep": "^1.3.1", "vite": "^6.4.1", @@ -2293,11 +2293,10 @@ } }, "node_modules/prettier": { - "version": "3.5.3", - "resolved": "https://registry.npmjs.org/prettier/-/prettier-3.5.3.tgz", - "integrity": "sha512-QQtaxnoDJeAkDvDKWCLiwIXkTgRhwYDEQCghU9Z6q03iyek/rxRh/2lC3HB7P8sWT2xC/y5JDctPLBIGzHKbhw==", + "version": "3.8.1", + "resolved": "https://registry.npmjs.org/prettier/-/prettier-3.8.1.tgz", + "integrity": "sha512-UOnG6LftzbdaHZcKoPFtOcCKztrQ57WkHDeRD9t/PTQtmT0NHSeWWepj6pS0z/N7+08BHFDQVUrfmfMRcZwbMg==", "dev": true, - "license": "MIT", "bin": { "prettier": "bin/prettier.cjs" }, diff --git a/package.json b/package.json index d3d2c10..c469a7c 100644 --- a/package.json +++ b/package.json @@ -19,7 +19,7 @@ "devDependencies": { "@vitest/coverage-v8": "^3.2.4", "jsdom": "^26.1.0", - "prettier": "^3.5.3", + "prettier": "^3.8.1", "rollup-plugin-copy": "^3.5.0", "tseep": "^1.3.1", "vite": "^6.4.1", diff --git a/src/advertizer/evaluator.js b/src/advertizer/evaluator.js new file mode 100644 index 0000000..a75daa8 --- /dev/null +++ b/src/advertizer/evaluator.js @@ -0,0 +1,144 @@ +import { LoggersFactory } from "@/shared/logger"; + +export class AdvertizerEvaluator { + #types = []; + #confIvalUs = 2_000_000; + + constructor(instName, port) { + this._tracks = {}; + this._switches = {}; + this._swCnt = 0; + + if (port) { + port.addEventListener("message", this._portMessageHandler.bind(this)); + port.postMessage("transp-discont-eval-ready"); + } else { + this._pendingActions = []; + } + + this._logger = LoggersFactory.create(instName, "Advertizer Eval", port); + } + + reset() { + this.clearPendingActions(); + } + + isApplicable() { + return this._swCnt > 0; + } + + computeShift(curTsUs, availUs) { + // skip for a while if a track is being switched + if (this._tracks.video === null || this._tracks.audio === null) return 0; + + let res = 0; + let preMatches = []; + let win = [Infinity, 0]; + let del = {}; + for (let j = 0; j < this.#types.length; j++) { + let trackId = this._tracks[this.#types[j]]; + let switches = this._switches[trackId]; + if (!switches) break; + + for (let i = 0; i < switches.length; i++) { + if (switches[i].to < curTsUs) { + del[trackId] = i; + continue; + } + if (switches[i].from - curTsUs > this.#confIvalUs) { + break; + } + if (win[0] > switches[i].from || win[0] < curTsUs) { + win[0] = switches[i].from; + } + if (win[1] < switches[i].to) win[1] = switches[i].to; + preMatches.push(i); + } + } + if (preMatches.length === this.#types.length && win[0] - curTsUs < 3_000) { + let delta = win[1] - curTsUs; + // this._logger.debug(`Switch delta is ${delta / 1000}ms, curTs = ${curTsUs/1000}, to = ${win[1] / 1000}, from = ${win[0] / 1000}`); + if (delta + this._bufToKeep <= availUs) { + for (let j = 0; j < this.#types.length; j++) { + let switches = this._switches[this._tracks[this.#types[j]]]; + switches.splice(0, preMatches[j] + 1); + this._logger.debug( + `Remove ${this.#types[j]} switches till ${preMatches[j] + 1}. Cur ts = ${curTsUs}. Left ${switches.length}`, + ); + } + del = null; + res = delta; + this._logger.debug( + `Computed init switch shift ${res / 1000}ms to ${win[1]}`, + ); + } + } + if (del) { + for (let tId in del) { + this._switches[tId].splice(0, del[tId] + 1); + this._logger.debug( + `Remove ${tId} switches till ${del[tId] + 1}. Cur ts = ${curTsUs}. Left ${this._switches[tId].length}`, + ); + } + } + + return res; + } + + handleAction(data) { + switch (data.op) { + case "init-switch": + if (!this._switches[data.id]) { + this._switches[data.id] = []; + } + this._switches[data.id].push({ + from: data.data.fromPtsUs, + to: data.data.toPtsUs, + }); + this._swCnt++; + break; + case "main": + this._tracks[data.type] = data.id; + if (!this.#types.includes(data.type)) { + this.#types.push(data.type); + } + break; + case "rem": + if (this._switches[data.id]) { + this._swCnt -= this._switches[data.id].length; + } + this._switches[data.id] = undefined; + if (this._tracks[data.type] === data.id) { + this._tracks[data.type] = null; + } + break; + default: + this._logger.error(`Unknown action ${data.op}`); + break; + } + } + + hasPendingActions() { + return !!this._pendingActions && this._pendingActions.length > 0; + } + + clearPendingActions() { + if (this._pendingActions) this._pendingActions.length = 0; + } + + get pendingActions() { + return this._pendingActions; + } + + set bufferToKeep(valUs) { + this._bufToKeep = valUs; + } + + _portMessageHandler(event) { + const msg = event.data; + if (!msg || msg.aux) return; + if (msg.type === "transp-track-action") { + this.handleAction(msg.data); + } + } +} diff --git a/src/audio/config.js b/src/audio/config.js index 134d47d..6d60843 100644 --- a/src/audio/config.js +++ b/src/audio/config.js @@ -25,7 +25,7 @@ export class AudioConfig { this._numberOfChannels = config.numberOfChannels; this._sampleCount = config.sampleCount; - return config; + return this; } smpCntToTsUs(smpCnt) { @@ -34,9 +34,9 @@ export class AudioConfig { } tsUsToSmpCnt(tsUs) { - let smpCnt = (tsUs / 1000) * (this._sampleRate / 1000); - if (smpCnt < 0) smpCnt = 0; - return (smpCnt + 0.5) >>> 0; + let k = tsUs > 0 ? 1 : -1; + let smpCnt = ((k * tsUs) / 1000) * (this._sampleRate / 1000); + return k * ((smpCnt + 0.5) >>> 0); } isCompatible(config) { diff --git a/src/audio/nimio-processor.js b/src/audio/nimio-processor.js index 8dfa402..ec0c493 100644 --- a/src/audio/nimio-processor.js +++ b/src/audio/nimio-processor.js @@ -5,11 +5,13 @@ import { AudioConfig } from "./config"; import { LoggersFactory } from "@/shared/logger"; import { LatencyController } from "@/latency-controller"; import { WsolaProcessor } from "@/media/processors/wsola-processor"; +import { AdvertizerEvaluator } from "@/advertizer/evaluator"; class AudioNimioProcessor extends AudioWorkletProcessor { constructor(options) { super(options); + this.port.start(); LoggersFactory.setLevel(options.processorOptions.logLevel); LoggersFactory.toggleWorkletLogs(options.processorOptions.enableLogs); this._logger = LoggersFactory.create( @@ -32,12 +34,18 @@ class AudioNimioProcessor extends AudioWorkletProcessor { this._sampleCount, ); + this._advertizerEval = new AdvertizerEvaluator( + options.processorOptions.instanceName, + this.port, + ); + this._idle = options.processorOptions.idle; this._targetLatencyMs = options.processorOptions.latency; this._latencyCtrl = new LatencyController( options.processorOptions.instanceName, this._stateManager, this._audioConfig, + this._advertizerEval, { latency: this._targetLatencyMs, tolerance: options.processorOptions.latencyTolerance, @@ -45,6 +53,7 @@ class AudioNimioProcessor extends AudioWorkletProcessor { video: options.processorOptions.videoEnabled, audio: !this._idle, port: this.port, + syncBuffer: options.processorOptions.syncBuffer, }, ); this._latencyCtrl.speedFn = this._setSpeed.bind(this); @@ -86,17 +95,14 @@ class AudioNimioProcessor extends AudioWorkletProcessor { } if (!this._idle) { - this._stateManager.incSilenceUs(this._samplesDurationUs(sampleCount)); + let durUs = (this._audioConfig.smpCntToTsUs(sampleCount) + 0.5) >>> 0; + this._stateManager.incSilenceUs(durUs); if (!this._audioBuffer.isShareable) { this._audioBuffer.ensureCapacity(); } } } - _samplesDurationUs(sampleCount) { - return (this._audioConfig.smpCntToTsUs(sampleCount) + 0.5) >>> 0; - } - _setSpeed(speed, availableMs) { if (this._speed === speed) return; this._speed = speed; diff --git a/src/audio/volume-controller.js b/src/audio/volume-controller.js index b07dbf2..6d2e447 100644 --- a/src/audio/volume-controller.js +++ b/src/audio/volume-controller.js @@ -22,7 +22,7 @@ class AudioVolumeController { this._storageId = settings.volumeId; this._lastVolume = this._getStoredVolume(); - if (settings.muted) { + if (settings.muted || this._muted) { this._gainer.gain.value = 0; this._muted = true; this._eventBus.emit("nimio:muted", true); diff --git a/src/latency-controller.js b/src/latency-controller.js index 58cef8b..9ac66de 100644 --- a/src/latency-controller.js +++ b/src/latency-controller.js @@ -2,12 +2,14 @@ import { LatencyBufferMeter } from "@/latency/buffer-meter"; import { LoggersFactory } from "@/shared/logger"; import { currentTimeGetterMs } from "@/shared/helpers"; import { clamp } from "@/shared/helpers"; +import { SyncModePolicy } from "./sync-mode/policy"; export class LatencyController { - constructor(instName, stateMgr, audioConfig, params) { + constructor(instName, stateMgr, audioConfig, discontinuityEval, params) { this._instName = instName; this._stateMgr = stateMgr; this._audioConfig = audioConfig; + this._discontEval = discontinuityEval; this._params = params; this._audio = !!this._params.audio; @@ -21,18 +23,19 @@ export class LatencyController { this._shortWindowMs, this._longWindowMs, ); - this._adjustFn = params.adjustMethod === "seek" ? this._seek : this._zap; this._startThreshUs = this._startingBufferLevel(); this._minThreshUs = 50_000; // 50ms + this._discontEval.bufferToKeep = this._latencyMs * 900; // 0.9 of latency this._warmupMs = 3000; this._holdMs = 500; this._startHoldMs = 100; - this._minRate = 0.95; + this._minRate = 0.9; this._minRateStep = 1 / 128; this._maxRate = 1.25; this._rateK = 0.00015; // proportional gain: rate = 1 + rateK * deltaMs + this._rateK = 0.0002; this._minLatencyDelta = 40; this._allowedLatencyDelta = params.tolerance - this._latencyMs; @@ -40,13 +43,20 @@ export class LatencyController { this._allowedLatencyDelta = this._minLatencyDelta; } this._minRateChangeIntervalMs = 500; - this._minSeekIntervalMs = 4000; // don't seek more frequently + this._minSeekIntervalMs = 3000; // don't seek more frequently this._getCurTimeMs = currentTimeGetterMs(); - this.reset(); this._logger = LoggersFactory.create(instName, "Latency ctrl", params.port); + this._latencyControlFn = this._adjustPlaybackLatency; + this._adjustFn = params.adjustMethod === "seek" ? this._seek : this._zap; + if (params.syncBuffer > 0) { + this._latencyControlFn = this._syncPlaybackLatency; + this._syncModePolicy = new SyncModePolicy(params.syncBuffer, params.port); + this._syncModePolicy.logger = this._logger; + } + this._logger.debug( `initialized: latency=${this._latencyMs}ms, start threshold=${this._startThreshUs}us, video=${this._video}, audio=${this._audio}`, ); @@ -59,7 +69,8 @@ export class LatencyController { this._bufferMeter.reset(); this._startTimeMs = -1; - this._lastActionTime = -this._minSeekIntervalMs; + this._lastSeekTime = -this._minSeekIntervalMs; + this._lastZapTime = -this._minRateChangeIntervalMs; this._audioAvailUs = this._videoAvailUs = undefined; this._pendingStableSince = null; this._restoreLatency = false; @@ -67,14 +78,14 @@ export class LatencyController { start() { if (this._pauseTime > 0) { - let pauseDuration = performance.now() - this._pauseTime; + let pauseDuration = this._getCurTimeMs() - this._pauseTime; this._prevVideoTime += pauseDuration; this._pauseTime = 0; } } pause() { - this._pauseTime = performance.now(); + this._pauseTime = this._getCurTimeMs(); } availableMs(type) { @@ -97,9 +108,13 @@ export class LatencyController { incCurrentAudioSamples(sampleCount) { if (this.isPending()) return this._curTsUs; + let samplesUs = this._audioConfig.smpCntToTsUs(sampleCount); + this._moveCurrentPosition(samplesUs, sampleCount); + + if (!this._handleDiscontinuity()) { + this._latencyControlFn(); + } - this._stateMgr.incCurrentTsSmp(sampleCount); - this._adjustPlaybackLatency(); return this._curTsUs; } @@ -107,14 +122,17 @@ export class LatencyController { this._getCurrentTsUs(); this._calculateAvailable(); let prevVideoTime = this._prevVideoTime; - this._prevVideoTime = performance.now(); + this._prevVideoTime = this._getCurTimeMs(); if (this._checkPending() || prevVideoTime === 0) { return this._curTsUs; } - let timeUsPast = (performance.now() - prevVideoTime) * speed * 1000; - this._stateMgr.incCurrentTsSmp(this._audioConfig.tsUsToSmpCnt(timeUsPast)); - this._adjustPlaybackLatency(); + let timeUsPast = (this._getCurTimeMs() - prevVideoTime) * speed * 1000; + this._moveCurrentPosition(timeUsPast); + + if (!this._handleDiscontinuity()) { + this._latencyControlFn(); + } return this._curTsUs; } @@ -151,6 +169,16 @@ export class LatencyController { this._audio = val; } + set syncModePtsOffset(val) { + if (!this._syncModePolicy) { + this._logger.error( + "Attempt to set pts offset, while sync mode is disabled", + ); + return; + } + this._syncModePolicy.ptsOffset = val; + } + _checkPending() { if (this.isUnderrun() && this._startThreshUs === 0) { this._logger.debug( @@ -176,19 +204,16 @@ export class LatencyController { this._availableUs = Number.MAX_VALUE; if (this._audio) { this._getAudioAvailableUs(); - let availableMs = (this._audioAvailUs / 1000 + 0.5) >>> 0; - this._stateMgr.setAvailableAudioMs(availableMs); + this._setAudioAvailableMs(); this._availableUs = this._audioAvailUs; } if (this._video) { this._getVideoAvailableUs(); - let availableMs = (this._videoAvailUs / 1000 + 0.5) >>> 0; - this._stateMgr.setAvailableVideoMs(availableMs); + this._setVideoAvailableMs(); this._availableUs = Math.min(this._availableUs, this._videoAvailUs); } // this._logger.debug(`Available ms=${this._availableUs / 1000}, start time=${this._startTimeMs}`); - if (this._startTimeMs >= 0) { this._bufferMeter.update(this._availableUs / 1000, this._getCurTimeMs()); } @@ -201,6 +226,7 @@ export class LatencyController { let curSmpCnt = this._stateMgr.getCurrentTsSmp(); this._curTsUs = this._audioConfig.smpCntToTsUs(curSmpCnt) + this._startTsUs; + return this._curTsUs; } _getVideoAvailableUs() { @@ -208,26 +234,29 @@ export class LatencyController { if (this._videoAvailUs < 0) this._videoAvailUs = 0; } + _setVideoAvailableMs() { + let availableMs = (this._videoAvailUs / 1000 + 0.5) >>> 0; + this._stateMgr.setAvailableVideoMs(availableMs); + } + _getAudioAvailableUs() { this._audioAvailUs = this._stateMgr.getAudioLatestTsUs() - this._curTsUs; if (this._audioAvailUs < 0) this._audioAvailUs = 0; } + _setAudioAvailableMs() { + let availableMs = (this._audioAvailUs / 1000 + 0.5) >>> 0; + this._stateMgr.setAvailableAudioMs(availableMs); + } + _adjustPlaybackLatency() { if (this._startTimeMs < 0) return; // not started yet const now = this._getCurTimeMs(); - const age = now - this._startTimeMs; - const useShort = age < this._warmupMs; - - const shortB = this._bufferMeter.short(now); - this._stateMgr.setMinBufferMs("short", shortB); - const longB = this._bufferMeter.long(now); - this._stateMgr.setMinBufferMs("long", longB); - const emaB = this._bufferMeter.ema(); - this._stateMgr.setMinBufferMs("ema", emaB); + this._updateBufferLevels(now); - const bufMin = useShort ? shortB : longB; + const age = now - this._startTimeMs; + const bufMin = age < this._warmupMs ? this._shortB : this._longB; const deltaMs = bufMin - this._latencyMs; // const stable = Math.abs(bufMin - bufEma) < (this._latencyMs * 0.1); @@ -235,26 +264,43 @@ export class LatencyController { this._restoreLatency = true; } - let goForward = false; + let goMove = false; if (deltaMs > this._minLatencyDelta && this._restoreLatency) { // this._logger.debug(`Delta ms=${deltaMs}, buffer ms=${bufMin}, age ms=${age}`); // wait for holdMs to avoid acting on single spikes if (!this._pendingStableSince) { this._pendingStableSince = now; } else if (now - this._pendingStableSince > this._holdMs) { - goForward = true; + goMove = true; } } else { this._pendingStableSince = null; this._restoreLatency = false; } - if (this._lastActionTime < 0) { + if (age < this._warmupMs / 2 && this._lastSeekTime < 0) { // Do seek on startup if possible if (this._tryInitialSeek(deltaMs, bufMin, now)) return; } - this._adjustFn(goForward, deltaMs, bufMin, now); + this._adjustFn(goMove, deltaMs, bufMin, now); + } + + _syncPlaybackLatency() { + if (this._startTimeMs < 0) return; // not started yet + + const now = this._getCurTimeMs(); + this._updateBufferLevels(now); + + let curTimeMs = this._curTsUs / 1000; + let availMs = this._availableUs / 1000; + let deltaMs = this._syncModePolicy.computeAdjustment(curTimeMs, availMs); + + if (Math.abs(deltaMs) >= 2000) { + this._seek(true, deltaMs, availMs, now); + } else { + this._zap(deltaMs !== 0, deltaMs, availMs, now); + } } _startingBufferLevel() { @@ -264,7 +310,7 @@ export class LatencyController { _tryInitialSeek(deltaMs, buf, now) { if (deltaMs > this._latencyMs) { let stableTime = now - this._pendingStableSince; - if (stableTime >= this._startHoldMs && stableTime < this._warmupMs) { + if (stableTime >= this._startHoldMs) { this._seek(true, deltaMs, buf, now); return true; } @@ -272,36 +318,77 @@ export class LatencyController { return false; } - _zap(goForward, deltaMs, bufMin, now) { + _handleDiscontinuity() { + if (!this._discontEval.isApplicable()) return false; + let shUs = this._discontEval.computeShift(this._curTsUs, this._availableUs); + if (shUs > 0) { + const now = this._getCurTimeMs(); + this._seek(true, shUs / 1000, this._availableUs / 1000, now); + } + + return shUs > 0; + } + + _zap(goMove, deltaMs, curBuf, now) { let rate = 1; - if (goForward) { - if (now - this._lastActionTime < this._minRateChangeIntervalMs) { + if (goMove) { + if (now - this._lastZapTime < this._minRateChangeIntervalMs) { return; } rate = clamp(1 + this._rateK * deltaMs, this._minRate, this._maxRate); - this._lastActionTime = now; + if (deltaMs < -0.6) { + rate = 0; // stop reading samples to adjust latency faster + } + this._lastZapTime = now; } - if (Math.abs(rate - 1) < this._minRateStep) rate = 1; // snap - this._setSpeed(rate, bufMin); - } - _seek(goForward, deltaMs, bufMin, now) { - if (!goForward) return; + // if (goMove) { + // this._logger.debug(`Zap with rate = ${rate}, deltaMs = ${deltaMs}`); + // } + this._stateMgr.setCurrentSpeed(rate); + this._setSpeed(rate, curBuf); + } - if (goForward) { - if (now - this._lastActionTime < this._minSeekIntervalMs) { - return; - } + _seek(goMove, deltaMs, curBuf, now) { + if ( + !goMove || + deltaMs === 0 || + now - this._lastSeekTime < this._minSeekIntervalMs + ) { + this._logger.debug(`Skip seek by ${deltaMs}ms (too frequent)`); + return; } - this._lastActionTime = now; - this._logger.debug(`Seek forward by ${deltaMs}ms, cur bufer ms=${bufMin}`); + this._lastSeekTime = now; + this._logger.debug(`Seek by ${deltaMs}ms, cur bufer ms=${curBuf}`); + this._moveCurrentPosition(deltaMs * 1000); + } + + _moveCurrentPosition(deltaUs, deltaSmpCnt) { + if (deltaSmpCnt === undefined) { + deltaSmpCnt = this._audioConfig.tsUsToSmpCnt(deltaUs); + } - let deltaUs = deltaMs * 1000; - this._stateMgr.incCurrentTsSmp(this._audioConfig.tsUsToSmpCnt(deltaUs)); - if (this._video) this._videoAvailUs -= deltaUs; - if (this._audio) this._audioAvailUs -= deltaUs; + this._stateMgr.incCurrentTsSmp(deltaSmpCnt); + if (this._video) { + this._videoAvailUs -= deltaUs; + this._setVideoAvailableMs(); + } + if (this._audio) { + this._audioAvailUs -= deltaUs; + this._setAudioAvailableMs(); + } this._availableUs -= deltaUs; + this._curTsUs += deltaUs; + } + + _updateBufferLevels(now) { + this._shortB = this._bufferMeter.short(now); + this._stateMgr.setMinBufferMs("short", this._shortB); + this._longB = this._bufferMeter.long(now); + this._stateMgr.setMinBufferMs("long", this._longB); + this._emaB = this._bufferMeter.ema(); + this._stateMgr.setMinBufferMs("ema", this._emaB); } } diff --git a/src/media/buffers/readable-audio-buffer.js b/src/media/buffers/readable-audio-buffer.js index 98abbc3..6e9d96b 100644 --- a/src/media/buffers/readable-audio-buffer.js +++ b/src/media/buffers/readable-audio-buffer.js @@ -1,5 +1,7 @@ import { SharedAudioBuffer } from "./shared-audio-buffer"; +let skipCnt = 0; + export class ReadableAudioBuffer extends SharedAudioBuffer { read(startTsNs, outputChannels, step = 1) { if (outputChannels.length !== this.numChannels) { @@ -99,8 +101,10 @@ export class ReadableAudioBuffer extends SharedAudioBuffer { this.setReadIdx(readPrms.startIdx + 1); } else if (skipIdx !== null) { this.setReadIdx(skipIdx); + skipCnt++; + let lastTs = this.lastFrameTs; console.error( - `No frames found in the requested range: ${startTsNs}..${endTsNs}`, + `No frames found in the requested range: ${startTsNs}..${endTsNs}, skipIdx = ${skipIdx}. Size = ${this.getSize()}. Last ts = ${lastTs}, dist=${(lastTs - startTsNs / 1000) / 1000}ms`, ); } if (readPrms.endIdx === null) readPrms.endTsNs = endTsNs; @@ -111,7 +115,6 @@ export class ReadableAudioBuffer extends SharedAudioBuffer { } if (!readPrms.rate) readPrms.rate = readPrms.prelimRate; - return this._fillOutput(outputChannels, readPrms); } @@ -123,34 +126,15 @@ export class ReadableAudioBuffer extends SharedAudioBuffer { } _fillOutput(outputChannels, rParams) { - let step = rParams.rate; - let outLength = rParams.outLength; - if (rParams.startIdx !== null && rParams.endIdx !== null) { - let expProcCnt = rParams.endOffset - rParams.startOffset; - if (rParams.startIdx !== rParams.endIdx) { - expProcCnt += rParams.startCount; - } - if (expProcCnt !== outLength) { - console.error(`fillOutput exp sample count = ${expProcCnt}`); - } - let steppedCount = (expProcCnt / step + 0.5) >>> 0; - if (steppedCount < outLength) { - let prevStep = step; - step = expProcCnt / outLength; - if (step < 0.95) step = 0.95; - console.log( - `Fixed step from ${prevStep} to ${step}, start=${rParams.startOffset}, end=${rParams.endOffset}, srate=${rParams.startRate}, erate=${rParams.endRate}, scount = ${rParams.startCount}, expected count: ${expProcCnt}, stepped count: ${steppedCount}`, - ); - } - } + let step = this._getReadStep(rParams); let processed = null; if (rParams.startIdx === rParams.endIdx) { - if (rParams.startIdx === null) { + if (rParams.startIdx === null || step === 0) { for (let c = 0; c < this.numChannels; c++) { outputChannels[c].fill(0); } - processed = (outLength * step + 0.5) >>> 0; + processed = (rParams.outLength * step + 0.5) >>> 0; } else { this._copyChannelsData( this._frames[rParams.startIdx], @@ -191,22 +175,45 @@ export class ReadableAudioBuffer extends SharedAudioBuffer { let fillCount; if (startCount === null) { - fillCount = outLength - endCount; - console.error("Fill silence (start)", fillCount); + fillCount = rParams.outLength - endCount; + console.warn("Fill silence (start)", fillCount); this._fillSilence(outputChannels, 0, fillCount); } else if (endCount === null) { - fillCount = outLength - startCount; - console.error("Fill silence (end)", fillCount); + fillCount = rParams.outLength - startCount; + console.warn("Fill silence (end)", fillCount); this._fillSilence(outputChannels, startCount, fillCount); - } else if (startCount + endCount < outLength) { - fillCount = outLength - startCount - endCount; - console.error("Fill silence (middle)", fillCount); + } else if (startCount + endCount < rParams.outLength) { + fillCount = rParams.outLength - startCount - endCount; + console.warn("Fill silence (middle)", fillCount); this._fillSilence(outputChannels, startCount, fillCount); } return this._calcProcessedSamples(rParams); } + _getReadStep(rParams) { + let step = rParams.rate; + if (rParams.startIdx === null || rParams.endIdx === null || step === 0) { + return step; + } + + let expProcCnt = rParams.endOffset - rParams.startOffset; + if (rParams.startIdx !== rParams.endIdx) { + expProcCnt += rParams.startCount; + } + let steppedCount = (expProcCnt / step + 0.5) >>> 0; + if (steppedCount < rParams.outLength) { + let curStep = step; + step = expProcCnt / rParams.outLength; + if (step > 0 && step < 0.9) step = 0.9; + console.log( + `Fixed step from ${curStep} to ${step}, start=(${rParams.startIdx}, ${rParams.startOffset}), end=(${rParams.endIdx}, ${rParams.endOffset}), srate=${rParams.startRate}, erate=${rParams.endRate}, scount = ${rParams.startCount}, expected count: ${expProcCnt}, stepped count: ${steppedCount}`, + ); + } + + return step; + } + _calcSampleTs(offset, fStartTsNs, sCount) { return (offset * this._frameNs) / sCount + fStartTsNs; } diff --git a/src/media/buffers/readable-trans-audio-buffer.js b/src/media/buffers/readable-trans-audio-buffer.js index aba3416..fdbea84 100644 --- a/src/media/buffers/readable-trans-audio-buffer.js +++ b/src/media/buffers/readable-trans-audio-buffer.js @@ -12,7 +12,6 @@ export class ReadableTransAudioBuffer extends ReadableAudioBuffer { init() { this._initMessaging(); this._startMsgDispatcher(); - this._minFreeSpan = this._overflowShift / 2; this._msgIvalMs = (3 * 1000 * this._sampleCount) / this.sampleRate; // 3 frames } @@ -43,16 +42,14 @@ export class ReadableTransAudioBuffer extends ReadableAudioBuffer { ensureCapacity() { const r = this.getReadIdx(); const w = this.getWriteIdx(); - if (r === 0 && w === 0) return false; + if (r === w) return false; const minr = this._dispData.rIdx ?? r; const free = this._dist(w, minr); - if (free < this._minFreeSpan) { - if ( - r === minr || - this._dist(minr, w) - this._dist(r, w) < this._overflowShift - ) { - this.setReadIdx(r + this._overflowShift); + if (free < this._overflowShift) { + let freeSize = 2 * this._overflowShift; + if (r === minr || this._dist(minr, w) - this._dist(r, w) < freeSize) { + this.setReadIdx(minr + freeSize); } this._sendReadStatus(minr); return true; @@ -97,7 +94,7 @@ export class ReadableTransAudioBuffer extends ReadableAudioBuffer { _handlePortMessage(event) { const msg = event.data; - if (!msg) return; + if (!msg || msg.aux) return; try { if (msg.type === "tb:frames") { @@ -109,6 +106,8 @@ export class ReadableTransAudioBuffer extends ReadableAudioBuffer { this._frames[idx] = msg.frames[i]; } this.setWriteIdx(idx + 1); + } else if (msg.type === "tb:overflow") { + this.ensureCapacity(); } else if (msg.type === "tb:reset") { this.reset(); } diff --git a/src/media/buffers/shared-audio-buffer.js b/src/media/buffers/shared-audio-buffer.js index c7c6721..f87ac32 100644 --- a/src/media/buffers/shared-audio-buffer.js +++ b/src/media/buffers/shared-audio-buffer.js @@ -36,6 +36,10 @@ export class SharedAudioBuffer { } static allocate(bufferSec, sampleRate, numChannels, sampleCount) { + const isShared = isSharedArrayBufferSupported(); + if (!isShared) { + bufferSec += 2; // add 2 seconds for overflow prevention + } const capacity = Math.ceil((bufferSec * sampleRate) / sampleCount); // one frame = Float64 timestamp(2 Float32) + Float32 rate + frame size @@ -46,7 +50,7 @@ export class SharedAudioBuffer { numChannels * sampleCount * Int16Array.BYTES_PER_ELEMENT; let fullSize = SharedAudioBuffer.HEADER_BYTES + fAuxSize + tempSize; - if (isSharedArrayBufferSupported()) { + if (isShared) { fullSize += audioFrameSize(numChannels, sampleCount) * capacity; } diff --git a/src/media/buffers/writable-trans-audio-buffer.js b/src/media/buffers/writable-trans-audio-buffer.js index 259cc0b..4d5fa9d 100644 --- a/src/media/buffers/writable-trans-audio-buffer.js +++ b/src/media/buffers/writable-trans-audio-buffer.js @@ -41,7 +41,7 @@ export class WritableTransAudioBuffer extends WritableAudioBuffer { _handlePortMessage(event) { const msg = event.data; - if (!msg || msg.log) return; + if (!msg || msg.log || msg.aux) return; try { if (msg.type === "tb:read") { @@ -61,11 +61,11 @@ export class WritableTransAudioBuffer extends WritableAudioBuffer { if (idx === this._capacity) idx = 0; } this.setReadIdx(msg.end); - // if (fCount === this._overflowShift) { + // if (fCount === this._overflowShift * 2) { // let w = this.getWriteIdx(); // let r = this.getReadIdx(); // let free = this._dist(w, r); - // console.log(`Returned ${this._overflowShift} frames from ${msg.start} to ${msg.end} due to overflow, free = ${free}`); + // console.log(`Returned ${this._overflowShift * 2} frames from ${msg.start} to ${msg.end} due to overflow, free = ${free}`); // } } else if (msg.type === "tb:reset") { this.reset(true); @@ -137,6 +137,17 @@ export class WritableTransAudioBuffer extends WritableAudioBuffer { this._dispData.frames.length = 0; this._dispData.buffers.length = 0; } + + _incWriteIdx(writeIdx) { + let wIdx = this.setWriteIdx(writeIdx + 1); + let rIdx = this.getReadIdx(); + if (this._dist(wIdx, rIdx) < this._overflowShift) { + console.warn( + `wIdx = ${wIdx}, rIdx = ${rIdx}, send req to free items from the reader`, + ); + this._sendMessage({ type: "tb:overflow" }); + } + } } Object.assign(WritableTransAudioBuffer.prototype, PortMessaging); diff --git a/src/media/decoders/decoder-audio.js b/src/media/decoders/decoder-audio.js index 634607f..c39849d 100644 --- a/src/media/decoders/decoder-audio.js +++ b/src/media/decoders/decoder-audio.js @@ -4,8 +4,8 @@ import { adjustCodecId } from "./checker"; let audioDecoder; let support; -let lastTimestampUs; -let frameDurationUs; +let lastTimestampUs = null; +let frameDurationUs = null; let timestampBuffer = new RingBuffer("Audio Decoder", 3000); const buffered = []; @@ -14,7 +14,7 @@ let config = {}; function processDecodedFrame(audioFrame) { let rawTimestamp = timestampBuffer.pop(); let decTimestamp = audioFrame.timestamp; - if (decTimestamp === rawTimestamp && lastTimestampUs !== undefined) { + if (decTimestamp === rawTimestamp && lastTimestampUs !== null) { decTimestamp = lastTimestampUs + frameDurationUs; } lastTimestampUs = decTimestamp; @@ -22,10 +22,10 @@ function processDecodedFrame(audioFrame) { self.postMessage( { type: "decodedFrame", - audioFrame: audioFrame, decoderQueue: audioDecoder.decodeQueueSize, - rawTimestamp: rawTimestamp, - decTimestamp: decTimestamp, + audioFrame, + rawTimestamp, + decTimestamp, }, [audioFrame], ); @@ -59,6 +59,12 @@ self.addEventListener("message", async function (e) { support = null; break; case "codecData": + if (audioDecoder) { + support = null; + await audioDecoder.flush(); + shutdownDecoder(); + lastTimestampUs = null; + } audioDecoder = new AudioDecoder({ output: (audioFrame) => { processDecodedFrame(audioFrame); diff --git a/src/media/decoders/decoder-video.js b/src/media/decoders/decoder-video.js index e8f475d..64c5c5d 100644 --- a/src/media/decoders/decoder-video.js +++ b/src/media/decoders/decoder-video.js @@ -22,9 +22,9 @@ function processDecodedFrame(videoFrame) { self.postMessage( { type: "decodedFrame", - videoFrame: videoFrame, decoderQueue: videoDecoder.decodeQueueSize, decoderLatency: latencyMs, + videoFrame, }, [videoFrame], ); @@ -89,6 +89,13 @@ self.addEventListener("message", async function (e) { support = null; break; case "codecData": + if (videoDecoder) { + support = null; + const vd = videoDecoder; + videoDecoder.flush().finally(function () { + if (typeof vd.close === "function") vd.close(); + }); + } videoDecoder = new VideoDecoder({ output: (frame) => { processDecodedFrame(frame); diff --git a/src/media/decoders/flow.js b/src/media/decoders/flow.js index 89ba9ee..ea0cc57 100644 --- a/src/media/decoders/flow.js +++ b/src/media/decoders/flow.js @@ -1,6 +1,7 @@ import { TimestampManager } from "./timestamp-manager"; import { MetricsManager } from "@/metrics/manager"; import { LoggersFactory } from "@/shared/logger"; +import { EventBus } from "@/event-bus"; const SWITCH_THRESHOLD_US = 8_000_000; @@ -24,6 +25,7 @@ export class DecoderFlow { this._timestampManager = TimestampManager.getInstance(instanceName); this._timestampManager.addTrack(this._trackId, this._type); + this._eventBus = EventBus.getInstance(instanceName); this._decoder = new Worker(new URL(url, import.meta.url), { type: "module", }); @@ -155,6 +157,11 @@ export class DecoderFlow { this._metricsManager.remove(this._trackId); this._timestampManager.removeTrack(this._trackId); this._decoder.postMessage({ type: "shutdown" }); + this._eventBus.emit("transp:track-action", { + op: "rem", + id: this._trackId, + type: this._type, + }); } async _handleDecoderMessage(e) { @@ -194,6 +201,11 @@ export class DecoderFlow { this._switchPeerFlow = null; this._switchContext = null; this._onSwitchResult(true); + this._eventBus.emit("transp:track-action", { + op: "main", + id: this._trackId, + type: this._type, + }); } break; default: diff --git a/src/media/decoders/timestamp-manager.js b/src/media/decoders/timestamp-manager.js index db24c94..b37c58e 100644 --- a/src/media/decoders/timestamp-manager.js +++ b/src/media/decoders/timestamp-manager.js @@ -1,10 +1,15 @@ import { multiInstanceService } from "@/shared/service"; import { LoggersFactory } from "@/shared/logger"; +import { EventBus } from "@/event-bus"; + +const DISCONT_THRESH_US = 10_000_000; class TimestampManager { constructor(instName) { this._instName = instName; this._tsValidators = new Map(); + this._eventBus = EventBus.getInstance(instName); + this._logger = LoggersFactory.create(instName, "TimestampManager"); } init(settings) { @@ -16,9 +21,27 @@ class TimestampManager { this._tsValidators.set(id, tv); } + rebaseTrack(id) { + let tv = this._tsValidators.get(id); + if (!tv) return false; + this._logger.debug(`Rebase track ${id}`); + if (tv.timeBase) { + if (!this._baseSwitch) this._baseSwitch = { ids: {}, cnt: 0, rcnt: 0 }; + if (this._baseSwitch.ids[id] === 0) { + this._baseSwitch.tb = undefined; + } + this._baseSwitch.ids[id] = 1; + this._baseSwitch.rcnt++; + this._baseSwitch.cnt++; + } + } + validateChunk(id, chunk) { let tv = this._tsValidators.get(id); if (!tv) return false; + + // check if new init segment arrived (advertizer) + this._checkBaseSwitch(tv, id, chunk); return tv.validateChunk(chunk); } @@ -33,6 +56,47 @@ class TimestampManager { removeTrack(id) { this._tsValidators.delete(id); } + + _checkBaseSwitch(tv, id, chunk) { + if (!this._baseSwitch || this._baseSwitch.ids[id] !== 1) return; + + let tbase = tv.timeBase; + + let chDts = chunk.pts - chunk.offset; + let dtsDiff = chDts - tbase.rawDts; + let data = { + fromPtsUs: tbase.dts + tbase.offset, + toPtsUs: tbase.dts + dtsDiff + chunk.offset, + }; + this._logger.debug( + `checkBaseSwitch track ${id}, dts diff = ${dtsDiff}, cur chunk pts = ${chunk.pts}, prev chunk dts = ${tbase.dts}, rawDts = ${tbase.rawDts}`, + ); + + if (dtsDiff < 0 || dtsDiff > DISCONT_THRESH_US) { + let newTbase = { rawDts: chDts }; + if (this._baseSwitch.tb === undefined) { + newTbase.dts = tbase.dts + 3_000_000; + this._baseSwitch.tb = newTbase; + } else { + let bsDtsDiff = chDts - this._baseSwitch.tb.rawDts; + newTbase.dts = this._baseSwitch.tb.dts + bsDtsDiff; + } + this._logger.debug( + `Apply base switch time for ${id}, new time base: dts = ${newTbase.dts}, rawDts = ${newTbase.rawDts}`, + ); + tv.timeBase = newTbase; + data.toPtsUs = newTbase.dts + chunk.offset; + } + this._baseSwitch.ids[id] = 0; + this._baseSwitch.cnt--; + if ( + this._baseSwitch.cnt === 0 && + this._baseSwitch.rcnt === this._tsValidators.size + ) { + this._baseSwitch = undefined; + } + this._eventBus.emit("transp:track-action", { op: "init-switch", id, data }); + } } class TimestampValidator { @@ -113,7 +177,7 @@ class TimestampValidator { if (this._hasDiscontinuity(dtsDiff)) { this._logger.debug( - `Incorrect DTS difference (${dtsDiff}) between previous (ts: ${curChunk.rawDts}, offset: ${curChunk.offset}, sap: ${curChunk.sap}) and current frame (ts: ${dts}, offset: ${data.offset})`, + `Incorrect DTS difference (${dtsDiff}) between previous (ts: ${curChunk.rawDts}, offset: ${curChunk.offset}) and current frame (ts: ${dts}, offset: ${data.offset})`, ); dtsDiff = this._lastChunkDuration; this._dtsDistCompensation = 0; @@ -131,6 +195,7 @@ class TimestampValidator { } set timeBase(tb) { + this._lastChunk = null; this._timeBase = tb; } @@ -139,13 +204,14 @@ class TimestampValidator { return { dts: this._lastChunk.dts, rawDts: this._lastChunk.rawDts, + offset: this._lastChunk.offset, }; } _applyTimeBase(chunk) { let chDts = chunk.pts - chunk.offset; let rawDts = chDts; - if (Math.abs(chDts - this._timeBase.rawDts) < 9_000_000) { + if (Math.abs(chDts - this._timeBase.rawDts) <= DISCONT_THRESH_US) { let dtsDiff = this._timeBase.dts - this._timeBase.rawDts; if (dtsDiff !== 0) { let newDts = chDts + dtsDiff; @@ -179,7 +245,7 @@ class TimestampValidator { } _hasDiscontinuity(tsDiff) { - return tsDiff < 0 || tsDiff > 10_000_000; + return tsDiff < 0 || tsDiff > DISCONT_THRESH_US; } _setLastChunk(dts, rawDts, offset) { diff --git a/src/media/parsers/opus-config-parser.js b/src/media/parsers/opus-config-parser.js index 4a06171..bf6e8e4 100644 --- a/src/media/parsers/opus-config-parser.js +++ b/src/media/parsers/opus-config-parser.js @@ -23,13 +23,13 @@ export function parseOpusConfig(opusPacket) { let durationMs = 0; if (cfg >= 0 && cfg <= 11) { // SILK: 10, 20, 40, 60 ms - durationMs = 10 * (1 << cfg % 4); + durationMs = 10 * (1 << (cfg % 4)); } else if (cfg >= 12 && cfg <= 15) { // Hybrid: only 10 or 20 ms durationMs = cfg % 2 === 0 ? 10 : 20; } else if (cfg >= 16 && cfg <= 31) { // CELT: 2.5, 5, 10, 20 ms - durationMs = 2.5 * (1 << cfg % 4); + durationMs = 2.5 * (1 << (cfg % 4)); } return durationMs * 48; // OPUS has constant sample rate 48Khz }; diff --git a/src/nimio-sync-mode.js b/src/nimio-sync-mode.js new file mode 100644 index 0000000..975feb9 --- /dev/null +++ b/src/nimio-sync-mode.js @@ -0,0 +1,33 @@ +export const NimioSyncMode = { + _createSyncModeParams() { + this._syncModeParams = {}; + this._eventBus.on("nimio:sync-mode-params", (data) => { + this._syncModeParams.playerTimeMs = data.playerTimeMs; + this._syncModeParams.serverTimeMs = data.serverTimeMs; + }); + }, + + _initSyncModeParams(frame) { + if (frame.chunkType === "key" || this._noVideo) { + let srvTimeDiffUs = frame.showTime - this._syncModeParams.serverTimeMs; + this._syncModeParams.ptsOffsetMs = + this._syncModeParams.playerTimeMs + (srvTimeDiffUs - frame.pts) / 1000; + this._syncModeParams.inited = true; + this._applySyncModeParams(); + } + }, + + _applySyncModeParams() { + let smParams = this._syncModeParams; + if (!smParams.inited || smParams.applied || !this._audioNode) { + return; + } + + this._latencyCtrl.syncModePtsOffset = smParams.ptsOffsetMs; + this._audioNode.port.postMessage({ + type: "sync-mode-params", + ptsOffsetMs: smParams.ptsOffsetMs, + }); + smParams.applied = true; + }, +}; diff --git a/src/nimio-transport.js b/src/nimio-transport.js index 4a84bc7..e464c55 100644 --- a/src/nimio-transport.js +++ b/src/nimio-transport.js @@ -1,3 +1,4 @@ +import { AudioConfig } from "./audio/config"; import { TransportAdapter } from "./transport/adapter"; export const NimioTransport = { @@ -13,6 +14,35 @@ export const NimioTransport = { audioChunk: this._onAudioChunkReceived.bind(this), disconnect: this._onDisconnect.bind(this), }; + this._eventBus.on("transp:track-action", this._onTrackAction.bind(this)); + }, + + _onTrackAction(data) { + this._advertizerEval.handleAction(data); + if (!this._audioNode) { + this._advertizerEval.pendingActions.push(data); + return; + } + + this._audioNode.port.postMessage({ type: "transp-track-action", data }); + }, + + _sendPendingAdvertizerActions() { + if (this._advertizerEval.hasPendingActions()) { + const hdlr = (event) => { + if (event.data != "transp-discont-eval-ready") return; + let pa = this._advertizerEval.pendingActions; + for (let i = 0; i < pa.length; i++) { + this._audioNode.port.postMessage({ + type: "transp-track-action", + data: pa[i], + }); + } + this._advertizerEval.clearPendingActions(); + this._audioNode.port.removeEventListener("message", hdlr); + }; + this._audioNode.port.addEventListener("message", hdlr); + } }, _onDisconnect(data) { @@ -79,6 +109,7 @@ export const NimioTransport = { _onVideoCodecDataReceived(data) { this._runMetrics(data); + this._timestampManager.rebaseTrack(data.trackId); if (this._abrController?.isProbing(data.trackId)) { return this._abrController.handleCodecData(data); @@ -98,31 +129,43 @@ export const NimioTransport = { _onAudioCodecDataReceived(data) { this._runMetrics(data); + this._timestampManager.rebaseTrack(data.trackId); - let audioAvailable = true; - let curConfigVals = this._audioConfig.get(); - let newConfigVals = this._audioConfig.parse(data.data, data.family); - let decoderFlow, buffer; + let audioAvailable, decoderFlow, buffer; + let newCfg = new AudioConfig().parse(data.data, data.family); if (this._isNextRenditionTrack(data.trackId)) { - if (!newConfigVals || !this._audioConfig.isCompatible(curConfigVals)) { + if (!this._audioConfig.isCompatible(newCfg)) { this._logger.warn( - "Received incompatible audio config for next rendition", + "Incompatible audio config for rendition switch", data.trackId, - curConfigVals, - newConfigVals, + this._audioConfig.get(), + newCfg.get(), ); - this._audioConfig.set(curConfigVals); + this._nextRenditionData.decoderFlow.destroy(); this._onRenditionSwitchResult("audio", false); this._sldpManager.cancelStream(data.trackId); return; } - decoderFlow = this._nextRenditionData.decoderFlow; + audioAvailable = true; + this._audioConfig = newCfg; buffer = this._tempBuffer; + decoderFlow = this._nextRenditionData.decoderFlow; this._decoderFlows["audio"].switchTo(decoderFlow); } else { - audioAvailable = this._prepareAudioOutput(newConfigVals); + if (!this._audioBuffer || this._audioConfig.isCompatible(newCfg)) { + this._audioConfig = newCfg; + audioAvailable = this._prepareAudioOutput(); + } else { + this._logger.warn( + "Incompatible audio config update", + data.trackId, + this._audioConfig.get(), + newCfg.get(), + ); + } + if (audioAvailable) { decoderFlow = this._decoderFlows["audio"]; buffer = this._audioBuffer; @@ -135,7 +178,7 @@ export const NimioTransport = { } if (audioAvailable) { - decoderFlow.setCodecData({ codecData: data.data, config: newConfigVals }); + decoderFlow.setCodecData({ codecData: data.data, config: newCfg.get() }); decoderFlow.setBuffer(buffer, this._state); } }, @@ -155,6 +198,10 @@ export const NimioTransport = { return; } + if (this._syncModeParams && !this._syncModeParams.inited) { + this._initSyncModeParams(data); + } + this._metricsManager.reportBandwidth( data.trackId, data.frameWithHeader.byteLength, diff --git a/src/nimio.js b/src/nimio.js index c6a7212..b71182c 100644 --- a/src/nimio.js +++ b/src/nimio.js @@ -20,6 +20,7 @@ import { NimioRenditions } from "./nimio-renditions"; import { NimioAbr } from "./nimio-abr"; import { NimioVolume } from "./nimio-volume"; import { NimioEvents } from "./nimio-events"; +import { NimioSyncMode } from "./nimio-sync-mode"; import { MetricsManager } from "./metrics/manager"; import { LoggersFactory } from "./shared/logger"; import { AudioContextProvider } from "./audio/context-provider"; @@ -31,6 +32,8 @@ import { WorkletLogReceiver } from "./shared/worklet-log-receiver"; import { createSharedBuffer, isSharedBuffer } from "./shared/shared-buffer"; import { resolveContainer } from "./shared/container"; import { Reconnector } from "./reconnector"; +import { SyncModeClock } from "./sync-mode/clock"; +import { AdvertizerEvaluator } from "./advertizer/evaluator"; let scriptPath; if (document.currentScript === null) { @@ -134,7 +137,11 @@ export default class Nimio { } this._createVUMeter(); + this._advertizerEval = new AdvertizerEvaluator(this._instName); this._createLatencyController(); + if (this._config.syncBuffer > 0) { + this._createSyncModeParams(); + } this._playCb = this.play.bind(this); if (this._config.autoplay) { @@ -302,6 +309,11 @@ export default class Nimio { this._sldpManager.cancelStream(decoderFlow.trackId); }; decoderFlow.setConfig(data.config); + this._eventBus.emit("transp:track-action", { + op: "main", + id: data.trackId, + type, + }); } _createNextRenditionFlow(type, data) { @@ -365,7 +377,7 @@ export default class Nimio { // if (this._audioConfig.numberOfChannels === 0) { // this._audioConfig.numberOfChannels = frame.numberOfChannels; // if (this._audioBuffer) this._audioBuffer.reset(); - // if (!this._prepareAudioOutput(this._audioConfig.get())) { + // if (!this._prepareAudioOutput()) { // return false; // } // this._decoderFlows["audio"].setBuffer(this._audioBuffer, this._state); @@ -421,8 +433,11 @@ export default class Nimio { this._noAudio = this._config.videoOnly; if (this._audioBuffer) { this._audioBuffer.reset(); + this._audioBuffer = null; } this._latencyCtrl.reset(); + this._syncModeParams = {}; + this._advertizerEval.reset(); if (this._nextRenditionData) { if (this._nextRenditionData.decoderFlow) { @@ -467,9 +482,9 @@ export default class Nimio { } } - _prepareAudioOutput(config) { + _prepareAudioOutput() { this._logger.debug("prepareAudioOutput"); - if (!config || config.numberOfChannels < 1) { + if (this._audioConfig.numberOfChannels < 1) { if (!this._noAudio) { this._startNoAudioMode(); } @@ -481,23 +496,25 @@ export default class Nimio { this._stopAudio(); } - let AudioBufferClass = this._sabShared - ? WritableAudioBuffer - : WritableTransAudioBuffer; - this._audioBuffer = AudioBufferClass.allocate( - this._bufferSec * 2, // reserve 2 times buffer size for development (TODO: reduce later) - config.sampleRate, - config.numberOfChannels, - config.sampleCount, - ); - - this._audioBuffer.addPreprocessor( - new AudioGapsProcessor( - this._audioConfig.sampleCount, + if (!this._audioBuffer) { + let AudioBufferClass = this._sabShared + ? WritableAudioBuffer + : WritableTransAudioBuffer; + this._audioBuffer = AudioBufferClass.allocate( + this._bufferSec * 6, // reserve 6 times buffer size for development (TODO: reduce later) this._audioConfig.sampleRate, - this._logger, - ), - ); + this._audioConfig.numberOfChannels, + this._audioConfig.sampleCount, + ); + + this._audioBuffer.addPreprocessor( + new AudioGapsProcessor( + this._audioConfig.sampleCount, + this._audioConfig.sampleRate, + this._logger, + ), + ); + } return true; } @@ -558,7 +575,6 @@ export default class Nimio { videoEnabled: !this._noVideo, logLevel: this._config.logLevel, enableLogs: this._config.workletLogs, - bufferSec: this._bufferSec * 2, }; if (!idle && this._audioBuffer) { @@ -567,6 +583,9 @@ export default class Nimio { if (this._audioBuffer.isShareable) { procOptions.audioSab = this._audioBuffer.buffer; } + if (this._config.syncBuffer > 0) { + procOptions.syncBuffer = this._config.syncBuffer; + } } this._audioNode = new AudioWorkletNode( @@ -579,6 +598,8 @@ export default class Nimio { processorOptions: procOptions, }, ); + + this._audioNode.port.start(); this._workletLogReceiver.add(this._audioNode); if (!this._state.isShared()) { this._state.attachPort(this._audioNode.port); @@ -595,6 +616,13 @@ export default class Nimio { this._audioContext.resume(); } + if (this._config.syncBuffer > 0) { + let smc = new SyncModeClock(this._audioNode.port); + await smc.sync(); + this._applySyncModeParams(); + } + this._sendPendingAdvertizerActions(); + if (this._vuMeterSvc.isInitialized() && !this._vuMeterSvc.isStarted()) { this._vuMeterSvc.start(); } @@ -625,6 +653,9 @@ export default class Nimio { this._audioContext.close(); this._audioContext = this._audioNode = this._audioWorkletReady = null; } + if (this._audioBuffer) { + this._audioBuffer.reset(); + } this._setNoAudio(false); } @@ -651,12 +682,14 @@ export default class Nimio { this._config.instanceName, this._state, this._audioConfig, + this._advertizerEval, { latency: this._config.latency, tolerance: this._config.latencyTolerance, adjustMethod: this._config.latencyAdjustMethod, video: !this._noVideo, audio: !this._noAudio, + syncBuffer: this._config.syncBuffer, }, ); this._speed = 1; @@ -669,6 +702,7 @@ Object.assign(Nimio.prototype, NimioTransport); Object.assign(Nimio.prototype, NimioRenditions); Object.assign(Nimio.prototype, NimioAbr); Object.assign(Nimio.prototype, NimioVolume); +Object.assign(Nimio.prototype, NimioSyncMode); if (typeof window !== "undefined") { // Expose globally when used via