Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Experimental feature of audio muxing #614

Draft
wants to merge 3 commits into
base: main
Choose a base branch
from
+358 −1
Draft

Experimental feature of audio muxing #614

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3a38f52
Worked for merge
cnderrauber Feb 28, 2023
25efb6e
work with webaudio
cnderrauber Mar 10, 2023
71c1bca
Merge remote-tracking branch 'origin/main' into asf
cnderrauber Mar 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/api/SignalClient.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import {
} from '../proto/livekit_models';
import {
AddTrackRequest,
AudioTrackMuxUpdate,
ConnectionQualityUpdate,
JoinResponse,
LeaveRequest,
Expand Down Expand Up @@ -134,6 +135,8 @@ export class SignalClient {

onLeave?: (leave: LeaveRequest) => void;

onAudioMuxUpdate?: (update: AudioTrackMuxUpdate) => void;

connectOptions?: ConnectOpts;

ws?: WebSocket;
Expand Down Expand Up @@ -588,6 +591,10 @@ export class SignalClient {
}
} else if (msg.$case === 'pong') {
this.resetPingTimeout();
} else if (msg.$case === 'audioMuxUpdate') {
if (this.onAudioMuxUpdate) {
this.onAudioMuxUpdate(msg.audioMuxUpdate);
}
} else if (msg.$case === 'pongResp') {
this.rtt = Date.now() - msg.pongResp.lastPingTimestamp;
this.resetPingTimeout();
Expand Down
159 changes: 158 additions & 1 deletion src/proto/livekit_rtc.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,8 @@ export interface SignalResponse {
| { $case: "trackUnpublished"; trackUnpublished: TrackUnpublishedResponse }
| { $case: "pong"; pong: number }
| { $case: "reconnect"; reconnect: ReconnectResponse }
| { $case: "pongResp"; pongResp: Pong };
| { $case: "pongResp"; pongResp: Pong }
| { $case: "audioMuxUpdate"; audioMuxUpdate: AudioTrackMuxUpdate };
}

export interface SimulcastCodec {
Expand Down Expand Up @@ -406,6 +407,16 @@ export interface SimulateScenario {
| { $case: "switchCandidateProtocol"; switchCandidateProtocol: CandidateProtocol };
}

export interface AudioTrackMuxUpdate {
audioTrackMuxes: AudioTrackMuxInfo[];
}

export interface AudioTrackMuxInfo {
sdpTrackId: string;
trackSid: string;
participantSid: string;
}

export interface Ping {
timestamp: number;
/** rtt in milliseconds calculated by client */
Expand Down Expand Up @@ -794,6 +805,9 @@ export const SignalResponse = {
Pong.encode(message.message.pongResp, writer.uint32(162).fork()).ldelim();
break;
}
if (message.message?.$case === "audioMuxUpdate") {
AudioTrackMuxUpdate.encode(message.message.audioMuxUpdate, writer.uint32(170).fork()).ldelim();
}
return writer;
},

Expand Down Expand Up @@ -882,6 +896,12 @@ export const SignalResponse = {
case 20:
message.message = { $case: "pongResp", pongResp: Pong.decode(reader, reader.uint32()) };
break;
case 21:
message.message = {
$case: "audioMuxUpdate",
audioMuxUpdate: AudioTrackMuxUpdate.decode(reader, reader.uint32()),
};
break;
default:
reader.skipType(tag & 7);
break;
Expand Down Expand Up @@ -936,6 +956,8 @@ export const SignalResponse = {
? { $case: "reconnect", reconnect: ReconnectResponse.fromJSON(object.reconnect) }
: isSet(object.pongResp)
? { $case: "pongResp", pongResp: Pong.fromJSON(object.pongResp) }
: isSet(object.audioMuxUpdate)
? { $case: "audioMuxUpdate", audioMuxUpdate: AudioTrackMuxUpdate.fromJSON(object.audioMuxUpdate) }
: undefined,
};
},
Expand Down Expand Up @@ -987,6 +1009,9 @@ export const SignalResponse = {
(obj.reconnect = message.message?.reconnect ? ReconnectResponse.toJSON(message.message?.reconnect) : undefined);
message.message?.$case === "pongResp" &&
(obj.pongResp = message.message?.pongResp ? Pong.toJSON(message.message?.pongResp) : undefined);
message.message?.$case === "audioMuxUpdate" && (obj.audioMuxUpdate = message.message?.audioMuxUpdate
? AudioTrackMuxUpdate.toJSON(message.message?.audioMuxUpdate)
: undefined);
return obj;
},

Expand Down Expand Up @@ -1122,6 +1147,16 @@ export const SignalResponse = {
) {
message.message = { $case: "pongResp", pongResp: Pong.fromPartial(object.message.pongResp) };
}
if (
object.message?.$case === "audioMuxUpdate" &&
object.message?.audioMuxUpdate !== undefined &&
object.message?.audioMuxUpdate !== null
) {
message.message = {
$case: "audioMuxUpdate",
audioMuxUpdate: AudioTrackMuxUpdate.fromPartial(object.message.audioMuxUpdate),
};
}
return message;
},
};
Expand Down Expand Up @@ -3572,6 +3607,128 @@ export const SimulateScenario = {
},
};

function createBaseAudioTrackMuxUpdate(): AudioTrackMuxUpdate {
return { audioTrackMuxes: [] };
}

export const AudioTrackMuxUpdate = {
encode(message: AudioTrackMuxUpdate, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
for (const v of message.audioTrackMuxes) {
AudioTrackMuxInfo.encode(v!, writer.uint32(10).fork()).ldelim();
}
return writer;
},

decode(input: _m0.Reader | Uint8Array, length?: number): AudioTrackMuxUpdate {
const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
let end = length === undefined ? reader.len : reader.pos + length;
const message = createBaseAudioTrackMuxUpdate();
while (reader.pos < end) {
const tag = reader.uint32();
switch (tag >>> 3) {
case 1:
message.audioTrackMuxes.push(AudioTrackMuxInfo.decode(reader, reader.uint32()));
break;
default:
reader.skipType(tag & 7);
break;
}
}
return message;
},

fromJSON(object: any): AudioTrackMuxUpdate {
return {
audioTrackMuxes: Array.isArray(object?.audioTrackMuxes)
? object.audioTrackMuxes.map((e: any) => AudioTrackMuxInfo.fromJSON(e))
: [],
};
},

toJSON(message: AudioTrackMuxUpdate): unknown {
const obj: any = {};
if (message.audioTrackMuxes) {
obj.audioTrackMuxes = message.audioTrackMuxes.map((e) => e ? AudioTrackMuxInfo.toJSON(e) : undefined);
} else {
obj.audioTrackMuxes = [];
}
return obj;
},

fromPartial<I extends Exact<DeepPartial<AudioTrackMuxUpdate>, I>>(object: I): AudioTrackMuxUpdate {
const message = createBaseAudioTrackMuxUpdate();
message.audioTrackMuxes = object.audioTrackMuxes?.map((e) => AudioTrackMuxInfo.fromPartial(e)) || [];
return message;
},
};

function createBaseAudioTrackMuxInfo(): AudioTrackMuxInfo {
return { sdpTrackId: "", trackSid: "", participantSid: "" };
}

export const AudioTrackMuxInfo = {
encode(message: AudioTrackMuxInfo, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
if (message.sdpTrackId !== "") {
writer.uint32(10).string(message.sdpTrackId);
}
if (message.trackSid !== "") {
writer.uint32(18).string(message.trackSid);
}
if (message.participantSid !== "") {
writer.uint32(26).string(message.participantSid);
}
return writer;
},

decode(input: _m0.Reader | Uint8Array, length?: number): AudioTrackMuxInfo {
const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
let end = length === undefined ? reader.len : reader.pos + length;
const message = createBaseAudioTrackMuxInfo();
while (reader.pos < end) {
const tag = reader.uint32();
switch (tag >>> 3) {
case 1:
message.sdpTrackId = reader.string();
break;
case 2:
message.trackSid = reader.string();
break;
case 3:
message.participantSid = reader.string();
break;
default:
reader.skipType(tag & 7);
break;
}
}
return message;
},

fromJSON(object: any): AudioTrackMuxInfo {
return {
sdpTrackId: isSet(object.sdpTrackId) ? String(object.sdpTrackId) : "",
trackSid: isSet(object.trackSid) ? String(object.trackSid) : "",
participantSid: isSet(object.participantSid) ? String(object.participantSid) : "",
};
},

toJSON(message: AudioTrackMuxInfo): unknown {
const obj: any = {};
message.sdpTrackId !== undefined && (obj.sdpTrackId = message.sdpTrackId);
message.trackSid !== undefined && (obj.trackSid = message.trackSid);
message.participantSid !== undefined && (obj.participantSid = message.participantSid);
return obj;
},

fromPartial<I extends Exact<DeepPartial<AudioTrackMuxInfo>, I>>(object: I): AudioTrackMuxInfo {
const message = createBaseAudioTrackMuxInfo();
message.sdpTrackId = object.sdpTrackId ?? "";
message.trackSid = object.trackSid ?? "";
message.participantSid = object.participantSid ?? "";
return message;
},
};

function createBasePing(): Ping {
return { timestamp: 0, rtt: 0 };
}
Expand Down
12 changes: 12 additions & 0 deletions src/room/RTCEngine.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,12 @@ enum PCState {
Closed,
}

interface AudioMuxTrack {
track: MediaStreamTrack;
stream: MediaStream;
receiver: RTCRtpReceiver;
}

/** @internal */
export default class RTCEngine extends (EventEmitter as new () => TypedEventEmitter<EngineEventCallbacks>) {
publisher?: PCTransport;
Expand All @@ -73,6 +79,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit

peerConnectionTimeout: number = roomConnectOptionDefaults.peerConnectionTimeout;

audioMuxTracks: Map<string, AudioMuxTrack> = new Map();

get isClosed() {
return this._isClosed;
}
Expand Down Expand Up @@ -361,6 +369,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
};

this.subscriber.pc.ontrack = (ev: RTCTrackEvent) => {
// todo-mux: firefox can't get TR_AX... track id
if (ev.track.kind === 'audio' && ev.track.id.includes('TR_AX')) {
this.audioMuxTracks.set(ev.track.id, {track: ev.track, stream: ev.streams[0], receiver: ev.receiver})
}
this.emit(EngineEvent.MediaTrackAdded, ev.track, ev.streams[0], ev.receiver);
};

Expand Down
65 changes: 65 additions & 0 deletions src/room/Room.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import {
UserPacket,
} from '../proto/livekit_models';
import {
AudioTrackMuxUpdate,
ConnectionQualityUpdate,
JoinResponse,
SimulateScenario,
Expand Down Expand Up @@ -50,6 +51,7 @@ import LocalTrackPublication from './track/LocalTrackPublication';
import LocalVideoTrack from './track/LocalVideoTrack';
import type RemoteTrack from './track/RemoteTrack';
import RemoteTrackPublication from './track/RemoteTrackPublication';
import MuxedRemoteAudioTrack from './track/MuxedRemoteAudioTrack';
import { Track } from './track/Track';
import type { TrackPublication } from './track/TrackPublication';
import type { AdaptiveStreamSettings } from './track/types';
Expand Down Expand Up @@ -133,6 +135,11 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)

private disconnectLock: Mutex;

private lastMuxUpdate?: AudioTrackMuxUpdate;

/** mapping of track id -> MuxedRemoteAudioTrack */
private muxedTracks: Map<string, MuxedRemoteAudioTrack> = new Map();

/**
* Creates a new Room, the primary construct for a LiveKit session.
* @param options
Expand Down Expand Up @@ -177,6 +184,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
this.engine.client.onStreamStateUpdate = this.handleStreamStateUpdate;
this.engine.client.onSubscriptionPermissionUpdate = this.handleSubscriptionPermissionUpdate;
this.engine.client.onConnectionQuality = this.handleConnectionQualityUpdate;
this.engine.client.onAudioMuxUpdate = this.handleAudioMuxUpdate;

this.engine
.on(
Expand Down Expand Up @@ -657,6 +665,8 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
this.options.audioOutput.deviceId = prevDeviceId;
throw e;
}

// TODO: set audio output for room audio tracks
}
}

Expand Down Expand Up @@ -1058,6 +1068,61 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
});
};

private handleAudioMuxUpdate = (update: AudioTrackMuxUpdate) => {

const newlySubTracks = update.audioTrackMuxes.filter(val => !(this.lastMuxUpdate?.audioTrackMuxes.some(lval => lval.trackSid === val.trackSid)));
const newlyUnsubTracks = this.lastMuxUpdate?.audioTrackMuxes.filter(lval => !(update.audioTrackMuxes.some(val => val.trackSid === lval.trackSid)));
this.lastMuxUpdate = update;

newlySubTracks.forEach(track => {
const muxTrack = this.engine.audioMuxTracks.get(track.sdpTrackId);
if (track.sdpTrackId !== '' && !muxTrack) {
log.error(`can't find mux track for sdp track id ${track.sdpTrackId}`);
}

const remoteTrack = new MuxedRemoteAudioTrack(track.trackSid, muxTrack?.track,muxTrack?.stream, muxTrack?.receiver, this.audioContext, this.options.audioOutput);

const participant = this.participants.get(track.participantSid);
if (!participant) {
log.error(`can't find participant for ${track.participantSid}`);
return;
}

participant.addMuxAudioTrack(remoteTrack, track.trackSid);
this.muxedTracks.set(track.trackSid, remoteTrack);
});

newlyUnsubTracks?.forEach(track => {
log.info(`unsub track ${track.trackSid}`);
const muxedTrack = this.muxedTracks.get(track.trackSid);
if (!muxedTrack) {
log.error(`can't find muxed track for participant ${track.participantSid}, track ${track.trackSid}`);
return
}
muxedTrack.close();
this.muxedTracks.delete(track.trackSid);
});

update.audioTrackMuxes.forEach(track => {
const muxedTrack = this.muxedTracks.get(track.trackSid);
if(!muxedTrack) {
log.error(`can't find muxed track for participant ${track.participantSid}, track ${track.trackSid}`);
return
}
if (track.sdpTrackId === '') {
muxedTrack.unbind();
} else {
const muxTrack = this.engine.audioMuxTracks.get(track.sdpTrackId)
if (!muxTrack) {
log.error(`can't find mux track for sdp track id ${track.sdpTrackId}`);
return
}
log.info(`track ${track.trackSid} muxed to ${track.sdpTrackId}`);
muxedTrack.bind(muxTrack.track, muxTrack.stream,muxTrack.receiver);
}
});
}

private async acquireAudioContext() {
if (
typeof this.options.expWebAudioMix !== 'boolean' &&
Expand Down
Loading