fix ports on TCP connection, fix start with ts 0 for few players, add…

… onNewBitrate for each client
pedroSG94 · Oct 28, 2024 · ada9ad0 · ada9ad0
1 parent 05f0686
commit ada9ad0
Show file tree

Hide file tree

Showing 8 changed files with 63 additions and 32 deletions.
diff --git a/app/src/main/java/com/pedro/sample/CameraDemoActivity.kt b/app/src/main/java/com/pedro/sample/CameraDemoActivity.kt
@@ -148,6 +148,8 @@ class CameraDemoActivity : AppCompatActivity(), ConnectChecker, ClientListener,
     toast("Client disconnected: ${client.getAddress()}")
   }
 
+  override fun onClientNewBitrate(bitrate: Long, client: ServerClient) {}
+
   override fun onSurfaceTextureAvailable(surface: SurfaceTexture, width: Int, height: Int) {
     if (!rtspServerCamera1.isOnPreview) {
       rtspServerCamera1.startPreview()

diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -1,7 +1,7 @@
 [versions]
 libraryGroup = "com.github.pedroSG94"
-versionCode = "133"
-versionName = "1.3.3"
+versionCode = "134"
+versionName = "1.3.4"
 
 #plugins versions
 agp = "8.6.1"

diff --git a/rtspserver/src/main/java/com/pedro/rtspserver/server/ClientListener.kt b/rtspserver/src/main/java/com/pedro/rtspserver/server/ClientListener.kt
@@ -8,4 +8,6 @@ interface ClientListener {
   fun onClientConnected(client: ServerClient)
 
   fun onClientDisconnected(client: ServerClient)
+
+  fun onClientNewBitrate(bitrate: Long, client: ServerClient)
 }
diff --git a/rtspserver/src/main/java/com/pedro/rtspserver/server/RtspServer.kt b/rtspserver/src/main/java/com/pedro/rtspserver/server/RtspServer.kt
@@ -5,6 +5,7 @@ import android.util.Log
 import com.pedro.common.AudioCodec
 import com.pedro.common.ConnectChecker
 import com.pedro.common.VideoCodec
+import com.pedro.common.onMainThread
 import com.pedro.common.onMainThreadHandler
 import com.pedro.rtsp.utils.RtpConstants
 import io.ktor.network.selector.SelectorManager
@@ -33,7 +34,7 @@ import java.util.concurrent.TimeUnit
 class RtspServer(
   private val connectChecker: ConnectChecker,
   val port: Int
-): ServerListener {
+): ClientListener {
 
   private val TAG = "RtspServer"
   private var server: ServerSocket? = null
@@ -100,8 +101,8 @@ class RtspServer(
             semaphore.tryAcquire(5000, TimeUnit.MILLISECONDS)
           }
           if (!serverCommandManager.videoInfoReady()) {
-            onMainThreadHandler {
-              connectChecker.onConnectionFailed("video info is null")
+            onMainThread {
+              connectChecker.onConnectionFailed("Video info is null")
             }
             return@launch
           }
@@ -110,7 +111,7 @@ class RtspServer(
         server = aSocket(selectorManager).tcp().bind("0.0.0.0", port)
         Log.i(TAG, "Server started: $serverIp:$port")
       } catch (e: Exception) {
-        onMainThreadHandler {
+        onMainThread {
           connectChecker.onConnectionFailed("Server creation failed")
         }
         Log.e(TAG, "Error", e)
@@ -122,16 +123,13 @@ class RtspServer(
           val socket = server?.accept() ?: continue
           val clientSocket = ClientSocket(socket)
           Log.i(TAG, "Client connected: ${clientSocket.getHost()}")
-          val client = ServerClient(clientSocket, serverIp, port, connectChecker,
+          val client = ServerClient(clientSocket, serverIp, port,
             serverCommandManager, this@RtspServer)
           client.setLogs(isEnableLogs)
           client.startClient()
           synchronized(clients) {
             clients.add(client)
           }
-          onMainThreadHandler {
-            clientListener?.onClientConnected(client)
-          }
         } catch (e: IOException) {
           // server.close called
           break
@@ -308,14 +306,20 @@ class RtspServer(
     }
   }
 
+  override fun onClientConnected(client: ServerClient) {
+    onMainThreadHandler { clientListener?.onClientConnected(client) }
+  }
+
   override fun onClientDisconnected(client: ServerClient) {
     synchronized(clients) {
       client.stopClient()
       clients.remove(client)
-      onMainThreadHandler {
-        clientListener?.onClientDisconnected(client)
-      }
     }
+    onMainThreadHandler { clientListener?.onClientDisconnected(client) }
+  }
+
+  override fun onClientNewBitrate(bitrate: Long, client: ServerClient) {
+    onMainThreadHandler { clientListener?.onClientNewBitrate(bitrate, client) }
   }
 
   private fun getIPAddress(): String {

diff --git a/rtspserver/src/main/java/com/pedro/rtspserver/server/ServerClient.kt b/rtspserver/src/main/java/com/pedro/rtspserver/server/ServerClient.kt
@@ -5,10 +5,9 @@ import android.util.Log
 import com.pedro.common.ConnectChecker
 import com.pedro.common.clone
 import com.pedro.common.frame.MediaFrame
-import com.pedro.common.onMainThreadHandler
-import com.pedro.common.toMediaFrameInfo
 import com.pedro.rtsp.rtsp.RtspSender
 import com.pedro.rtsp.rtsp.commands.Method
+import com.pedro.rtspserver.util.toMediaFrameInfo
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -19,17 +18,32 @@ import java.nio.ByteBuffer
 
 class ServerClient(
   private val socket: ClientSocket, serverIp: String, serverPort: Int,
-  private val connectChecker: ConnectChecker,
   private val serverCommandManager: ServerCommandManager,
-  private val listener: ServerListener
+  private val listener: ClientListener
 ) {
 
   private val TAG = "Client"
+  private val connectChecker = object: ConnectChecker {
+    override fun onAuthError() {}
+    override fun onAuthSuccess() {}
+    override fun onConnectionStarted(url: String) {}
+    override fun onConnectionSuccess() {}
+    override fun onDisconnect() {}
+
+    override fun onNewBitrate(bitrate: Long) {
+      listener.onClientNewBitrate(bitrate, this@ServerClient)
+    }
+    override fun onConnectionFailed(reason: String) {
+      listener.onClientDisconnected(this@ServerClient)
+    }
+  }
   private val rtspSender = RtspSender(connectChecker, serverCommandManager)
   private val scope = CoroutineScope(Dispatchers.IO)
   private var job: Job? = null
   var canSend = false
     private set
+  //few players need start with timestamp 0 to work
+  private var startTs = 0L
 
   val droppedAudioFrames: Long
     get() = rtspSender.droppedAudioFrames
@@ -49,6 +63,7 @@ class ServerClient(
 
   fun startClient() {
     job = scope.launch {
+      startTs = 0L
       socket.connect()
       while (job?.isActive == true) {
         try {
@@ -96,16 +111,11 @@ class ServerClient(
             )
             rtspSender.setSocket(socket)
             rtspSender.start()
-            onMainThreadHandler {
-              connectChecker.onConnectionSuccess()
-            }
+            listener.onClientConnected(this@ServerClient)
             canSend = true
           } else if (request.method == Method.TEARDOWN) {
             Log.i(TAG, "Client disconnected")
             listener.onClientDisconnected(this@ServerClient)
-            onMainThreadHandler {
-              connectChecker.onDisconnect()
-            }
           }
         } catch (e: IOException) { // Client has left
           Log.e(TAG, "Client disconnected", e)
@@ -121,6 +131,7 @@ class ServerClient(
   fun stopClient() {
     CoroutineScope(Dispatchers.IO).launch {
       canSend = false
+      startTs = 0L
       rtspSender.stop()
       job?.cancelAndJoin()
       job = null
@@ -166,11 +177,17 @@ class ServerClient(
   fun getItemsInCache(): Int = rtspSender.getItemsInCache()
 
   fun sendVideoFrame(videoBuffer: ByteBuffer, info: MediaCodec.BufferInfo) {
-    rtspSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.VIDEO))
+    if (canSend) {
+      if (startTs == 0L) startTs = info.presentationTimeUs
+      rtspSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(startTs), MediaFrame.Type.VIDEO))
+    }
   }
 
   fun sendAudioFrame(audioBuffer: ByteBuffer, info: MediaCodec.BufferInfo) {
-    rtspSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.AUDIO))
+    if (canSend) {
+      if (startTs == 0L) startTs = info.presentationTimeUs
+      rtspSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(startTs), MediaFrame.Type.AUDIO))
+    }
   }
 
   fun setBitrateExponentialFactor(factor: Float) {

diff --git a/rtspserver/src/main/java/com/pedro/rtspserver/server/ServerCommandManager.kt b/rtspserver/src/main/java/com/pedro/rtspserver/server/ServerCommandManager.kt
@@ -28,8 +28,8 @@ class ServerCommandManager: CommandsManager() {
   private var serverPort: Int = 0
 
   private val TAG = "ServerCommandManager"
-  var audioPorts = ArrayList<Int?>()
-  var videoPorts = ArrayList<Int?>()
+  var audioPorts = mutableListOf<Int?>(null, null)
+  var videoPorts = mutableListOf<Int?>(null, null)
 
   fun setServerInfo(serverIp: String, serverPort: Int) {
     this.serverIp = serverIp

diff --git a/rtspserver/src/main/java/com/pedro/rtspserver/server/ServerListener.kt b/rtspserver/src/main/java/com/pedro/rtspserver/server/ServerListener.kt
diff --git a/rtspserver/src/main/java/com/pedro/rtspserver/util/Extensions.kt b/rtspserver/src/main/java/com/pedro/rtspserver/util/Extensions.kt
@@ -0,0 +1,11 @@
+package com.pedro.rtspserver.util
+
+import android.media.MediaCodec
+import com.pedro.common.frame.MediaFrame
+import com.pedro.common.isKeyframe
+
+/**
+ * Created by pedro on 28/10/24.
+ */
+
+fun MediaCodec.BufferInfo.toMediaFrameInfo(startTs: Long) = MediaFrame.Info(offset, size, presentationTimeUs - startTs, isKeyframe())