rtmp/rtsp: unblock a sender stuck in a socket write on disconnect

rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt

@@ -59,6 +59,9 @@ import java.nio.ByteBuffer
 import java.util.concurrent.atomic.AtomicLong
 import javax.net.ssl.TrustManager
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+private val SENDER_STOP_TIMEOUT = 1.seconds
 
 /**
  * Created by pedro on 8/04/21.
@@ -566,7 +569,16 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
   }
 
   private suspend fun disconnect(clear: Boolean) {
-    if (isStreaming) rtmpSender.stop(clear)
+    if (isStreaming) {
+      //the sender can be blocked in a socket write (TCP backpressure) that ignores cancellation,
+      //only closing the socket unblocks it. Try a cooperative stop first to keep the graceful close.
+      val stopped = withTimeoutOrNull(SENDER_STOP_TIMEOUT) { rtmpSender.stop(clear) } != null
+      if (!stopped) {
+        Log.w(TAG, "sender blocked in socket write, closing socket to unblock it")
+        runCatching { socket?.close() }
+        rtmpSender.stop(clear)
+      }
+    }
     runCatching {
       withTimeoutOrNull(100.milliseconds) {
         socket?.let { commandsManager.sendClose(it) }

rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt

@@ -46,6 +46,9 @@ import java.net.URISyntaxException
 import java.nio.ByteBuffer
 import javax.net.ssl.TrustManager
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+private val SENDER_STOP_TIMEOUT = 1.seconds
 
 /**
  * Created by pedro on 10/02/17.
@@ -426,7 +429,16 @@ class RtspClient(private val connectChecker: ConnectChecker) {
   }
 
   private suspend fun disconnect(clear: Boolean) {
-    if (isStreaming) rtspSender.stop()
+    if (isStreaming) {
+      //the sender can be blocked in a socket write (TCP backpressure) that ignores cancellation,
+      //only closing the socket unblocks it. Try a cooperative stop first to keep the graceful close.
+      val stopped = withTimeoutOrNull(SENDER_STOP_TIMEOUT) { rtspSender.stop() } != null
+      if (!stopped) {
+        Log.w(TAG, "sender blocked in socket write, closing socket to unblock it")
+        runCatching { socket?.close() }
+        rtspSender.stop()
+      }
+    }
     val error = runCatching {
       withTimeoutOrNull(100.milliseconds) {
         socket?.write(commandsManager.createTeardown())