fix propagate valid error message

common/src/main/java/com/pedro/common/Extensions.kt

@@ -135,4 +135,8 @@ fun String.getIndexes(char: Char): Array<Int> {
   val indexes = mutableListOf<Int>()
   forEachIndexed { index, c -> if (c == char) indexes.add(index) }
   return indexes.toTypedArray()
+}
+
+fun Throwable.validMessage(): String {
+  return (message ?: "").ifEmpty { javaClass.simpleName }
 }

common/src/main/java/com/pedro/common/socket/TcpStreamSocket.kt

@@ -32,7 +32,7 @@ import io.ktor.utils.io.readUTF8Line
 import io.ktor.utils.io.writeByte
 import io.ktor.utils.io.writeFully
 import io.ktor.utils.io.writeStringUtf8
-import java.io.IOException
+import java.net.ConnectException
 import java.security.SecureRandom
 import javax.net.ssl.TrustManager
 import kotlin.coroutines.coroutineContext
@@ -112,7 +112,8 @@ class TcpStreamSocket(
   }
 
   suspend fun read(): Int {
-    return input?.readByte()?.toInt() ?: throw IOException("read with socket closed")
+    val input = input ?: throw ConnectException("Read with socket closed, broken pipe")
+    return input.readByte().toInt()
   }
 
   suspend fun readUInt16(): Int {
@@ -132,11 +133,12 @@ class TcpStreamSocket(
   }
 
   suspend fun readUntil(b: ByteArray) {
-    return input?.readFully(b) ?: throw IOException("read with socket closed")
+    val input = input ?: throw ConnectException("Read with socket closed, broken pipe")
+    return input.readFully(b)
   }
 
   suspend fun readLine(): String? {
-    val input = input ?: throw IOException("read with socket closed")
+    val input = input ?: throw ConnectException("Read with socket closed, broken pipe")
     return input.readUTF8Line()
   }
 

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

@@ -20,10 +20,12 @@ import android.media.MediaCodec
 import android.util.Log
 import com.pedro.common.AudioCodec
 import com.pedro.common.ConnectChecker
+import com.pedro.common.ConnectionFailed
 import com.pedro.common.TimeUtils
 import com.pedro.common.UrlParser
 import com.pedro.common.VideoCodec
 import com.pedro.common.onMainThread
+import com.pedro.common.validMessage
 import com.pedro.rtmp.amf.AmfVersion
 import com.pedro.rtmp.rtmp.message.*
 import com.pedro.rtmp.rtmp.message.command.Command
@@ -269,7 +271,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
         if (error != null) {
           Log.e(TAG, "connection error", error)
           onMainThread {
-            connectChecker.onConnectionFailed("Error configure stream, ${error.message}")
+            connectChecker.onConnectionFailed("Error configure stream, ${error.validMessage()}")
           }
           return@launch
         }
@@ -281,6 +283,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
     while (scope.isActive && isStreaming) {
       val error = runCatching {
         if (isAlive()) {
+          delay(2000)
           //ignore packet after connect if tunneled to avoid spam idle
           if (!tunneled) handleMessages()
         } else {
@@ -290,7 +293,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
           scope.cancel()
         }
       }.exceptionOrNull()
-      if (error != null && error !is SocketTimeoutException) {
+      if (ConnectionFailed.parse(error?.validMessage() ?: "") != ConnectionFailed.TIMEOUT) {
         scope.cancel()
       }
     }
@@ -491,7 +494,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
     }
   }
 
-  suspend fun closeConnection() {
+  private suspend fun closeConnection() {
     socket?.close()
     commandsManager.reset()
   }

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

@@ -24,6 +24,7 @@ import com.pedro.common.ConnectChecker
 import com.pedro.common.VideoCodec
 import com.pedro.common.onMainThread
 import com.pedro.common.trySend
+import com.pedro.common.validMessage
 import com.pedro.rtmp.flv.BasePacket
 import com.pedro.rtmp.flv.FlvPacket
 import com.pedro.rtmp.flv.FlvType
@@ -183,7 +184,7 @@ class RtmpSender(
         }.exceptionOrNull()
         if (error != null) {
           onMainThread {
-            connectChecker.onConnectionFailed("Error send packet, " + error.message)
+            connectChecker.onConnectionFailed("Error send packet, ${error.validMessage()}")
           }
           Log.e(TAG, "send error: ", error)
           return@launch

rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpTunneledSocket.kt

@@ -83,7 +83,7 @@ class TcpTunneledSocket(private val host: String, private val port: Int, private
       val bytes = socket.inputStream.readBytes()
       if (bytes.size > 1) input = ByteArrayInputStream(bytes, 1, bytes.size)
       val success = socket.responseCode == HttpURLConnection.HTTP_OK
-      if (!success) throw IOException("send packet failed: ${socket.responseMessage}")
+      if (!success) throw IOException("send packet failed: ${socket.responseMessage}, broken pipe")
     } finally {
       socket.disconnect()
     }
@@ -96,7 +96,7 @@ class TcpTunneledSocket(private val host: String, private val port: Int, private
       socket.connect()
       val data = socket.inputStream.readBytes()
       val success = socket.responseCode == HttpURLConnection.HTTP_OK
-      if (!success) throw IOException("receive packet failed: ${socket.responseMessage}")
+      if (!success) throw IOException("receive packet failed: ${socket.responseMessage}, broken pipe")
       return data
     } finally {
       socket.disconnect()

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

@@ -20,10 +20,12 @@ import android.media.MediaCodec
 import android.util.Log
 import com.pedro.common.AudioCodec
 import com.pedro.common.ConnectChecker
+import com.pedro.common.ConnectionFailed
 import com.pedro.common.UrlParser
 import com.pedro.common.VideoCodec
 import com.pedro.common.onMainThread
 import com.pedro.common.socket.TcpStreamSocket
+import com.pedro.common.validMessage
 import com.pedro.rtsp.rtsp.commands.CommandsManager
 import com.pedro.rtsp.rtsp.commands.Method
 import com.pedro.rtsp.utils.RtpConstants
@@ -38,7 +40,6 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.withTimeoutOrNull
 import java.io.*
-import java.net.SocketTimeoutException
 import java.net.URISyntaxException
 import java.nio.ByteBuffer
 import javax.net.ssl.TrustManager
@@ -340,7 +341,7 @@ class RtspClient(private val connectChecker: ConnectChecker) {
         if (error != null) {
           Log.e(TAG, "connection error", error)
           onMainThread {
-            connectChecker.onConnectionFailed("Error configure stream, ${error.message}")
+            connectChecker.onConnectionFailed("Error configure stream, ${error.validMessage()}")
           }
           return@launch
         }
@@ -367,7 +368,7 @@ class RtspClient(private val connectChecker: ConnectChecker) {
           scope.cancel()
         }
       }.exceptionOrNull()
-      if (error != null && error !is SocketTimeoutException) {
+      if (ConnectionFailed.parse(error?.validMessage() ?: "") != ConnectionFailed.TIMEOUT) {
         scope.cancel()
       }
     }

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

@@ -25,6 +25,7 @@ import com.pedro.common.VideoCodec
 import com.pedro.common.onMainThread
 import com.pedro.common.socket.TcpStreamSocket
 import com.pedro.common.trySend
+import com.pedro.common.validMessage
 import com.pedro.rtsp.rtcp.BaseSenderReport
 import com.pedro.rtsp.rtp.packets.*
 import com.pedro.rtsp.rtp.sockets.BaseRtpSocket
@@ -195,7 +196,7 @@ class RtspSender(
         }.exceptionOrNull()
         if (error != null) {
           onMainThread {
-            connectChecker.onConnectionFailed("Error send packet, ${error.message}")
+            connectChecker.onConnectionFailed("Error send packet, ${error.validMessage()}")
           }
           Log.e(TAG, "send error: ", error)
           return@launch

srt/src/main/java/com/pedro/srt/srt/SrtClient.kt

@@ -20,9 +20,11 @@ import android.media.MediaCodec
 import android.util.Log
 import com.pedro.common.AudioCodec
 import com.pedro.common.ConnectChecker
+import com.pedro.common.ConnectionFailed
 import com.pedro.common.UrlParser
 import com.pedro.common.VideoCodec
 import com.pedro.common.onMainThread
+import com.pedro.common.validMessage
 import com.pedro.srt.srt.packets.ControlPacket
 import com.pedro.srt.srt.packets.DataPacket
 import com.pedro.srt.srt.packets.SrtPacket
@@ -51,7 +53,6 @@ import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull
 import java.io.IOException
-import java.net.SocketTimeoutException
 import java.net.URISyntaxException
 import java.nio.ByteBuffer
 
@@ -241,7 +242,7 @@ class SrtClient(private val connectChecker: ConnectChecker) {
         if (error != null) {
           Log.e(TAG, "connection error", error)
           onMainThread {
-            connectChecker.onConnectionFailed("Error configure stream, ${error.message}")
+            connectChecker.onConnectionFailed("Error configure stream, ${error.validMessage()}")
           }
           return@launch
         }
@@ -301,6 +302,7 @@ class SrtClient(private val connectChecker: ConnectChecker) {
     while (scope.isActive && isStreaming) {
       val error = runCatching {
         if (isAlive()) {
+          delay(2000)
           //ignore packet after connect if tunneled to avoid spam idle
           handleMessages()
         } else {
@@ -310,7 +312,7 @@ class SrtClient(private val connectChecker: ConnectChecker) {
           scope.cancel()
         }
       }.exceptionOrNull()
-      if (error != null && error !is SocketTimeoutException) {
+      if (ConnectionFailed.parse(error?.validMessage() ?: "") != ConnectionFailed.TIMEOUT) {
         scope.cancel()
       }
     }

srt/src/main/java/com/pedro/srt/srt/SrtSender.kt

@@ -23,6 +23,7 @@ import com.pedro.common.BitrateManager
 import com.pedro.common.ConnectChecker
 import com.pedro.common.onMainThread
 import com.pedro.common.trySend
+import com.pedro.common.validMessage
 import com.pedro.srt.mpeg2ts.MpegTsPacket
 import com.pedro.srt.mpeg2ts.MpegTsPacketizer
 import com.pedro.srt.mpeg2ts.Pid
@@ -171,7 +172,7 @@ class SrtSender(
         }.exceptionOrNull()
         if (error != null) {
           onMainThread {
-            connectChecker.onConnectionFailed("Error send packet, " + error.message)
+            connectChecker.onConnectionFailed("Error send packet, ${error.validMessage()}")
           }
           Log.e(TAG, "send error: ", error)
           return@launch

udp/src/main/java/com/pedro/udp/UdpClient.kt

@@ -23,6 +23,7 @@ import com.pedro.common.ConnectChecker
 import com.pedro.common.UrlParser
 import com.pedro.common.VideoCodec
 import com.pedro.common.onMainThread
+import com.pedro.common.validMessage
 import com.pedro.udp.utils.UdpSocket
 import com.pedro.udp.utils.UdpType
 import kotlinx.coroutines.CoroutineScope
@@ -169,7 +170,7 @@ class UdpClient(private val connectChecker: ConnectChecker) {
         if (error != null) {
           Log.e(TAG, "connection error", error)
           onMainThread {
-            connectChecker.onConnectionFailed("Error configure stream, ${error.message}")
+            connectChecker.onConnectionFailed("Error configure stream, ${error.validMessage()}")
           }
           return@launch
         }

udp/src/main/java/com/pedro/udp/UdpSender.kt

@@ -23,6 +23,7 @@ import com.pedro.common.BitrateManager
 import com.pedro.common.ConnectChecker
 import com.pedro.common.onMainThread
 import com.pedro.common.trySend
+import com.pedro.common.validMessage
 import com.pedro.srt.mpeg2ts.MpegTsPacket
 import com.pedro.srt.mpeg2ts.MpegTsPacketizer
 import com.pedro.srt.mpeg2ts.Pid
@@ -171,7 +172,7 @@ class UdpSender(
         }.exceptionOrNull()
         if (error != null) {
           onMainThread {
-            connectChecker.onConnectionFailed("Error send packet, " + error.message)
+            connectChecker.onConnectionFailed("Error send packet, ${error.validMessage()}")
           }
           Log.e(TAG, "send error: ", error)
           return@launch