log and guard the sender unlock in BaseSender.stop

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) }

common/src/test/java/com/pedro/common/base/BaseSenderTest.kt

@@ -3,21 +3,25 @@ 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
  * the shape of the resulting buffer so the packetizers keep behaving like they did with a copy.
  */
 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()