add setSocketTimeout method and add speed parameter to Media3 sources

pedroSG94 · pedroSG94 · commit a0665ee6739a · 2025-09-23T12:08:28.000+02:00
diff --git a/common/src/main/java/com/pedro/common/socket/base/StreamSocket.kt b/common/src/main/java/com/pedro/common/socket/base/StreamSocket.kt
@@ -28,30 +28,39 @@ import javax.net.ssl.TrustManager
  * Created by pedro on 22/9/24.
  */
 abstract class StreamSocket {
-  protected val timeout = 5000L
+  protected var timeout = DEFAULT_TIMEOUT
   abstract suspend fun connect()
   abstract suspend fun close()
   abstract fun isConnected(): Boolean
   abstract fun isReachable(): Boolean
 
   companion object {
+    const val DEFAULT_TIMEOUT = 5000L
     fun createTcpSocket(
       type: SocketType,
-      host: String, port: Int, secured: Boolean, certificates: TrustManager? = null
+      host: String, port: Int, secured: Boolean,
+      timeout: Long,
+      certificates: TrustManager? = null
     ): TcpStreamSocket {
       return when (type) {
         SocketType.KTOR -> TcpStreamSocketKtor(host, port, secured, certificates)
         SocketType.JAVA -> TcpStreamSocketJava(host, port, secured, certificates)
+      }.apply {
+        this.timeout = timeout
       }
     }
 
     fun createUdpSocket(
       type: SocketType,
-      host: String, port: Int, sourcePort: Int? = null, receiveSize: Int? = null, udpType: UdpType = UdpType.UNICAST
+      host: String, port: Int, timeout: Long,
+      sourcePort: Int? = null, receiveSize: Int? = null,
+      udpType: UdpType = UdpType.UNICAST
     ): UdpStreamSocket {
       return when (type) {
         SocketType.KTOR -> UdpStreamSocketKtor(host, port, sourcePort, receiveSize, udpType)
         SocketType.JAVA -> UdpStreamSocketJava(host, port, sourcePort, receiveSize, udpType)
+      }.apply {
+        this.timeout = timeout
       }
     }
   }
diff --git a/extra-sources/src/main/java/com/pedro/extrasources/Media3AudioSource.kt b/extra-sources/src/main/java/com/pedro/extrasources/Media3AudioSource.kt
@@ -4,6 +4,7 @@ import android.content.Context
 import android.net.Uri
 import androidx.annotation.OptIn
 import androidx.media3.common.MediaItem
+import androidx.media3.common.PlaybackParameters
 import androidx.media3.common.Player
 import androidx.media3.common.util.UnstableApi
 import androidx.media3.exoplayer.ExoPlayer
@@ -17,7 +18,8 @@ import com.pedro.extrasources.extractor.Media3Extractor
 @OptIn(UnstableApi::class)
 class Media3AudioSource(
     private val context: Context,
-    private val path: Uri
+    private val path: Uri,
+    private val speed: Float = 1f
 ): AudioSource() {
 
     private var player: ExoPlayer? = null
@@ -32,7 +34,7 @@ class Media3AudioSource(
         try {
             mediaExtractor.initialize(context, path)
             mediaExtractor.selectTrack(MediaFrame.Type.AUDIO)
-        } catch (e: Exception) {
+        } catch (_: Exception) {
             throw IllegalArgumentException("Audio file track not found")
         }
         val audioInfo = mediaExtractor.getAudioInfo()
@@ -46,6 +48,7 @@ class Media3AudioSource(
         player = ExoPlayer.Builder(context, TracksRenderersFactory(context, MediaFrame.Type.AUDIO, processor)).build().also { exoPlayer ->
             val mediaItem = MediaItem.fromUri(path)
             exoPlayer.setMediaItem(mediaItem)
+            exoPlayer.playbackParameters = PlaybackParameters(speed)
             exoPlayer.prepare()
             exoPlayer.repeatMode = Player.REPEAT_MODE_ALL
         }
diff --git a/extra-sources/src/main/java/com/pedro/extrasources/Media3VideoSource.kt b/extra-sources/src/main/java/com/pedro/extrasources/Media3VideoSource.kt
@@ -6,6 +6,7 @@ import android.net.Uri
 import android.view.Surface
 import androidx.annotation.OptIn
 import androidx.media3.common.MediaItem
+import androidx.media3.common.PlaybackParameters
 import androidx.media3.common.Player
 import androidx.media3.common.util.UnstableApi
 import androidx.media3.exoplayer.ExoPlayer
@@ -18,7 +19,8 @@ import com.pedro.extrasources.extractor.Media3Extractor
 @OptIn(UnstableApi::class)
 class Media3VideoSource(
     private val context: Context,
-    private val path: Uri
+    private val path: Uri,
+    private val speed: Float = 1f
 ): VideoSource() {
 
     private var player: ExoPlayer? = null
@@ -29,14 +31,15 @@ class Media3VideoSource(
         try {
             mediaExtractor.initialize(context, path)
             mediaExtractor.selectTrack(MediaFrame.Type.VIDEO)
-        } catch (e: Exception) {
+        } catch (_: Exception) {
             throw IllegalArgumentException("Video file track not found")
         }
         mediaExtractor.release()
         player = ExoPlayer.Builder(context, TracksRenderersFactory(context, MediaFrame.Type.VIDEO)).build().also { exoPlayer ->
             exoPlayer.setVideoSurface(surface)
             val mediaItem = MediaItem.fromUri(path)
             exoPlayer.setMediaItem(mediaItem)
+            exoPlayer.playbackParameters = PlaybackParameters(speed)
             exoPlayer.prepare()
             exoPlayer.repeatMode = Player.REPEAT_MODE_ALL
         }
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/GenericStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/GenericStreamClient.kt
@@ -43,6 +43,16 @@ class GenericStreamClient(
     udpClient.setSocketType(type)
   }
 
+  /**
+   * Set timeout ms for connection, write and read in sockets by default 5000ms
+   */
+  override fun setSocketTimeout(timeout: Long) {
+    rtmpClient.setSocketTimeout(timeout)
+    rtspClient.setSocketTimeout(timeout)
+    srtClient.setSocketTimeout(timeout)
+    udpClient.setSocketTimeout(timeout)
+  }
+
   /**
    * Set stream delay in millis.
    * This will create a cache and wait the delay to start send packets in real time
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt
@@ -175,6 +175,13 @@ class RtmpStreamClient(
     rtmpClient.socketType = type
   }
 
+  /**
+   * Set timeout ms for connection, write and read in sockets by default 5000ms
+   */
+  override fun setSocketTimeout(timeout: Long) {
+    rtmpClient.socketTimeout = timeout
+  }
+
   fun setCustomAmfObject(amfObject: Map<String, Any>) {
     rtmpClient.setCustomAmfObject(amfObject)
   }
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/RtspStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/RtspStreamClient.kt
@@ -149,4 +149,11 @@ class RtspStreamClient(
   override fun setSocketType(type: SocketType) {
     rtspClient.socketType = type
   }
+
+  /**
+   * Set timeout ms for connection, write and read in sockets by default 5000ms
+   */
+  override fun setSocketTimeout(timeout: Long) {
+    rtspClient.socketTimeout = timeout
+  }
 }
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/SrtStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/SrtStreamClient.kt
@@ -148,6 +148,13 @@ class SrtStreamClient(
     srtClient.socketType = type
   }
 
+  /**
+   * Set timeout ms for connection, write and read in sockets by default 5000ms
+   */
+  override fun setSocketTimeout(timeout: Long) {
+    srtClient.socketTimeout = timeout
+  }
+
   /**
    * Set a custom Mpeg2TsService with specified parameters
    * Must be called before connecting to the server
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/StreamBaseClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/StreamBaseClient.kt
@@ -79,6 +79,10 @@ abstract class StreamBaseClient {
    * Set if you want use java.io or ktor socket
    */
   abstract fun setSocketType(type: SocketType)
+  /**
+   * Set timeout ms for connection, write and read in sockets by default 5000ms
+   */
+  abstract fun setSocketTimeout(timeout: Long)
   /**
    * Set stream delay in millis.
    * This will create a cache and wait the delay to start send packets in real time
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/UdpStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/UdpStreamClient.kt
@@ -132,4 +132,11 @@ class UdpStreamClient(
   override fun setSocketType(type: SocketType) {
     udpClient.socketType = type
   }
+
+  /**
+   * Set timeout ms for connection, write and read in sockets by default 5000ms
+   */
+  override fun setSocketTimeout(timeout: Long) {
+    udpClient.socketTimeout = timeout
+  }
 }
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt
@@ -28,6 +28,7 @@ import com.pedro.common.clone
 import com.pedro.common.frame.MediaFrame
 import com.pedro.common.onMainThread
 import com.pedro.common.socket.base.SocketType
+import com.pedro.common.socket.base.StreamSocket
 import com.pedro.common.toMediaFrameInfo
 import com.pedro.common.validMessage
 import com.pedro.rtmp.amf.AmfVersion
@@ -106,6 +107,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
   val bytesSend: Long
     get() = rtmpSender.bytesSend
   var socketType = SocketType.KTOR
+  var socketTimeout = StreamSocket.DEFAULT_TIMEOUT
 
   /**
    * Add certificates for TLS connection
@@ -336,7 +338,7 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
     val socket = if (tunneled) {
       TcpTunneledSocket(commandsManager.host, commandsManager.port, tlsEnabled)
     } else {
-      TcpSocket(socketType, commandsManager.host, commandsManager.port, tlsEnabled, certificates)
+      TcpSocket(socketType, commandsManager.host, commandsManager.port, tlsEnabled, socketTimeout, certificates)
     }
     this.socket = socket
     socket.connect()
diff --git a/rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpSocket.kt b/rtmp/src/main/java/com/pedro/rtmp/utils/socket/TcpSocket.kt
@@ -29,10 +29,12 @@ import javax.net.ssl.TrustManager
  */
 class TcpSocket(
   type: SocketType,
-  host: String, port: Int, secured: Boolean, certificates: TrustManager?
+  host: String, port: Int, secured: Boolean,
+  timeout: Long,
+  certificates: TrustManager?
 ): RtmpSocket() {
 
-  private val socket = StreamSocket.createTcpSocket(type, host, port, secured, certificates)
+  private val socket = StreamSocket.createTcpSocket(type, host, port, secured, timeout, certificates)
 
   override suspend fun flush(isPacket: Boolean) {
     socket.flush()
diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtcp/BaseSenderReport.kt b/rtsp/src/main/java/com/pedro/rtsp/rtcp/BaseSenderReport.kt
@@ -46,17 +46,18 @@ abstract class BaseSenderReport internal constructor() {
     fun getInstance(
       socketType: SocketType,
       protocol: Protocol, host: String,
+      timeout: Long,
       videoSourcePort: Int?, audioSourcePort: Int?,
       videoServerPort: Int?, audioServerPort: Int?,
     ): BaseSenderReport {
       return if (protocol === Protocol.TCP) {
         SenderReportTcp()
       } else {
         val videoSocket = if (videoServerPort != null) {
-          StreamSocket.createUdpSocket(socketType, host, videoServerPort, videoSourcePort)
+          StreamSocket.createUdpSocket(socketType, host, videoServerPort, timeout, videoSourcePort)
         } else null
         val audioSocket = if (audioServerPort != null) {
-          StreamSocket.createUdpSocket(socketType, host, audioServerPort, audioSourcePort)
+          StreamSocket.createUdpSocket(socketType, host, audioServerPort, timeout, audioSourcePort)
         } else null
         SenderReportUdp(videoSocket, audioSocket)
       }
diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtp/sockets/BaseRtpSocket.kt b/rtsp/src/main/java/com/pedro/rtsp/rtp/sockets/BaseRtpSocket.kt
@@ -32,18 +32,18 @@ abstract class BaseRtpSocket {
     @JvmStatic
     fun getInstance(
       socketType: SocketType,
-      protocol: Protocol, host: String,
+      protocol: Protocol, host: String, timeout: Long,
       videoSourcePort: Int?, audioSourcePort: Int?,
       videoServerPort: Int?, audioServerPort: Int?,
     ): BaseRtpSocket {
       return if (protocol === Protocol.TCP) {
         RtpSocketTcp()
       } else {
         val videoSocket = if (videoServerPort != null) {
-          StreamSocket.createUdpSocket(socketType, host, videoServerPort, videoSourcePort)
+          StreamSocket.createUdpSocket(socketType, host, videoServerPort, timeout, videoSourcePort)
         } else null
         val audioSocket = if (audioServerPort != null) {
-          StreamSocket.createUdpSocket(socketType, host, audioServerPort, audioSourcePort)
+          StreamSocket.createUdpSocket(socketType, host, audioServerPort, timeout, audioSourcePort)
         } else null
         RtpSocketUdp(videoSocket, audioSocket)
       }
diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt
@@ -94,6 +94,7 @@ class RtspClient(private val connectChecker: ConnectChecker) {
   val bytesSend: Long
     get() = rtspSender.bytesSend
   var socketType = SocketType.KTOR
+  var socketTimeout = StreamSocket.DEFAULT_TIMEOUT
 
   /**
    * Add certificates for TLS connection
@@ -245,7 +246,7 @@ class RtspClient(private val connectChecker: ConnectChecker) {
             }
             rtspSender.setVideoInfo(commandsManager.sps!!, commandsManager.pps, commandsManager.vps)
           }
-          val socket = StreamSocket.createTcpSocket(socketType, host, port, tlsEnabled, certificates)
+          val socket = StreamSocket.createTcpSocket(socketType, host, port, tlsEnabled, socketTimeout, certificates)
           this@RtspClient.socket = socket
           socket.connect()
           socket.write(commandsManager.createOptions())
@@ -350,7 +351,7 @@ class RtspClient(private val connectChecker: ConnectChecker) {
           rtspSender.setSocketsInfo(
             socketType,
             commandsManager.protocol,
-            host,
+            host, socketTimeout,
             videoClientPorts,
             audioClientPorts,
             videoServerPorts,
diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspSender.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspSender.kt
@@ -60,12 +60,12 @@ class RtspSender(
   @Throws(IOException::class)
   fun setSocketsInfo(
     socketType: SocketType,
-    protocol: Protocol, host: String,
+    protocol: Protocol, host: String, timeout: Long,
     videoSourcePorts: Array<Int?>, audioSourcePorts: Array<Int?>,
     videoServerPorts: Array<Int?>, audioServerPorts: Array<Int?>,
   ) {
-    rtpSocket = BaseRtpSocket.getInstance(socketType, protocol, host, videoSourcePorts[0], audioSourcePorts[0], videoServerPorts[0], audioServerPorts[0])
-    baseSenderReport = BaseSenderReport.getInstance(socketType, protocol, host, videoSourcePorts[1], audioSourcePorts[1], videoServerPorts[1], audioServerPorts[1])
+    rtpSocket = BaseRtpSocket.getInstance(socketType, protocol, host, timeout, videoSourcePorts[0], audioSourcePorts[0], videoServerPorts[0], audioServerPorts[0])
+    baseSenderReport = BaseSenderReport.getInstance(socketType, protocol, host, timeout, videoSourcePorts[1], audioSourcePorts[1], videoServerPorts[1], audioServerPorts[1])
   }
 
   @Throws(IOException::class)
diff --git a/rtsp/src/test/java/com/pedro/rtsp/rtcp/RtcpReportTest.kt b/rtsp/src/test/java/com/pedro/rtsp/rtcp/RtcpReportTest.kt
@@ -18,6 +18,7 @@ package com.pedro.rtsp.rtcp
 
 import com.pedro.common.TimeUtils
 import com.pedro.common.socket.base.SocketType
+import com.pedro.common.socket.base.StreamSocket
 import com.pedro.common.socket.base.TcpStreamSocket
 import com.pedro.common.socket.base.UdpStreamSocket
 import com.pedro.rtsp.Utils
@@ -65,7 +66,8 @@ class RtcpReportTest {
   @Test
   fun `GIVEN multiple video or audio rtp frames WHEN update rtcp tcp send THEN send only 1 of video and 1 of audio each 3 seconds`() = runTest {
     Utils.useStatics(listOf(timeUtilsMocked)) {
-      val senderReportTcp = BaseSenderReport.getInstance(SocketType.JAVA, Protocol.TCP, "127.0.0.1", 0, 1, 2, 3)
+      val senderReportTcp = BaseSenderReport.getInstance(SocketType.JAVA, Protocol.TCP, "127.0.0.1",
+        StreamSocket.DEFAULT_TIMEOUT, 0, 1, 2, 3)
       senderReportTcp.setSocket(tcpSocket)
       senderReportTcp.setSSRC(0, 1)
       val fakeFrameVideo = RtpFrame(byteArrayOf(0x00, 0x00, 0x00), 0, 3, RtpConstants.trackVideo)
diff --git a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt
@@ -27,6 +27,7 @@ import com.pedro.common.clone
 import com.pedro.common.frame.MediaFrame
 import com.pedro.common.onMainThread
 import com.pedro.common.socket.base.SocketType
+import com.pedro.common.socket.base.StreamSocket
 import com.pedro.common.toMediaFrameInfo
 import com.pedro.common.validMessage
 import com.pedro.srt.mpeg2ts.service.Mpeg2TsService
@@ -106,6 +107,7 @@ class SrtClient(private val connectChecker: ConnectChecker) {
     private set
   private var latency = 120_000 //in micro
   var socketType = SocketType.KTOR
+  var socketTimeout = StreamSocket.DEFAULT_TIMEOUT
 
   fun setVideoCodec(videoCodec: VideoCodec) {
     if (!isStreaming) {
@@ -225,7 +227,7 @@ class SrtClient(private val connectChecker: ConnectChecker) {
         commandsManager.host = host
 
         val error = runCatching {
-          socket = SrtSocket(socketType, host, port)
+          socket = SrtSocket(socketType, host, port, socketTimeout)
           socket?.connect()
           commandsManager.loadStartTs()
 
diff --git a/srt/src/main/java/com/pedro/srt/utils/SrtSocket.kt b/srt/src/main/java/com/pedro/srt/utils/SrtSocket.kt
@@ -23,9 +23,9 @@ import com.pedro.srt.srt.packets.SrtPacket
 /**
  * Created by pedro on 22/8/23.
  */
-class SrtSocket(type: SocketType, host: String, port: Int) {
+class SrtSocket(type: SocketType, host: String, port: Int, timeout: Long) {
 
-  private val socket = StreamSocket.createUdpSocket(type, host, port, receiveSize = Constants.MTU)
+  private val socket = StreamSocket.createUdpSocket(type, host, port, timeout, receiveSize = Constants.MTU)
 
   suspend fun connect() {
     socket.connect()
diff --git a/udp/src/main/java/com/pedro/udp/UdpClient.kt b/udp/src/main/java/com/pedro/udp/UdpClient.kt
@@ -26,6 +26,7 @@ import com.pedro.common.clone
 import com.pedro.common.frame.MediaFrame
 import com.pedro.common.onMainThread
 import com.pedro.common.socket.base.SocketType
+import com.pedro.common.socket.base.StreamSocket
 import com.pedro.common.socket.base.UdpType
 import com.pedro.common.toMediaFrameInfo
 import com.pedro.common.validMessage
@@ -79,6 +80,7 @@ class UdpClient(private val connectChecker: ConnectChecker) {
   val bytesSend: Long
     get() = udpSender.bytesSend
   var socketType = SocketType.KTOR
+  var socketTimeout = StreamSocket.DEFAULT_TIMEOUT
 
   fun setVideoCodec(videoCodec: VideoCodec) {
     if (!isStreaming) {
@@ -169,7 +171,7 @@ class UdpClient(private val connectChecker: ConnectChecker) {
 
         val error = runCatching {
           val type = UdpType.getTypeByHost(host)
-          socket = UdpSocket(socketType, host, type, port)
+          socket = UdpSocket(socketType, host, type, port, socketTimeout)
           socket?.connect()
 
           udpSender.socket = socket
diff --git a/udp/src/main/java/com/pedro/udp/utils/UdpSocket.kt b/udp/src/main/java/com/pedro/udp/utils/UdpSocket.kt

Original file line number	Diff line number	Diff line change
`@@ -175,6 +175,13 @@ class RtmpStreamClient(`
`175`	`175`	`rtmpClient.socketType = type`
`176`	`176`	`}`
`177`	`177`
	`178`	`+ /**`
	`179`	`+ * Set timeout ms for connection, write and read in sockets by default 5000ms`
	`180`	`+ */`
	`181`	`+ override fun setSocketTimeout(timeout: Long) {`
	`182`	`+ rtmpClient.socketTimeout = timeout`
	`183`	`+ }`
	`184`	`+`
`178`	`185`	`fun setCustomAmfObject(amfObject: Map<String, Any>) {`
`179`	`186`	`rtmpClient.setCustomAmfObject(amfObject)`
`180`	`187`	`}`
Original file line number	Diff line number	Diff line change
`@@ -149,4 +149,11 @@ class RtspStreamClient(`
`149`	`149`	`override fun setSocketType(type: SocketType) {`
`150`	`150`	`rtspClient.socketType = type`
`151`	`151`	`}`
	`152`	`+`
	`153`	`+ /**`
	`154`	`+ * Set timeout ms for connection, write and read in sockets by default 5000ms`
	`155`	`+ */`
	`156`	`+ override fun setSocketTimeout(timeout: Long) {`
	`157`	`+ rtspClient.socketTimeout = timeout`
	`158`	`+ }`
`152`	`159`	`}`
Original file line number	Diff line number	Diff line change
`@@ -132,4 +132,11 @@ class UdpStreamClient(`
`132`	`132`	`override fun setSocketType(type: SocketType) {`
`133`	`133`	`udpClient.socketType = type`
`134`	`134`	`}`
	`135`	`+`
	`136`	`+ /**`
	`137`	`+ * Set timeout ms for connection, write and read in sockets by default 5000ms`
	`138`	`+ */`
	`139`	`+ override fun setSocketTimeout(timeout: Long) {`
	`140`	`+ udpClient.socketTimeout = timeout`
	`141`	`+ }`
`135`	`142`	`}`