diff --git a/.changeset/fix-encrypted-audio-codec-selection.md b/.changeset/fix-encrypted-audio-codec-selection.md new file mode 100644 index 000000000..d0300d159 --- /dev/null +++ b/.changeset/fix-encrypted-audio-codec-selection.md @@ -0,0 +1,5 @@ +--- +"client-sdk-android": patch +--- + +Support per-subscriber RED and Opus codec selection for end-to-end encrypted audio tracks. diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt b/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt index bc3fc77b4..cbb470091 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/e2ee/E2EEManager.kt @@ -48,7 +48,7 @@ constructor( dataPacketCryptorManagerFactory: DataPacketCryptorManager.Factory, ) { private var room: Room? = null - private val frameCryptors = mutableMapOf, FrameCryptor>() + private val frameCryptors = mutableMapOf, MutableList>() private var algorithm: FrameCryptorAlgorithm = FrameCryptorAlgorithm.AES_GCM private lateinit var emitEvent: (roomEvent: RoomEvent) -> Unit? @@ -57,9 +57,8 @@ constructor( var enabled: Boolean = false set(value) { field = value - for (item in frameCryptors.entries) { - val frameCryptor = item.value - frameCryptor.isEnabled = enabled + for (cryptors in frameCryptors.values) { + cryptors.forEach { it.isEnabled = enabled } } } @@ -129,12 +128,7 @@ constructor( fun removeSubscribedTrack(track: Track, publication: TrackPublication, participant: RemoteParticipant, room: Room) { val trackId = publication.sid val participantId = participant.identity - val frameCryptor = frameCryptors.get(trackId to participantId) - if (frameCryptor != null) { - frameCryptor.isEnabled = false - frameCryptor.dispose() - frameCryptors.remove(trackId to participantId) - } + removeFrameCryptors(trackId, participantId) } fun addPublishedTrack(track: Track, publication: TrackPublication, participant: LocalParticipant, room: Room) { @@ -146,6 +140,15 @@ constructor( } } ?: throw IllegalArgumentException("rtpSender is null") + addPublishedSender(rtpSender, publication, participant) + } + + internal fun addPublishedSender( + rtpSender: RtpSender, + publication: TrackPublication, + participant: LocalParticipant, + ) { + val room = room ?: return val frameCryptor = addRtpSender(rtpSender, participant.identity!!, publication.sid, publication.track!!.kind.name.lowercase()) frameCryptor.setObserver { trackId, state -> LKLog.i { "Sender::onFrameCryptionStateChanged: $trackId, state: $state" } @@ -164,11 +167,14 @@ constructor( fun removePublishedTrack(track: Track, publication: TrackPublication, participant: LocalParticipant, room: Room) { val trackId = publication.sid val participantId = participant.identity - val frameCryptor = frameCryptors.get(trackId to participantId) - if (frameCryptor != null) { + removeFrameCryptors(trackId, participantId) + } + + private fun removeFrameCryptors(trackId: String, participantId: Participant.Identity?) { + if (participantId == null) return + frameCryptors.remove(trackId to participantId)?.forEach { frameCryptor -> frameCryptor.isEnabled = false frameCryptor.dispose() - frameCryptors.remove(trackId to participantId) } } @@ -194,7 +200,7 @@ constructor( keyProvider.rtcKeyProvider, ) - frameCryptors[trackId to participantId] = frameCryptor + frameCryptors.getOrPut(trackId to participantId, ::mutableListOf).add(frameCryptor) frameCryptor.isEnabled = enabled frameCryptor.keyIndex = keyProvider.getLatestKeyIndex(participantId.value) return frameCryptor @@ -209,7 +215,7 @@ constructor( keyProvider.rtcKeyProvider, ) - frameCryptors[trackId to participantId] = frameCryptor + frameCryptors.getOrPut(trackId to participantId, ::mutableListOf).add(frameCryptor) frameCryptor.isEnabled = enabled frameCryptor.keyIndex = keyProvider.getLatestKeyIndex(participantId.value) return frameCryptor @@ -232,9 +238,7 @@ constructor( } internal fun cleanup() { - for (frameCryptor in frameCryptors.values) { - frameCryptor.dispose() - } + frameCryptors.values.flatten().forEach(FrameCryptor::dispose) frameCryptors.clear() } diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt index e0e64e7db..df12d8550 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt @@ -1057,6 +1057,7 @@ internal constructor( fun onUserPacket(packet: LivekitModels.UserPacket, kind: LivekitModels.DataPacket.Kind, encryptionType: LivekitModels.Encryption.Type) fun onStreamStateUpdate(streamStates: List) fun onSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate) + fun onSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate: LivekitRtc.SubscribedAudioCodecUpdate) fun onSubscriptionPermissionUpdate(subscriptionPermissionUpdate: LivekitRtc.SubscriptionPermissionUpdate) fun onSubscriptionError(subscriptionResponse: LivekitRtc.SubscriptionResponse) fun onSignalConnected(isResume: Boolean) @@ -1293,6 +1294,10 @@ internal constructor( listener?.onSubscribedQualityUpdate(subscribedQualityUpdate) } + override fun onSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate: LivekitRtc.SubscribedAudioCodecUpdate) { + listener?.onSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate) + } + override fun onSubscriptionPermissionUpdate(subscriptionPermissionUpdate: LivekitRtc.SubscriptionPermissionUpdate) { listener?.onSubscriptionPermissionUpdate(subscriptionPermissionUpdate) } diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt index 10aa87821..d6380f571 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt @@ -1416,6 +1416,13 @@ constructor( localParticipant.handleSubscribedQualityUpdate(subscribedQualityUpdate) } + /** + * @suppress + */ + override fun onSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate: LivekitRtc.SubscribedAudioCodecUpdate) { + localParticipant.handleSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate) + } + /** * @suppress */ diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt index 3f2432705..3eafe3aac 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt @@ -861,7 +861,7 @@ constructor( } LivekitRtc.SignalResponse.MessageCase.SUBSCRIBED_AUDIO_CODEC_UPDATE -> { - // TODO + listener?.onSubscribedAudioCodecUpdate(response.subscribedAudioCodecUpdate) } LivekitRtc.SignalResponse.MessageCase.PUBLISH_DATA_TRACK_RESPONSE -> { @@ -965,6 +965,7 @@ constructor( fun onError(error: Throwable) fun onStreamStateUpdate(streamStates: List) fun onSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate) + fun onSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate: LivekitRtc.SubscribedAudioCodecUpdate) fun onSubscriptionPermissionUpdate(subscriptionPermissionUpdate: LivekitRtc.SubscriptionPermissionUpdate) fun onSubscriptionError(subscriptionResponse: LivekitRtc.SubscriptionResponse) fun onRefreshToken(token: String) diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt index 75ac69c11..b2b957f26 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt @@ -62,6 +62,7 @@ import io.livekit.android.rpc.RpcError import io.livekit.android.util.LKLog import io.livekit.android.util.flow import io.livekit.android.util.rethrowIfCancellationSignal +import io.livekit.android.webrtc.setAudioCodecPreferences import io.livekit.android.webrtc.sortVideoCodecPreferences import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Job @@ -142,6 +143,8 @@ internal constructor( private val sourcePubLocks = Track.Source.entries.associateWith { Mutex() } internal val enabledPublishVideoCodecs = Collections.synchronizedList(mutableListOf()) + private val pendingSubscribedAudioCodecUpdates = + Collections.synchronizedMap(mutableMapOf()) private var defaultAudioTrack: LocalAudioTrack? = null private var defaultVideoTrack: LocalVideoTrack? = null @@ -482,6 +485,8 @@ internal constructor( return false } + val encryptedCodecSimulcast = engine.e2EEManager != null && dynacast && options.red + track.codec = if (options.red) AUDIO_CODEC_RED else AUDIO_CODEC_OPUS val encodings = listOf( RtpParameters.Encoding(null, true, null).apply { if (options.audioBitrate != null && options.audioBitrate > 0) { @@ -497,6 +502,19 @@ internal constructor( requestConfig = { disableDtx = !options.dtx disableRed = !options.red + if (encryptedCodecSimulcast) { + addSimulcastCodecs( + SimulcastCodec.newBuilder() + .setCodec(AUDIO_CODEC_RED) + .setCid(track.rtcTrack.id()) + .build(), + ) + addSimulcastCodecs( + SimulcastCodec.newBuilder() + .setCodec(AUDIO_CODEC_OPUS) + .build(), + ) + } addAllAudioFeatures(options.getFeaturesList()) source = options.source?.toProto() ?: LivekitModels.TrackSource.MICROPHONE }, @@ -769,6 +787,9 @@ internal constructor( (track as LocalVideoTrack).codec = finalOptions.videoCodec transceiver.applyDegradationPreference(finalOptions.degradationPreference, trackSource) + } else if (finalOptions is AudioTrackPublishOptions) { + val audioTrack = track as LocalAudioTrack + transceiver.setAudioCodecPreferences(audioTrack.codec, capabilitiesGetter) } // PublisherTransportObserver.onRenegotiationNeeded() gets triggered automatically @@ -840,6 +861,9 @@ internal constructor( options = options, ) addTrackPublication(publication) + if (track is LocalAudioTrack) { + pendingSubscribedAudioCodecUpdates.remove(publication.sid)?.let(::handleSubscribedAudioCodecUpdate) + } LKLog.v { "add track publication $publication" } publishListener?.onPublishSuccess(publication) @@ -1041,6 +1065,10 @@ internal constructor( engine.stopTransceivers(listOfNotNull(track.transceiver) + track.simulcastTransceivers) track.transceiver = null track.clearSimulcastCodecs() + } else if (track is LocalAudioTrack) { + engine.stopTransceivers(listOfNotNull(track.transceiver) + track.simulcastTransceivers) + track.transceiver = null + track.clearSimulcastCodecs() } } if (stopOnUnpublish) { @@ -1262,6 +1290,95 @@ internal constructor( } } + internal fun handleSubscribedAudioCodecUpdate(update: LivekitRtc.SubscribedAudioCodecUpdate) { + if (!dynacast) return + val publication = trackPublications[update.trackSid] as? LocalTrackPublication + if (publication == null) { + pendingSubscribedAudioCodecUpdates[update.trackSid] = update + return + } + val track = publication.track as? LocalAudioTrack ?: return + val options = publication.options as? AudioTrackPublishOptions ?: return + + update.subscribedAudioCodecsList.forEach { subscribedCodec -> + val codec = subscribedCodec.codec.lowercase().removePrefix("audio/") + if (codec == track.codec) { + track.setPublishingCodecEnabled(codec, subscribedCodec.enabled) + } else if (codec == AUDIO_CODEC_OPUS && options.red) { + track.setPublishingCodecEnabled(codec, subscribedCodec.enabled) + if (subscribedCodec.enabled && track.beginPublishingSimulcastCodec(codec)) { + publishAdditionalAudioCodecForTrack(track, publication, codec, options) + } + } + } + } + + private fun publishAdditionalAudioCodecForTrack( + track: LocalAudioTrack, + publication: LocalTrackPublication, + codec: String, + options: AudioTrackPublishOptions, + ) { + val encoding = RtpParameters.Encoding(null, true, null).apply { + options.audioBitrate?.takeIf { it > 0 }?.let { maxBitrateBps = it } + } + val transceiverInit = RtpTransceiverInit( + RtpTransceiver.RtpTransceiverDirection.SEND_ONLY, + listOf(this.sid.value), + listOf(encoding), + ) + + scope.launch { + val transceiver = try { + engine.createSenderTransceiver(track.rtcTrack, transceiverInit)?.transceiver + } catch (error: Exception) { + error.rethrowIfCancellationSignal() + track.cancelPublishingSimulcastCodec(codec) + LKLog.w(error) { "couldn't create additional $codec audio transceiver" } + return@launch + } + if (transceiver == null) { + track.cancelPublishingSimulcastCodec(codec) + LKLog.w { "couldn't create additional $codec audio transceiver" } + return@launch + } + transceiver.setAudioCodecPreferences(codec, capabilitiesGetter) + track.addSimulcastTransceiver(codec, transceiver) + engine.e2EEManager?.addPublishedSender(transceiver.sender, publication, this@LocalParticipant) + + val request = AddTrackRequest.newBuilder().apply { + sid = publication.sid + muted = !track.enabled + source = publication.source.toProto() + addSimulcastCodecs( + SimulcastCodec.newBuilder() + .setCodec(codec) + .setCid(transceiver.sender.id()) + .build(), + ) + } + try { + coroutineScope { + val negotiateJob = launch { engine.negotiatePublisher() } + val publishJob = async { + engine.addTrack( + cid = transceiver.sender.id(), + name = options.name ?: track.name, + kind = track.kind.toProto(), + stream = options.stream, + builder = request, + ) + } + negotiateJob.join() + publishJob.await() + } + } catch (error: Exception) { + error.rethrowIfCancellationSignal() + LKLog.w(error) { "exception when publishing $codec for audio track ${track.sid}" } + } + } + } + private fun publishAdditionalCodecForTrack(track: LocalVideoTrack, codec: VideoCodec, options: VideoTrackPublishOptions) { val existingPublication = trackPublications[track.sid] ?: run { LKLog.w { "attempting to publish additional codec for non-published track?!" } @@ -1418,6 +1535,7 @@ internal constructor( * @suppress */ fun cleanup() { + pendingSubscribedAudioCodecUpdates.clear() for (pub in trackPublications.values) { val track = pub.track @@ -1768,6 +1886,8 @@ internal fun VideoTrackPublishOptions.hasBackupCodec(): Boolean { return backupCodec?.codec != null && videoCodec != backupCodec.codec } +private const val AUDIO_CODEC_OPUS = "opus" +private const val AUDIO_CODEC_RED = "red" private val backupCodecs = listOf(VideoCodec.VP8.codecName, VideoCodec.H264.codecName) private fun isBackupCodec(codecName: String) = backupCodecs.contains(codecName) diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalAudioTrack.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalAudioTrack.kt index 8c83c165a..9391365c0 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalAudioTrack.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalAudioTrack.kt @@ -81,9 +81,59 @@ constructor( private val delegateScope = CoroutineScope(dispatcher + SupervisorJob()) internal var transceiver: RtpTransceiver? = null + internal var codec: String = "opus" + private val simulcastTransceiverMap = mutableMapOf() + private val pendingSimulcastCodecs = mutableSetOf() + private val simulcastCodecEnabled = mutableMapOf() + internal val simulcastTransceivers: List + @Synchronized get() = simulcastTransceiverMap.values.toList() internal val sender: RtpSender? get() = transceiver?.sender + @Synchronized + internal fun beginPublishingSimulcastCodec(codec: String): Boolean { + val normalizedCodec = codec.lowercase().removePrefix("audio/") + if (simulcastTransceiverMap.containsKey(normalizedCodec)) return false + return pendingSimulcastCodecs.add(normalizedCodec) + } + + @Synchronized + internal fun addSimulcastTransceiver(codec: String, transceiver: RtpTransceiver) { + val normalizedCodec = codec.lowercase().removePrefix("audio/") + simulcastTransceiverMap[normalizedCodec] = transceiver + pendingSimulcastCodecs.remove(normalizedCodec) + setPublishingCodecEnabled(normalizedCodec, simulcastCodecEnabled[normalizedCodec] ?: true) + } + + @Synchronized + internal fun cancelPublishingSimulcastCodec(codec: String) { + pendingSimulcastCodecs.remove(codec.lowercase().removePrefix("audio/")) + } + + @Synchronized + internal fun setPublishingCodecEnabled(codec: String, enabled: Boolean) { + val normalizedCodec = codec.lowercase().removePrefix("audio/") + if (this.codec != normalizedCodec) { + simulcastCodecEnabled[normalizedCodec] = enabled + } + val target = if (this.codec == normalizedCodec) transceiver else simulcastTransceiverMap[normalizedCodec] + val sender = target?.sender ?: return + val parameters = sender.parameters ?: return + val encoding = parameters.encodings?.firstOrNull() ?: return + if (encoding.active == enabled) return + encoding.active = enabled + if (!sender.setParameters(parameters)) { + LKLog.w { "failed to ${if (enabled) "enable" else "disable"} $normalizedCodec audio sender" } + } + } + + @Synchronized + internal fun clearSimulcastCodecs() { + simulcastTransceiverMap.clear() + pendingSimulcastCodecs.clear() + simulcastCodecEnabled.clear() + } + private val trackSinks = mutableSetOf() /** diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/RtpTransceiverExt.kt b/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/RtpTransceiverExt.kt index e10b69005..4b9ae2985 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/RtpTransceiverExt.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/RtpTransceiverExt.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023-2024 LiveKit, Inc. + * Copyright 2023-2026 LiveKit, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -59,3 +59,23 @@ internal fun RtpTransceiver.sortVideoCodecPreferences(targetCodec: String, capab } setCodecPreferences(matched.plus(partialMatched).plus(unmatched)) } + +internal fun RtpTransceiver.setAudioCodecPreferences(targetCodec: String, capabilitiesGetter: CapabilitiesGetter) { + val capabilities = capabilitiesGetter(MediaStreamTrack.MediaType.MEDIA_TYPE_AUDIO) + val normalizedTarget = targetCodec.lowercase().removePrefix("audio/") + val selected = capabilities.codecs.filter { codec -> + val mimeType = codec.mimeType.lowercase() + when (normalizedTarget) { + "red" -> mimeType == "audio/red" || mimeType == "audio/opus" + "opus" -> mimeType != "audio/red" + else -> true + } + }.sortedBy { codec -> + when (codec.mimeType.lowercase()) { + "audio/$normalizedTarget" -> 0 + "audio/opus" -> 1 + else -> 2 + } + } + setCodecPreferences(selected) +} diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt index f547f899f..ad9e78cd1 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023-2025 LiveKit, Inc. + * Copyright 2023-2026 LiveKit, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,13 +25,16 @@ import org.mockito.kotlin.whenever import java.util.UUID object MockRtpSender { - fun create(id: String = "sender_id"): RtpSender { + fun create( + id: String = "sender_id", + encodings: MutableList = mutableListOf(), + ): RtpSender { var rtpParameters: RtpParameters = MockRtpParameters( transactionId = UUID.randomUUID().toString(), degradationPreference = null, rtcp = MockRtpParameters.MockRtcp("", false), headerExtensions = mutableListOf(), - encodings = mutableListOf(), + encodings = encodings, codecs = mutableListOf(), ) return Mockito.mock(RtpSender::class.java).apply { diff --git a/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt b/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt index c3c373424..0ce9b02b9 100644 --- a/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt +++ b/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023-2025 LiveKit, Inc. + * Copyright 2023-2026 LiveKit, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -41,7 +41,7 @@ object MockRtpTransceiver { when (direction) { RtpTransceiverDirection.SEND_RECV, RtpTransceiverDirection.SEND_ONLY -> { - val sender = MockRtpSender.create(id = id) + val sender = MockRtpSender.create(id = id, encodings = init.sendEncodings.toMutableList()) Mockito.`when`(mock.sender) .then { sender } } diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/SignalClientTest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/SignalClientTest.kt index a10dfe3fa..8005986b4 100644 --- a/livekit-android-test/src/test/java/io/livekit/android/room/SignalClientTest.kt +++ b/livekit-android-test/src/test/java/io/livekit/android/room/SignalClientTest.kt @@ -501,6 +501,32 @@ class SignalClientTest : BaseTest() { } } + @Test + fun subscribedAudioCodecUpdateNotifiesListener() = runTest { + val joinJob = async { client.join(EXAMPLE_URL, "") } + connectWebsocketAndJoin() + joinJob.await() + client.onReadyForResponses() + val update = LivekitRtc.SubscribedAudioCodecUpdate.newBuilder() + .setTrackSid("track_sid") + .addSubscribedAudioCodecs( + LivekitModels.SubscribedAudioCodec.newBuilder() + .setCodec("opus") + .setEnabled(true), + ) + .build() + + client.onMessage( + wsFactory.ws, + LivekitRtc.SignalResponse.newBuilder() + .setSubscribedAudioCodecUpdate(update) + .build() + .toOkioByteString(), + ) + + Mockito.verify(listener).onSubscribedAudioCodecUpdate(update) + } + // mock data companion object } diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt index 0498b8ed1..8628ed093 100644 --- a/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt +++ b/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt @@ -22,6 +22,7 @@ import android.content.Context import androidx.test.core.app.ApplicationProvider import com.google.protobuf.ByteString import io.livekit.android.ConnectOptions +import io.livekit.android.e2ee.E2EEManager import io.livekit.android.events.ParticipantEvent import io.livekit.android.events.RoomEvent import io.livekit.android.room.DefaultsManager @@ -318,6 +319,106 @@ class LocalParticipantMockE2ETest : MockE2ETest() { transceivers.forEach { Mockito.verify(it).stopInternal() } } + @Test + fun encryptedRedAudioAdvertisesOpusSimulcastCodec() = runTest { + room.dynacast = true + connect() + component.rtcEngine().e2EEManager = mock(E2EEManager::class.java) + wsFactory.ws.clearRequests() + + room.localParticipant.publishAudioTrack(createMockLocalAudioTrack()) + + val addTrack = sentAddTrackRequests().single() + assertEquals(listOf("red", "opus"), addTrack.simulcastCodecsList.map { it.codec }) + assertTrue(addTrack.simulcastCodecsList.first { it.codec == "red" }.cid.isNotEmpty()) + assertTrue(addTrack.simulcastCodecsList.first { it.codec == "opus" }.cid.isEmpty()) + } + + @Test + fun unencryptedRedAudioDoesNotAdvertiseAudioSimulcast() = runTest { + room.dynacast = true + connect() + wsFactory.ws.clearRequests() + + room.localParticipant.publishAudioTrack(createMockLocalAudioTrack()) + + assertTrue(sentAddTrackRequests().single().simulcastCodecsList.isEmpty()) + } + + @Test + fun subscribedOpusCreatesOneAdditionalAudioTransceiver() = runTest { + room.dynacast = true + connect() + val e2eeManager = mock(E2EEManager::class.java) + component.rtcEngine().e2EEManager = e2eeManager + val track = createMockLocalAudioTrack() + room.localParticipant.publishAudioTrack(track) + val trackSid = room.localParticipant.audioTrackPublications.first().first.sid + wsFactory.ws.clearRequests() + + receiveSubscribedAudioCodecUpdate(trackSid, "opus" to true) + receiveSubscribedAudioCodecUpdate(trackSid, "opus" to true) + advanceUntilIdle() + + assertEquals(2, getPublisherPeerConnection().transceivers.size) + val addTrack = sentAddTrackRequests().single() + assertEquals(trackSid, addTrack.sid) + assertEquals(listOf("opus"), addTrack.simulcastCodecsList.map { it.codec }) + assertTrue(addTrack.simulcastCodecsList.single().cid.isNotEmpty()) + val opusTransceiver = getPublisherPeerConnection().transceivers.last() + Mockito.verify(e2eeManager).addPublishedSender( + opusTransceiver.sender, + room.localParticipant.audioTrackPublications.first().first, + room.localParticipant, + ) + + receiveSubscribedAudioCodecUpdate(trackSid, "opus" to false) + advanceUntilIdle() + assertFalse(opusTransceiver.sender.parameters.encodings.single().active) + + receiveSubscribedAudioCodecUpdate(trackSid, "opus" to true) + advanceUntilIdle() + assertTrue(opusTransceiver.sender.parameters.encodings.single().active) + + room.localParticipant.unpublishTrack(track) + getPublisherPeerConnection().transceivers.forEach { Mockito.verify(it).stopInternal() } + } + + @Test + fun subscribedOpusUpdateBeforePublicationIsReplayed() = runTest { + room.dynacast = true + connect() + component.rtcEngine().e2EEManager = mock(E2EEManager::class.java) + wsFactory.ws.clearRequests() + + receiveSubscribedAudioCodecUpdate(TestData.LOCAL_AUDIO_TRACK.sid, "opus" to true) + room.localParticipant.publishAudioTrack(createMockLocalAudioTrack()) + advanceUntilIdle() + + assertEquals(2, getPublisherPeerConnection().transceivers.size) + assertEquals( + listOf("opus"), + sentAddTrackRequests().last().simulcastCodecsList.map { it.codec }, + ) + } + + @Test + fun subscribedOpusDisabledWhilePublishingRemainsInactive() = runTest { + room.dynacast = true + connect() + component.rtcEngine().e2EEManager = mock(E2EEManager::class.java) + val track = createMockLocalAudioTrack() + room.localParticipant.publishAudioTrack(track) + val trackSid = room.localParticipant.audioTrackPublications.first().first.sid + + receiveSubscribedAudioCodecUpdate(trackSid, "opus" to true) + receiveSubscribedAudioCodecUpdate(trackSid, "opus" to false) + advanceUntilIdle() + + val opusTransceiver = getPublisherPeerConnection().transceivers.last() + assertFalse(opusTransceiver.sender.parameters.encodings.single().active) + } + @Test fun republishAfterBackupCodecUnpublishCreatesNewBackupTransceiver() = runTest { room.videoTrackPublishDefaults = room.videoTrackPublishDefaults.copy( @@ -449,6 +550,32 @@ class LocalParticipantMockE2ETest : MockE2ETest() { ) } + private fun receiveSubscribedAudioCodecUpdate(trackSid: String, vararg codecs: Pair) { + wsFactory.receiveMessage( + LivekitRtc.SignalResponse.newBuilder() + .setSubscribedAudioCodecUpdate( + LivekitRtc.SubscribedAudioCodecUpdate.newBuilder() + .setTrackSid(trackSid) + .addAllSubscribedAudioCodecs( + codecs.map { (codec, enabled) -> + LivekitModels.SubscribedAudioCodec.newBuilder() + .setCodec(codec) + .setEnabled(enabled) + .build() + }, + ), + ) + .build(), + ) + } + + private fun sentAddTrackRequests(): List = + wsFactory.ws.sentRequests.mapNotNull { requestBytes -> + LivekitRtc.SignalRequest.parseFrom(requestBytes.toPBByteString()) + .takeIf { it.hasAddTrack() } + ?.addTrack + } + private fun createLocalTrack( width: Int = 1280, height: Int = 720,