import RPC from 'bare-rpc'
// cross-worker will be "pear-run" in Pear v2 can be removed when is out
import { spawn } from 'cross-worker-pear'
import { RPCCommands } from './rpc-commands'
export class Call {
#workerLink
#videoChromium
#live
#decoder
#rpc
#sampleRate
#renderer
#audioCtx
#audioPlaybackStartTime
#videoPlaybackStartTime
#videoTimeBase
#audioTimeBase
#videoTimeouts
#audioNodes
#onMetadata
#onStats
#onFrame
#onStop
#onState
#maxVideoDelay = 0
#maxAudioDelay = 0
constructor({
workerLink,
videoChromium,
onMetadata,
onStats,
onFrame,
onStop,
onState
}) {
this.#workerLink =
workerLink || './node_modules/@holepunchto/broadcast-call/workers'
this.#videoChromium = videoChromium
this.#decoder = new TextDecoder()
this.#videoTimeouts = new Set()
this.#audioNodes = new Set()
this.#onMetadata = onMetadata
this.#onStats = onStats
this.#onFrame = onFrame
this.#onStop = onStop
this.#onState = onState
}
isCallStarted() {
return !!this.#rpc
}
// known codecs
// codecs="av01.0.08M.08, opus" test/integration/avsynctest.webm
// codecs="vp8, vorbis" canada army video
// codecs="vp9, opus" OBS on mac
// codecs="av01.0.05M.08, opus" ffmpeg test generation
async startCall(callKey, { chromium, live, codecs } = {}) {
codecs = codecs || 'av01.0.05M.08, opus'
if (this.isCallStarted()) return
if (!callKey) {
const writeWorker = await spawn(`${this.#workerLink}/write.js`)
this.#rpc = new RPC(writeWorker.IPC)
return
}
if (chromium) {
const mediaSource = new MediaSource()
this.#videoChromium.src = URL.createObjectURL(mediaSource)
mediaSource.addEventListener('sourceopen', async () => {
if (mediaSource.sourceBuffers.length > 0) {
return
}
const mime = `video/webm; codecs="${codecs}"` // passthrough generator, gitlab video
if (!MediaSource.isTypeSupported(mime)) {
throw new Error(`MIME type or codec not supported: ${mime}`)
}
const sourceBuffer = mediaSource.addSourceBuffer(mime)
const queue = []
this.#onMetadata?.({})
this.#videoChromium.play().catch((err) => {
console.error(err)
})
const processData = () => {
if (this.#videoChromium.error) throw this.#videoChromium.error
if (sourceBuffer.updating) return
if (queue.length === 0) return
const nextChunk = queue.shift()
try {
sourceBuffer.appendBuffer(nextChunk)
} catch (err) {
if (err.name === 'QuotaExceededError') {
queue.unshift(nextChunk)
// Remove till 10s before current time of playback.
const start = sourceBuffer.buffered.start(0)
const removeBefore = this.#videoChromium.currentTime - 10
if (removeBefore <= start) return
sourceBuffer.remove(start, removeBefore)
} else {
throw err
}
}
}
let didSeekToStartTime = false
sourceBuffer.addEventListener('updateend', () => {
processData()
if (!didSeekToStartTime && sourceBuffer.buffered.length > 0) {
const start = sourceBuffer.buffered.start(0)
this.#videoChromium.currentTime = start
didSeekToStartTime = true
}
})
const chromiumWorker = await spawn(`${this.#workerLink}/chromium.js`)
this.#rpc = new RPC(chromiumWorker.IPC, async (req) => {
switch (req.command) {
case RPCCommands.VIDEO_CHROMIUM:
queue.push(req.data)
processData()
break
case RPCCommands.STATS:
const decodedStatsData = this.#decoder.decode(req.data)
this.#onStats?.(decodedStatsData)
break
case RPCCommands.STATE:
const state = this.#decoder.decode(req.data)
this.#onState?.({ state })
break
default:
throw new Error(`Unexpected command: ${req.command}`)
}
})
const startRequest = this.#rpc.request(RPCCommands.START)
const startData = {
key: callKey,
live
}
startRequest.send(JSON.stringify(startData))
})
return
}
let videoPts
let audioPts
const readWorker = await spawn(`${this.#workerLink}/read.js`)
this.#rpc = new RPC(readWorker.IPC, (req) => {
switch (req.command) {
case RPCCommands.METADATA:
const decodedMetadata = this.#decoder.decode(req.data)
const message = JSON.parse(decodedMetadata)
const { width, height, sampleRate, videoTimeBase, audioTimeBase } =
message
this.#sampleRate = sampleRate
this.#videoTimeBase = videoTimeBase
this.#audioTimeBase = audioTimeBase
this.#onMetadata?.({ width, height })
break
case RPCCommands.STATS:
const decodedStatsData = this.#decoder.decode(req.data)
this.#onStats?.(decodedStatsData)
break
case RPCCommands.VIDEO_PTS:
const decodedVideoPtsData = this.#decoder.decode(req.data)
videoPts = Number(decodedVideoPtsData)
break
case RPCCommands.VIDEO:
this.#scheduleVideo(req.data, videoPts)
videoPts = undefined
req.reply()
break
case RPCCommands.AUDIO_PTS:
const decodedAudioPtsData = this.#decoder.decode(req.data)
audioPts = Number(decodedAudioPtsData)
break
case RPCCommands.AUDIO:
this.#scheduleAudio(req.data, audioPts)
audioPts = undefined
req.reply()
break
default:
throw new Error(`Unexpected command: ${req.command}`)
}
})
const startRequest = this.#rpc.request(RPCCommands.START)
const startData = {
key: callKey,
live
}
startRequest.send(JSON.stringify(startData))
this.#audioCtx = new AudioContext()
}
stopCall() {
if (!this.isCallStarted()) return
const stopRequest = this.#rpc.request(RPCCommands.STOP)
stopRequest.send()
this.#rpc = null
if (this.#videoChromium) {
this.#videoChromium.pause()
this.#videoChromium.currentTime = 0
} else {
for (const timeout of this.#videoTimeouts) {
clearTimeout(timeout)
}
this.#videoTimeouts.clear()
for (const audioNode of this.#audioNodes) {
audioNode.stop()
}
this.#audioNodes.clear()
this.#audioCtx = null
this.#renderer = null
}
this.#onStop()
}
destroy() {
if (this.isCallStarted()) {
this.stopCall()
}
}
#playAudio(buffer, delayInSeconds = 0) {
if (!this.#audioCtx) return
const dataView = new DataView(
buffer.buffer,
buffer.byteOffset,
buffer.byteLength
)
const floatBuffer = new Float32Array(buffer.length)
const MAX_UINT8 = 2 ** 8 - 1
for (let i = 0; i < floatBuffer.length; i++) {
floatBuffer[i] = (2 * dataView.getUint8(i)) / MAX_UINT8 - 1
}
const channels = 1
const length = floatBuffer.length / channels
const audioBuffer = this.#audioCtx.createBuffer(
channels,
length,
this.#sampleRate
)
for (let c = 0; c < channels; c++) {
const channelData = audioBuffer.getChannelData(c)
for (let i = 0; i < length; i++) {
channelData[i] = floatBuffer[i * channels + c]
}
}
const source = this.#audioCtx.createBufferSource()
source.buffer = audioBuffer
source.connect(this.#audioCtx.destination)
source.start(this.#audioCtx.currentTime + delayInSeconds)
source.onended = () => {
this.#audioNodes.delete(source)
}
this.#audioNodes.add(source)
}
#scheduleVideo(buffer, pts) {
if (!this.#audioCtx) return
const ptsSeconds =
(pts * this.#videoTimeBase.numerator) / this.#videoTimeBase.denominator
// Use audio clock for sync
const nowSeconds = this.#audioCtx.currentTime
if (this.#videoPlaybackStartTime === undefined) {
this.#videoPlaybackStartTime = nowSeconds - ptsSeconds
}
let delay = this.#videoPlaybackStartTime - nowSeconds + ptsSeconds
if (delay < 0 && this.#maxVideoDelay < -delay) {
this.#maxVideoDelay = -delay
}
delay += this.#maxVideoDelay
const timeout = setTimeout(() => {
this.#onFrame(buffer)
this.#videoTimeouts.delete(timeout)
}, delay * 1000)
this.#videoTimeouts.add(timeout)
}
#scheduleAudio(buffer, pts) {
if (!this.#audioCtx) return
const ptsSeconds =
(pts * this.#audioTimeBase.numerator) / this.#audioTimeBase.denominator
// Use audio clock for sync
const nowSeconds = this.#audioCtx.currentTime
if (this.#audioPlaybackStartTime === undefined) {
this.#audioPlaybackStartTime = nowSeconds - ptsSeconds
}
let delay = this.#audioPlaybackStartTime - nowSeconds + ptsSeconds
if (delay < 0 && this.#maxAudioDelay < -delay) {
this.#maxAudioDelay = -delay
}
delay += this.#maxAudioDelay
this.#playAudio(buffer, delay)
}
}
// in keet, demuxWorkerLink should be
// `${CALL_WORKER_LINK}/node_modules/@holepunchto/vcr-core/util/workers`
export class Record {
#workerLink
#demuxWorkerLink
#decoder
#rpc
#onStart
#onStop
#outputStream
#streamController
// Add method to set stream controller
setStreamController(controller) {
this.#streamController = controller
}
constructor({ workerLink, demuxWorkerLink, onStart, onStop }) {
this.#workerLink =
workerLink || './node_modules/@holepunchto/broadcast-call/workers'
this.#demuxWorkerLink =
demuxWorkerLink ||
'./workers/node_modules/@holepunchto/vcr-core/util/workers'
this.#decoder = new TextDecoder()
this.#onStart = onStart
this.#onStop = onStop
}
isCallStarted() {
return !!this.#rpc
}
async start(opts = { fragmentDuration: 1000 }) {
if (this.isCallStarted()) return
const worker = await spawn(`${this.#workerLink}/obs.js`)
this.#rpc = new RPC(worker.IPC, async (req) => {
switch (req.command) {
case RPCCommands.METADATA:
const decodedMetadata = this.#decoder.decode(req.data)
const message = JSON.parse(decodedMetadata)
this.#onStart?.(message)
break
case RPCCommands.STATE:
const state = this.#decoder.decode(req.data)
this.#onStop?.(state)
// weird but we have to close out the server
this.#rpc.request(RPCCommands.STOP).send()
break
case RPCCommands.VIDEO_CHROMIUM:
// Push data to the stream
if (this.#streamController) {
this.#streamController.enqueue(req.data)
}
break
case RPCCommands.EXPORT_COMPLETE:
if (this.#streamController) {
this.#streamController.close()
this.#streamController = null
}
break
default:
throw new Error(`Unexpected command: ${req.command}`)
}
})
const startRequest = this.#rpc.request(RPCCommands.START)
const startData = {
demuxWorkerLink: this.#demuxWorkerLink,
fragmentDuration: opts.fragmentDuration
}
startRequest.send(JSON.stringify(startData))
return
}
async stop() {
if (!this.isCallStarted()) return
this.#rpc.request(RPCCommands.STOP).send()
this.#onStop?.()
}
async saveRecording(stream) {
this.#outputStream = stream
// Start the export process
this.#rpc.request(RPCCommands.VIDEO).send()
}
async destroy() {
if (this.isCallStarted()) {
this.#rpc.request(RPCCommands.CLEANUP).send()
this.#rpc = null
}
}
}