diff --git a/.changeset/set-track-enabled-cancellation-cleanup.md b/.changeset/set-track-enabled-cancellation-cleanup.md new file mode 100644 index 00000000..e0ef688a --- /dev/null +++ b/.changeset/set-track-enabled-cancellation-cleanup.md @@ -0,0 +1,5 @@ +--- +"client-sdk-android": patch +--- + +Fixed setTrackEnabled leaking track resources when cancelled before the track is published, including the sender negotiated for a failed publish, and LocalScreencastVideoTrack leaking its SurfaceTextureHelper on dispose. 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 19d3464f..b67a4b5e 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 @@ -133,6 +133,7 @@ internal constructor( } when (newVal) { ConnectionState.CONNECTED -> { + signalSessionState = SignalSessionState(ended = false) if (oldVal == ConnectionState.DISCONNECTED || oldVal == ConnectionState.CONNECTING) { LKLog.d { "primary ICE connected" } listener?.onEngineConnected() @@ -158,8 +159,17 @@ internal constructor( @Volatile internal var reconnectType: ReconnectType = ReconnectType.DEFAULT + + @Volatile private var reconnectingJob: Job? = null + /** + * Replaced at each signal session boundary. Sender handles retain the state that owned their + * transceiver, so cleanup cannot mutate a publisher after that session ends. + */ + @Volatile + private var signalSessionState = SignalSessionState(ended = true) + @Volatile private var fullReconnectOnNext = false @@ -424,10 +434,13 @@ internal constructor( internal suspend fun createSenderTransceiver( rtcTrack: MediaStreamTrack, transInit: RtpTransceiverInit, - ): RtpTransceiver? { - return publisher?.withPeerConnection { + ): SenderTransceiverHandle? { + val sessionState = signalSessionState + val publisher = publisher ?: return null + val transceiver = publisher.withPeerConnection { addTransceiver(rtcTrack, transInit) - } + } ?: return null + return SenderTransceiverHandle(publisher, transceiver, sessionState) } fun updateSubscriptionPermissions( @@ -505,6 +518,9 @@ internal constructor( } private fun abortPendingPublishTracks() { + // Ordered ahead of the resumes: each one unwinds a publish into cleanup that reads this, + // and on some paths the reconnect is only triggered later, by the socket closing. + signalSessionState = SignalSessionState(ended = true) synchronized(pendingTrackResolvers) { pendingTrackResolvers.values.forEach { it.resumeWithException(TrackException.PublishException("pending track aborted")) @@ -535,6 +551,7 @@ internal constructor( } val forceFullReconnect = fullReconnectOnNext fullReconnectOnNext = false + signalSessionState = SignalSessionState(ended = true) val job = coroutineScope.launch { var hasResumedOnce = false var hasReconnectedOnce = false @@ -1515,6 +1532,45 @@ internal constructor( } } + /** + * Detaches a sender transceiver whose publish attempt produced no publication. + * + * A publish fails most often because the signal connection died, and that same failure + * starts a reconnect which renegotiates this peer connection. Mutating its senders and + * transceivers alongside that negotiation races it, so the rollback is skipped unless the + * publisher is connected and idle. The sender retains its signal session state so an aborted + * publish cannot mutate the same publisher after a soft reconnect. + * + * @param stopTransceiver releases the transceiver and its SDP m-section for reuse. Reserved + * for video, matching the teardown an unpublish performs. + * @return whether the sender was detached, so the track can drop its reference to it. + */ + internal fun rollbackSenderTransceiver( + senderTransceiver: SenderTransceiverHandle, + stopTransceiver: Boolean, + ): Boolean { + val publisher = senderTransceiver.publisher + return runBlocking { + publisher.withPeerConnection { + val sessionState = signalSessionState + val publisherIsIdle = this@RTCEngine.publisher === publisher && + connectionState == ConnectionState.CONNECTED && + reconnectingJob?.isActive != true && + senderTransceiver.signalSessionState === sessionState && + !sessionState.ended + if (!publisherIsIdle) { + return@withPeerConnection false + } + val transceiver = senderTransceiver.transceiver + removeTrack(transceiver.sender) + if (stopTransceiver && !transceiver.isStopped) { + transceiver.stopInternal() + } + true + } ?: false + } + } + @VisibleForTesting fun getPublisherPeerConnection() = publisher!!.peerConnection @@ -1524,6 +1580,22 @@ internal constructor( subscriber!!.peerConnection } +/** + * Identity of one signal session, replaced at every session boundary. + * + * [ended] distinguishes a session that is finishing from one that is serving, for work that + * starts after the boundary and would otherwise match the current identity. + */ +internal class SignalSessionState( + val ended: Boolean, +) + +internal class SenderTransceiverHandle( + internal val publisher: PeerConnectionTransport, + internal val transceiver: RtpTransceiver, + internal val signalSessionState: SignalSessionState, +) + /** * @suppress */ 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 a1fe8861..882535be 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 @@ -35,6 +35,7 @@ import io.livekit.android.room.ConnectionState import io.livekit.android.room.DefaultsManager import io.livekit.android.room.RTCEngine import io.livekit.android.room.Room +import io.livekit.android.room.SenderTransceiverHandle import io.livekit.android.room.TrackBitrateInfo import io.livekit.android.room.datastream.outgoing.OutgoingDataStreamManager import io.livekit.android.room.isSVCCodec @@ -64,11 +65,14 @@ import io.livekit.android.util.rethrowIfCancellationSignal import io.livekit.android.webrtc.sortVideoCodecPreferences import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext import livekit.LivekitModels import livekit.LivekitModels.AudioTrackFeature import livekit.LivekitModels.Codec @@ -86,7 +90,9 @@ import livekit.org.webrtc.SurfaceTextureHelper import livekit.org.webrtc.VideoCapturer import livekit.org.webrtc.VideoProcessor import java.util.Collections +import java.util.concurrent.atomic.AtomicBoolean import javax.inject.Named +import kotlin.coroutines.coroutineContext import kotlin.math.max import kotlin.math.min import kotlin.time.Duration @@ -316,6 +322,9 @@ internal constructor( * * For screenshare audio, a [ScreenAudioCapturer] can be used. * + * If a screen capture track is created but enabling fails or is cancelled, the capture is + * stopped and released, and [ScreenCaptureParams.onStop] is invoked. + * * @param screenCaptureParams When enabling the screenshare, this must be provided with * [ScreenCaptureParams.mediaProjectionPermissionResultData] containing resultData returned from launching * [MediaProjectionManager.createScreenCaptureIntent()](https://developer.android.com/reference/android/media/projection/MediaProjectionManager#createScreenCaptureIntent()). @@ -355,25 +364,29 @@ internal constructor( when (source) { Track.Source.CAMERA -> { val track = getOrCreateDefaultVideoTrack() - track.start() - track.startCapture() - if (!publishVideoTrack(track)) { - track.stopCapture() - track.stop() - } else { - success = true + try { + track.start() + track.startCapture() + success = publishVideoTrack(track) + } finally { + if (!success) { + track.stopCapture() + track.stop() + } } } Track.Source.MICROPHONE -> { val track = getOrCreateDefaultAudioTrack() - track.prewarm() - track.start() - if (!publishAudioTrack(track)) { - track.stop() - track.stopPrewarm() - } else { - success = true + try { + track.prewarm() + track.start() + success = publishAudioTrack(track) + } finally { + if (!success) { + track.stop() + track.stopPrewarm() + } } } @@ -381,22 +394,32 @@ internal constructor( if (screenCaptureParams == null) { throw IllegalArgumentException("Media Projection params is required to create a screen share track.") } + // MediaProjection termination and setup failure may race, but onStop + // is a single terminal signal for each enable attempt. + val stopNotified = AtomicBoolean(false) + fun notifyStopped() { + if (stopNotified.compareAndSet(false, true)) { + screenCaptureParams.onStop?.invoke() + } + } + val track = createScreencastTrack(mediaProjectionPermissionResultData = screenCaptureParams.mediaProjectionPermissionResultData) { unpublishTrack(it) - screenCaptureParams.onStop?.invoke() + notifyStopped() } - track.startForegroundService(screenCaptureParams.notificationId, screenCaptureParams.notification) - track.startCapture() - if (!publishVideoTrack(track, options = VideoTrackPublishOptions(null, screenShareTrackPublishDefaults))) { - screenCaptureParams.onStop?.invoke() - track.apply { - stopCapture() - stop() - dispose() + try { + track.startForegroundService(screenCaptureParams.notificationId, screenCaptureParams.notification) + track.startCapture() + success = publishVideoTrack(track, options = VideoTrackPublishOptions(null, screenShareTrackPublishDefaults)) + } finally { + if (!success) { + // stop() releases the screen capture service binding as well. + track.stopCapture() + track.stop() + track.dispose() + notifyStopped() } - } else { - success = true } } @@ -672,6 +695,8 @@ internal constructor( } // For fast publish, we can negotiate PC and request add track at the same time + var senderTransceiver: SenderTransceiverHandle? = null + var previousTrackTransceiver: RtpTransceiver? = null suspend fun negotiate() { if (this.engine.publisher == null) { throw IllegalStateException("publisher is not configured yet!") @@ -682,15 +707,14 @@ internal constructor( listOf(this.sid.value), encodings, ) - val transceiver = engine.createSenderTransceiver(track.rtcTrack, transInit) - - when (track) { - is LocalVideoTrack -> track.transceiver = transceiver - is LocalAudioTrack -> track.transceiver = transceiver - else -> { - throw IllegalArgumentException("Trying to publish a non local track of type ${track.javaClass}") + // Ownership must be recorded before cancellation can discard the RTC-thread result. + val createdTransceiver = withContext(NonCancellable) { + engine.createSenderTransceiver(track.rtcTrack, transInit).also { + senderTransceiver = it } } + coroutineContext.ensureActive() + val transceiver = createdTransceiver?.transceiver if (transceiver == null) { val exception = TrackException.PublishException("null sender returned from peer connection") @@ -698,6 +722,22 @@ internal constructor( throw exception } + when (track) { + is LocalVideoTrack -> { + previousTrackTransceiver = track.transceiver + track.transceiver = transceiver + } + + is LocalAudioTrack -> { + previousTrackTransceiver = track.transceiver + track.transceiver = transceiver + } + + else -> { + throw IllegalArgumentException("Trying to publish a non local track of type ${track.javaClass}") + } + } + track.statsGetter = engine.createStatsGetter(transceiver.sender) val finalOptions = options @@ -744,62 +784,90 @@ internal constructor( } } - val trackInfo: TrackInfo? - if (enabledPublishVideoCodecs.isNotEmpty()) { - // Can simultaneous publish and negotiate. - // codec is pre-verified in publishVideoTrack - trackInfo = coroutineScope { - val negotiateJob = launch { negotiate() } - val publishJob = async { requestAddTrack() } + var publication: LocalTrackPublication? = null + try { + val trackInfo: TrackInfo? + if (enabledPublishVideoCodecs.isNotEmpty()) { + // Can simultaneous publish and negotiate. + // codec is pre-verified in publishVideoTrack + trackInfo = coroutineScope { + val negotiateJob = launch { negotiate() } + val publishJob = async { requestAddTrack() } + + negotiateJob.join() + return@coroutineScope publishJob.await() + } + } else { + // legacy path. + trackInfo = requestAddTrack() + if (trackInfo != null) { + if (options is VideoTrackPublishOptions) { + // server might not support the codec the client has requested, in that case, fallback + // to a supported codec + val primaryCodecMime = trackInfo.codecsList.firstOrNull()?.mimeType + + if (primaryCodecMime != null) { + val updatedCodec = primaryCodecMime.mimeTypeToVideoCodec() + if (updatedCodec != null && updatedCodec != options.videoCodec) { + LKLog.d { "falling back to server selected codec: $updatedCodec" } + options = options.copy(videoCodec = updatedCodec) + + // recompute encodings since bitrates/etc could have changed + val videoTrack = track as LocalVideoTrack + + encodings = computeVideoEncodings(videoTrack.options.isScreencast, videoTrack.dimensions, options) + encodings // encodings is used in negotiate, this suppresses unused lint + } + } + } - negotiateJob.join() - return@coroutineScope publishJob.await() + negotiate() + } } - } else { - // legacy path. - trackInfo = requestAddTrack() + if (trackInfo != null) { - if (options is VideoTrackPublishOptions) { - // server might not support the codec the client has requested, in that case, fallback - // to a supported codec - val primaryCodecMime = trackInfo.codecsList.firstOrNull()?.mimeType - - if (primaryCodecMime != null) { - val updatedCodec = primaryCodecMime.mimeTypeToVideoCodec() - if (updatedCodec != null && updatedCodec != options.videoCodec) { - LKLog.d { "falling back to server selected codec: $updatedCodec" } - options = options.copy(videoCodec = updatedCodec) - - // recompute encodings since bitrates/etc could have changed - val videoTrack = track as LocalVideoTrack - - encodings = computeVideoEncodings(videoTrack.options.isScreencast, videoTrack.dimensions, options) - encodings // encodings is used in negotiate, this suppresses unused lint + publication = LocalTrackPublication( + info = trackInfo, + track = track, + participant = this, + options = options, + ) + addTrackPublication(publication) + LKLog.v { "add track publication $publication" } + + publishListener?.onPublishSuccess(publication) + internalListener?.onTrackPublished(publication, this) + eventBus.postEvent(ParticipantEvent.LocalTrackPublished(this, publication), scope) + } + } finally { + if (publication == null) { + // Negotiation can win the race against a failed or cancelled add track request. + // Without a publication there is no unpublish to stop the transceiver, so it + // would hold its sender and SDP m-section until the connection closes. + senderTransceiver?.let { createdTransceiver -> + val canRestorePreviousTransceiver = engine.rollbackSenderTransceiver( + senderTransceiver = createdTransceiver, + stopTransceiver = track is LocalVideoTrack, + ) + val transceiver = createdTransceiver.transceiver + when (track) { + is LocalVideoTrack -> { + if (track.transceiver === transceiver) { + track.transceiver = previousTrackTransceiver.takeIf { canRestorePreviousTransceiver } + } + } + + is LocalAudioTrack -> { + if (track.transceiver === transceiver) { + track.transceiver = previousTrackTransceiver.takeIf { canRestorePreviousTransceiver } + } } } } - - negotiate() } } - return if (trackInfo != null) { - val publication = LocalTrackPublication( - info = trackInfo, - track = track, - participant = this, - options = options, - ) - addTrackPublication(publication) - LKLog.v { "add track publication $publication" } - - publishListener?.onPublishSuccess(publication) - internalListener?.onTrackPublished(publication, this) - eventBus.postEvent(ParticipantEvent.LocalTrackPublished(this, publication), scope) - publication - } else { - null - } + return publication } private fun computeVideoEncodings( @@ -1207,7 +1275,7 @@ internal constructor( ) scope.launch { - val transceiver = engine.createSenderTransceiver(track.rtcTrack, transceiverInit) + val transceiver = engine.createSenderTransceiver(track.rtcTrack, transceiverInit)?.transceiver if (transceiver == null) { LKLog.w { "couldn't create new transceiver! $codec" } return@launch diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt index b5dcaa57..563c7dc5 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt @@ -28,6 +28,7 @@ import android.view.WindowManager import dagger.assisted.Assisted import dagger.assisted.AssistedFactory import dagger.assisted.AssistedInject +import io.livekit.android.memory.SurfaceTextureHelperCloser import io.livekit.android.room.DefaultsManager import io.livekit.android.room.participant.LocalParticipant import io.livekit.android.room.track.screencapture.ScreenCaptureConnection @@ -190,7 +191,15 @@ constructor( } override fun stop() { - super.stop() + // The projection's onStop arrives on its own callback thread and stops this track, so + // it can lose a race against dispose(), after which the capturer throws when stopped. + if (!isDisposed) { + try { + super.stop() + } catch (e: Exception) { + LKLog.w(e) { "Exception when stopping the screen share track." } + } + } serviceConnection.stop() orientationEventListener.disable() } @@ -243,21 +252,28 @@ constructor( addOnStopCallback(onStop) } val capturer = createScreenCapturer(mediaProjectionPermissionResultData, callback) + val surfaceTextureHelper = SurfaceTextureHelper.create("ScreenVideoCaptureThread", rootEglBase.eglBaseContext) capturer.initialize( - SurfaceTextureHelper.create("ScreenVideoCaptureThread", rootEglBase.eglBaseContext), + surfaceTextureHelper, context, source.capturerObserver, ) - val track = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source) + val rtcTrack = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source) - return screencastVideoTrackFactory.create( + val track = screencastVideoTrackFactory.create( capturer = capturer, source = source, options = options, name = name, - rtcTrack = track, + rtcTrack = rtcTrack, mediaProjectionCallback = callback, ) + // The track owns the SurfaceTextureHelper; dispose() releases its capture thread. + track.closeableManager.registerResource( + rtcTrack, + SurfaceTextureHelperCloser(surfaceTextureHelper), + ) + return track } private fun createScreenCapturer( diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt index 925e46e3..358ed6fb 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt @@ -119,7 +119,7 @@ constructor( internal val simulcastTransceivers: List get() = simulcastCodecs.values.mapNotNull { it.transceiver } - private val closeableManager = CloseableManager() + internal val closeableManager = CloseableManager() /** * Starts the [capturer] with the capture params contained in [options]. diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt index e2992424..6a51cfeb 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.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. @@ -170,7 +170,7 @@ class MockPeerConnection( } override fun removeTrack(sender: RtpSender?): Boolean { - return super.removeTrack(sender) + return true } override fun addTransceiver(track: MediaStreamTrack): RtpTransceiver { diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt new file mode 100644 index 00000000..a0410fdd --- /dev/null +++ b/livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt @@ -0,0 +1,356 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.livekit.android.room.participant + +import android.Manifest +import android.app.Application +import android.app.Notification +import android.content.ComponentName +import android.content.Intent +import io.livekit.android.room.ConnectionState +import io.livekit.android.room.track.LocalScreencastVideoTrack +import io.livekit.android.room.track.screencapture.ScreenCaptureParams +import io.livekit.android.room.track.screencapture.ScreenCaptureService +import io.livekit.android.test.MockE2ETest +import io.livekit.android.test.mock.camera.MockCameraProvider +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runCurrent +import livekit.LivekitRtc +import livekit.org.webrtc.ScreenCapturerAndroid +import livekit.org.webrtc.SurfaceTextureHelper +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.MockedConstruction +import org.mockito.Mockito.atLeastOnce +import org.mockito.Mockito.doThrow +import org.mockito.Mockito.mock +import org.mockito.Mockito.mockConstruction +import org.mockito.Mockito.mockStatic +import org.mockito.Mockito.never +import org.mockito.Mockito.verify +import org.mockito.kotlin.any +import org.mockito.kotlin.anyOrNull +import org.robolectric.RobolectricTestRunner +import org.robolectric.Shadows.shadowOf + +/** + * Enabling a track suspends while publishing, so a cancellation can land after the track and its + * resources were created. These tests pin that setTrackEnabled always tears down a track it could + * not publish, at every suspension point. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class SetTrackEnabledCleanupMockE2ETest : MockE2ETest() { + + private lateinit var screenCapturers: MockedConstruction + private var projectionCallback: LocalScreencastVideoTrack.MediaProjectionCallback? = null + + @Before + fun mockScreenCapturerConstruction() { + screenCapturers = mockConstruction(ScreenCapturerAndroid::class.java) { _, creation -> + projectionCallback = creation.arguments()[1] as LocalScreencastVideoTrack.MediaProjectionCallback + } + } + + @After + fun releaseScreenCapturerConstruction() { + screenCapturers.close() + } + + private val resumeLeave = with(LivekitRtc.SignalResponse.newBuilder()) { + leave = with(LivekitRtc.LeaveRequest.newBuilder()) { + action = LivekitRtc.LeaveRequest.Action.RESUME + build() + } + build() + } + + private fun screenCaptureParams(onStop: (() -> Unit)? = null) = ScreenCaptureParams( + mediaProjectionPermissionResultData = Intent(), + // An explicit notification, so the service does not reach for a notification manager. + notificationId = 42, + notification = Notification(), + onStop = onStop, + ) + + private fun connectScreenCaptureService() { + shadowOf(context as Application).boundServiceConnections.single().onServiceConnected( + ComponentName(context, ScreenCaptureService::class.java), + ScreenCaptureService().onBind(null), + ) + } + + @Test + fun cancelDuringScreenShareServiceBindCleansUpTrack() = runTest { + connect() + + var onStopCalled = false + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCalled = true }) + } + // Suspended in startForegroundService, waiting for the service to connect. + runCurrent() + assertEquals(1, shadowOf(context as Application).boundServiceConnections.size) + + job.cancelAndJoin() + + val capturer = screenCapturers.constructed().single() + verify(capturer, atLeastOnce()).stopCapture() + verify(capturer).dispose() + assertTrue(onStopCalled) + assertEquals(0, shadowOf(context as Application).boundServiceConnections.size) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + } + + @Test + fun cancelDuringScreenSharePublishCleansUpTrackAndUnbindsService() = runTest { + connect() + // Never answer the add track request, so publishing suspends until cancelled. + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + var onStopCalled = false + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCalled = true }) + } + runCurrent() + connectScreenCaptureService() + // Suspended in publishVideoTrack, with capture running and the service bound. + runCurrent() + val capturer = screenCapturers.constructed().single() + verify(capturer, never()).stopCapture() + + job.cancelAndJoin() + + verify(capturer, atLeastOnce()).stopCapture() + verify(capturer).dispose() + assertTrue(onStopCalled) + assertEquals(0, shadowOf(context as Application).boundServiceConnections.size) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + + // Negotiation ran concurrently with the add track request, so the transceiver it + // created must be rolled back; nothing else stops it once the track is abandoned. + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single()).stopInternal() + } + + @Test + fun projectionStopDuringScreenSharePublishDeliversOnStopOnce() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + var onStopCount = 0 + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCount++ }) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + // The user ends the projection from the system UI while publishing is still in flight. + projectionCallback!!.onStop() + assertEquals(1, onStopCount) + + job.cancelAndJoin() + + assertEquals(1, onStopCount) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + } + + @Test + fun projectionStopArrivingAfterCancelledEnableDoesNotThrow() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + job.cancelAndJoin() + + // A projection stop that was already dispatched can arrive after cleanup disposed the + // track, and the capturer throws once it is disposed. The callback must not propagate. + val capturer = screenCapturers.constructed().single() + doThrow(RuntimeException("capturer is disposed.")).`when`(capturer).stopCapture() + projectionCallback!!.onStop() + } + + @Test + fun successfulScreenShareKeepsTrackAndServiceBinding() = runTest { + connect() + + var onStopCalled = false + val result = async(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCalled = true }) + } + runCurrent() + connectScreenCaptureService() + advanceUntilIdle() + + assertTrue(result.await()) + val capturer = screenCapturers.constructed().single() + verify(capturer, never()).stopCapture() + verify(capturer, never()).dispose() + assertFalse(onStopCalled) + assertEquals(1, shadowOf(context as Application).boundServiceConnections.size) + assertEquals(1, room.localParticipant.trackPublications.size) + } + + @Test + fun cancelDuringMicrophonePublishStopsTrackAndKeepsTransceiver() = runTest { + connect() + shadowOf(context as Application).grantPermissions(Manifest.permission.RECORD_AUDIO) + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + + job.cancelAndJoin() + + assertFalse(room.localParticipant.getOrCreateDefaultAudioTrack().enabled) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + + // Detaching the sender is the whole rollback for audio: a single m-section is a cheap + // thing to leave behind, and stopping it renegotiates the publisher for no gain. + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun cancelDuringScreenSharePublishWhileReconnectingKeepsTransceiver() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + // Losing the socket with the add track request outstanding is what fails the publish, + // and it starts the reconnect that renegotiates this publisher. + wsFactory.ws.cancel() + runCurrent() + + job.cancelAndJoin() + + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun resumeLeaveDuringScreenSharePublishKeepsTransceiver() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + // A resume leave fails the pending publish and leaves the reconnect to the socket close, + // so the publisher still reads as connected while the failure unwinds. + simulateMessageFromServer(resumeLeave) + runCurrent() + job.join() + + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun resumeLeaveKeepsTransceiverWhenCleanupTrailsTheNextSession() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + simulateMessageFromServer(resumeLeave) + // The failed publish stays queued until the next session is serving. A soft reconnect + // keeps the publisher, so the transceiver this attempt created is the resumed session's + // to negotiate, not this cleanup's to remove. + component.rtcEngine().connectionState = ConnectionState.RESUMING + component.rtcEngine().connectionState = ConnectionState.CONNECTED + runCurrent() + job.join() + + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun cancelDuringCameraPublishStopsTrack() = runTest { + connect() + MockCameraProvider.register() + shadowOf(context as Application).grantPermissions(Manifest.permission.CAMERA) + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setCameraEnabled(true) + } + runCurrent() + + job.cancelAndJoin() + + assertFalse(room.localParticipant.getOrCreateDefaultVideoTrack().enabled) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + } + + @Test + fun disposeReleasesScreencastSurfaceTextureHelper() { + val helper = mock(SurfaceTextureHelper::class.java) + mockStatic(SurfaceTextureHelper::class.java).use { helperFactory -> + helperFactory.`when` { + SurfaceTextureHelper.create(any(), anyOrNull()) + }.thenReturn(helper) + + val track = room.localParticipant.createScreencastTrack( + mediaProjectionPermissionResultData = Intent(), + ) {} + track.dispose() + } + + verify(helper).stopListening() + verify(helper).dispose() + } +}