const { Readable: e } = require("streamx"),
t = require("safety-catch"),
s = require("debounceify"),
{ createTracer: r } = require("hypertrace"),
{ MEDIA_EVENTS: a, PRESENCE_EVENTS: i } = require("../../lib/common"),
n = require("./webrtc"),
o = require("./remotes-map"),
{ STREAM_TYPE: c, LOCAL_CHANGE_EVENTS: l, LOCAL_CLOSE_EVENTS: h } = require("../local/constants")
function d(e) {
;("Stream was destroyed" !== e.message && console.warn(e), t(e))
}
module.exports = class {
constructor(e, t, { onerror: a = d } = {}) {
;((this.api = e.api),
(this.local = e.local),
(this.swarmId = e.swarmId),
(this.callId = t),
(this.onerror = a),
(this.closed = !1),
(this._pendingApiTasks = []),
(this._debouncedFlushTasks = s(this._flushTasks.bind(this))),
(this._remoteStreamSubscriptions = new Map()),
(this._pendingSignals = new Map()),
(this._remotes = new o()),
(this._joinedPeers = new Set()),
(this._signalSubscription = null),
(this._proxySubscription = null),
(this._presenceSubscription = null),
(this._signalObserver = this._observeSignalEvents()),
(this._proxyObserver = this._observeProxyEvents()),
(this._presenceObserver = this._observePresenceEvents()),
(this._onremoteschange = this._onRemotesChange.bind(this)),
this._remotes.on("change", this._onremoteschange),
(this._onlocalcameratrackschange = this._handleLocalCameraTracksChange.bind(this)),
this.local.on(l.CAMERA, this._onlocalcameratrackschange),
(this._onlocalscreentrackschange = this._handleLocalScreenTracksChange.bind(this)),
this.local.on(l.SCREEN, this._onlocalscreentrackschange),
(this._onlocalstreamclosed = this._handleLocalStreamClosed.bind(this)),
this.local.on(h.CAMERA, this._onlocalstreamclosed),
this.local.on(h.SCREEN, this._onlocalstreamclosed),
(this.tracer = r(this, { props: { callId: t, swarmId: this.swarmId } })),
this._internalBackgroundRun(() => {
this._updateLocalPresence()
}))
}
async _flushTasks() {
for (; this._pendingApiTasks.length; ) {
const e = this._pendingApiTasks.shift()
try {
if ((await e(), this.closed)) return
} catch (e) {
this.onerror(e)
}
}
}
_internalBackgroundRun(e) {
;(this._pendingApiTasks.push(e), this._debouncedFlushTasks())
}
_onRemotesChange(e, t) {
const s = this._remoteStreamSubscriptions.get(e)
if (!s) return
const r = t.ready
? { videoMuted: t.videoMuted, audioMuted: t.audioMuted }
: { videoMuted: !0, audioMuted: !0 }
for (const e of s) e.push(r)
}
async _observePresenceEvents() {
;((this._presenceSubscription = this.api.subscribePresence(this.callId)),
this._presenceSubscription.on("error", this.onerror))
try {
for await (const e of this._presenceSubscription)
(this.tracer.trace(e),
e.type === i.PEER_PRESENT
? this._internalBackgroundRun(
() => (this._joinedPeers.add(e.swarmId), this.api.joinPeer(e.swarmId)),
)
: e.type === i.PEER_NOT_PRESENT
? this._internalBackgroundRun(
() => (this._joinedPeers.delete(e.swarmId), this.api.leavePeer(e.swarmId)),
)
: e.type === i.STREAM_STATUS_CHANGED
? this._remotes.updateStatus(e.swarmId, e.streamId, e.videoMuted, e.audioMuted)
: e.type === i.STREAM_UNAVAILABLE && this._remotes.deleteStatus(e.streamId))
} catch (e) {
this.onerror(e)
}
}
_updateLocalPresence(e = null, t = null) {
;((e = e || this.local.cameraStreamStatus), (t = t || this.local.screenStreamStatus))
const s = []
return (
this.local.cameraStreamOpened &&
s.push({ id: this.local.cameraStream.id, name: c.CAMERA, ...e }),
this.local.screenStreamOpened &&
s.push({ id: this.local.screenStream.id, name: c.SCREEN, ...t }),
this.api.setPresent(this.callId, { streams: s })
)
}
_handleLocalCameraTracksChange(e, t, s) {
this._internalBackgroundRun(async () => {
for (const s of this._remotes.clients())
!s.closed &&
s.ready &&
(e && (await s.setVideoTrack(this.local.videoTrack)),
t && (await s.setAudioTrack(this.local.audioTrack)))
this._updateLocalPresence(s)
})
}
_handleLocalScreenTracksChange(e, t) {
this._internalBackgroundRun(async () => {
for (const t of this._remotes.clients())
!t.closed && t.ready && e && (await t.setScreenTrack(this.local.screenTrack))
this._updateLocalPresence(null, t)
})
}
_handleLocalStreamClosed() {
this._internalBackgroundRun(() => {
this._updateLocalPresence()
})
}
async _observeProxyEvents() {
;((this._proxySubscription = this.api.subscribeProxyUpdates(this.callId)),
this._proxySubscription.on("error", this.onerror))
try {
for await (const e of this._proxySubscription)
(this.tracer.trace(e),
e.type === a.PROXY_OPEN
? this._handleProxyOpen(e.swarmId, e.proxyPort)
: e.type === a.PROXY_CLOSE && this._handleProxyClose(e.swarmId))
} catch (e) {
this.onerror(e)
}
}
_handleProxyOpen(e, t) {
const s = this.swarmId < e,
r = new n(s, e, t, this.local.cameraStream.id, this.local.screenStream.id, this._remotes, {
onerror: this.onerror,
parentTracer: this.tracer,
})
;(this._remotes.addClient(e, r),
r.on("signal", this._handleWebRtcSignal.bind(this)),
r.on("ready", this._handleWebRtcReady.bind(this)),
r.on("close", this._handleWebRtcClose.bind(this)))
const a = this._pendingSignals.get(e)
a && (this._pendingSignals.delete(e), r.handleSignalMessage(a))
}
_handleProxyClose(e) {
const t = this._remotes.getClient(e)
t && (this._remotes.deleteClient(e), this._pendingSignals.delete(e), t.close())
}
async _observeSignalEvents() {
;((this._signalSubscription = this.api.subscribeMediaSignals(this.callId)),
this._signalSubscription.on("error", this.onerror))
try {
for await (const { swarmId: e, signal: t } of this._signalSubscription) {
this.tracer.trace({ swarmId: e, signal: t })
const s = this._remotes.getClient(e)
s ? s.handleSignalMessage(t) : this._pendingSignals.set(e, t)
}
} catch (e) {
this.onerror(e)
}
}
async _handleWebRtcReady(e) {
;(this._remotes.updateStatus(e.swarmId, e.remoteCameraStreamId),
this._remotes.updateStatus(e.swarmId, e.remoteScreenStreamId),
this._internalBackgroundRun(async () => {
;(await e.setVideoTrack(this.local.videoTrack),
await e.setAudioTrack(this.local.audioTrack),
await e.setScreenTrack(this.local.screenTrack),
this._updateLocalPresence())
}))
}
_handleWebRtcSignal(e, t) {
this._internalBackgroundRun(() => this.api.signalMedia(this.callId, e.swarmId, t))
}
_handleWebRtcClose(e) {
;(null !== e.remoteCameraStreamId &&
(this._cleanupRemoteStream(e.remoteCameraStreamId),
this._remoteStreamSubscriptions.delete(e.remoteCameraStreamId)),
null !== e.remoteScreenStreamId &&
(this._cleanupRemoteStream(e.remoteScreenStreamId),
this._remoteStreamSubscriptions.delete(e.remoteScreenStreamId)))
}
_cleanupRemoteStream(e) {
const t = this._remotes._statusById.get(e)
if (t) for (const e of t.stream.getTracks()) t.stream.removeTrack(e)
}
subscribeRemoteStream(t) {
const s = this,
r = new e({
destroy: function (e) {
const a = s._remoteStreamSubscriptions.get(t)
if (!a) return e(null)
const i = a.indexOf(r)
return -1 === i
? e(null)
: (a.splice(i, 1),
a.length > 0 ? e(null) : (s._remoteStreamSubscriptions.delete(t), void e(null)))
},
})
let a = this._remoteStreamSubscriptions.get(t)
;(a || ((a = []), this._remoteStreamSubscriptions.set(t, a)), a.push(r))
const { videoMuted: i, audioMuted: n } = this._remotes.getStatus(t)
return (r.push({ videoMuted: i, audioMuted: n }), r)
}
getRemoteSrcObject(e) {
return this._remotes.getStatus(e).stream
}
close() {
if (this.closed) return
;(this.tracer.trace(),
(this.closed = !0),
this._remotes.removeListener("change", this._onremoteschange),
this.local.removeListener(l.CAMERA, this._onlocalcameratrackschange),
this.local.removeListener(l.SCREEN, this._onlocalscreentrackschange),
this.local.removeListener(h.CAMERA, this._onlocalstreamclosed),
this.local.removeListener(h.SCREEN, this._onlocalstreamclosed),
this._signalSubscription.destroy(),
this._proxySubscription.destroy(),
this._presenceSubscription.destroy())
for (const e of this._remotes.clients()) e.close()
let e = !1
;(this.local.cameraStreamOpened &&
((e = !0), this.local.closeStream(this.local.cameraStream.id)),
this.local.screenStreamOpened &&
((e = !0), this.local.closeStream(this.local.screenStream.id)),
this._internalBackgroundRun(async () => {
e && (await this._updateLocalPresence())
for (const e of this._joinedPeers) await this.api.leavePeer(e)
await this.api.setNotPresent(this.callId)
}))
}
}