diff --git a/common/src/main/java/com/pedro/common/base/BaseSender.kt b/common/src/main/java/com/pedro/common/base/BaseSender.kt index 1c4cb8be6..120145c7b 100644 --- a/common/src/main/java/com/pedro/common/base/BaseSender.kt +++ b/common/src/main/java/com/pedro/common/base/BaseSender.kt @@ -131,8 +131,18 @@ abstract class BaseSender( resetBytesSend() val stopped = withTimeoutOrNull(1000.milliseconds) { job?.cancelAndJoin() } != null if (!stopped) { - unlockNeeded() - withTimeoutOrNull(1000.milliseconds) { job?.cancelAndJoin() } + Log.w(TAG, "sender did not stop in time, probably blocked in a socket write, unlocking it") + try { + unlockNeeded() + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + Log.e(TAG, "error unlocking sender", e) + } + val stoppedAfterUnlock = withTimeoutOrNull(1000.milliseconds) { job?.cancelAndJoin() } != null + if (!stoppedAfterUnlock) { + Log.w(TAG, "sender job did not finish after unlock") + } } job = null queue.clear { bufferPool.release(it.data) } diff --git a/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt b/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt index f9ee0c981..149cc5c9b 100644 --- a/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt +++ b/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt @@ -3,13 +3,17 @@ package com.pedro.common.base import com.pedro.common.ConnectChecker import com.pedro.common.frame.MediaFrame import com.pedro.common.removeInfo +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.delay import kotlinx.coroutines.runBlocking import org.junit.Assert.assertArrayEquals import org.junit.Assert.assertEquals import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue import org.junit.Test import org.mockito.Mockito import java.nio.ByteBuffer +import java.util.concurrent.atomic.AtomicBoolean /** * The sender copies each frame into a pooled array bigger than the frame itself. These tests lock @@ -17,7 +21,7 @@ import java.nio.ByteBuffer */ class BaseSenderTest { - private class FakeSender: BaseSender(Mockito.mock(ConnectChecker::class.java), "FakeSender") { + private open class FakeSender: BaseSender(Mockito.mock(ConnectChecker::class.java), "FakeSender") { override fun setVideoInfo(sps: ByteBuffer, pps: ByteBuffer?, vps: ByteBuffer?) {} override fun setAudioInfo(sampleRate: Int, isStereo: Boolean) {} override suspend fun onRun() {} @@ -127,6 +131,61 @@ class BaseSenderTest { } } + @Test + fun `GIVEN sender blocked WHEN stop THEN unlockNeeded is invoked and stop returns`() = runBlocking { + val unlockCalled = AtomicBoolean(false) + val canExit = AtomicBoolean(false) + val sender = object : FakeSender() { + override suspend fun onRun() { + while (!canExit.get()) { + try { + delay(Long.MAX_VALUE) + } catch (e: CancellationException) { + if (!canExit.get()) { + Thread.sleep(10) + continue + } + throw e + } + } + } + } + sender.start() + sender.stop(unlockNeeded = { + unlockCalled.set(true) + canExit.set(true) + }) + assertTrue(unlockCalled.get()) + } + + @Test + fun `GIVEN unlockNeeded throws WHEN stop THEN exception does not propagate`() = runBlocking { + val canExit = AtomicBoolean(false) + val sender = object : FakeSender() { + override suspend fun onRun() { + while (!canExit.get()) { + try { + delay(Long.MAX_VALUE) + } catch (e: CancellationException) { + if (!canExit.get()) { + Thread.sleep(10) + continue + } + throw e + } + } + } + } + sender.start() + try { + sender.stop(unlockNeeded = { + throw RuntimeException("unlock failed") + }) + } finally { + canExit.set(true) + } + } + @Test fun `GIVEN an info bigger than the frame WHEN removeInfo THEN do not expose stale bytes`() = runBlocking { val sender = FakeSender()