Make so RootEncoder delivers bandwidth info needed for bitrate changing

Author: bratta-dk

app/src/main/java/com/pedro/streamer/file/FromFileActivity.kt

@@ -31,6 +31,7 @@ import androidx.activity.result.contract.ActivityResultContracts
 import androidx.annotation.RequiresApi
 import androidx.appcompat.app.AppCompatActivity
 import com.pedro.common.ConnectChecker
+import com.pedro.common.Throughput
 import com.pedro.encoder.input.decoder.AudioDecoderInterface
 import com.pedro.encoder.input.decoder.VideoDecoderInterface
 import com.pedro.library.base.recording.RecordController
@@ -194,6 +195,13 @@ class FromFileActivity : AppCompatActivity(), ConnectChecker,
 
   override fun onNewBitrate(bitrate: Long) {}
 
+  override fun onStreamingStats(
+    bitrate: Long,
+    bytesSent: Long,
+    bytesQueued: Long,
+    throughput: Throughput
+  ) {
+  }
   override fun onDisconnect() {
     toast("Disconnected")
   }

app/src/main/java/com/pedro/streamer/oldapi/OldApiActivity.kt

@@ -25,6 +25,7 @@ import android.widget.ImageView
 import android.widget.Toast
 import androidx.appcompat.app.AppCompatActivity
 import com.pedro.common.ConnectChecker
+import com.pedro.common.Throughput
 import com.pedro.encoder.input.video.CameraHelper
 import com.pedro.encoder.input.video.CameraOpenException
 import com.pedro.library.base.recording.RecordController
@@ -151,6 +152,13 @@ class OldApiActivity : AppCompatActivity(), ConnectChecker, TextureView.SurfaceT
 
   override fun onNewBitrate(bitrate: Long) {}
 
+  override fun onStreamingStats(
+    bitrate: Long,
+    bytesSent: Long,
+    bytesQueued: Long,
+    throughput: Throughput
+  ) {
+  }
   override fun onDisconnect() {
     toast("Disconnected")
   }

common/src/main/java/com/pedro/common/BitrateChecker.java

@@ -21,4 +21,6 @@
  */
 public interface BitrateChecker {
   default void onNewBitrate(long bitrate) {}
+
+  default void onStreamingStats(long bitrate, long bytesSent, long bytesQueued, Throughput throughput) {}
 }

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

@@ -30,6 +30,63 @@ open class BitrateManager(private val bitrateChecker: BitrateChecker) {
   var exponentialFactor: Float = 1f
   private var timeStamp = TimeUtils.getCurrentTimeMillis()
 
+  private var byterate = 0L
+  private var queuedBytes: Long = 0
+  private val measureInterval = 3
+  private val previousQueueBytesOut: MutableList<Long> = mutableListOf()
+
+  fun queueBytes(size: Long) {
+    queuedBytes += size
+  }
+
+  suspend fun calculateBitrateAndBandwidth(bytesSendPerSecond: Long, myQueueValue: Long) {
+    // Track bitrate and queue management
+    bitrate += (bytesSendPerSecond * 8)
+    byterate += bytesSendPerSecond
+    queuedBytes -= bytesSendPerSecond
+    
+    val timeDiff = TimeUtils.getCurrentTimeMillis() - timeStamp
+    if (timeDiff >= 1000) {
+      // Calculate bitrate with exponential moving average
+      val currentValue = (bitrate / (timeDiff / 1000f)).toLong()
+      if (bitrateOld == 0L) { bitrateOld = currentValue }
+      bitrateOld = (bitrateOld + exponentialFactor * (currentValue - bitrateOld)).toLong()
+
+      // Calculate bandwidth and throughput analysis
+      var throughput: Throughput = Throughput.Unknown
+      previousQueueBytesOut.add(queuedBytes)
+      if (measureInterval <= previousQueueBytesOut.size) {
+        var countQueuedBytesGrowing = 0
+        for (i in 0 until previousQueueBytesOut.size - 1) {
+          if (previousQueueBytesOut[i] < previousQueueBytesOut[i + 1]) {
+            countQueuedBytesGrowing++
+          }
+        }
+        if (countQueuedBytesGrowing == measureInterval - 1) {
+          throughput = Throughput.Insufficient
+        } else if (countQueuedBytesGrowing == 0) {
+          throughput = Throughput.Sufficient
+        }
+        previousQueueBytesOut.removeAt(0)
+      }
+
+      // Call both callbacks on main thread
+      onMainThread {
+        bitrateChecker.onNewBitrate(bitrateOld)
+        bitrateChecker.onStreamingStats(
+          ((bytesSendPerSecond * 8) / (timeDiff / 1000f)).toLong(),
+          bytesSendPerSecond,
+          myQueueValue,
+          throughput
+        )
+      }
+
+      // Reset all counters
+      timeStamp = TimeUtils.getCurrentTimeMillis()
+      bitrate = 0
+    }
+  }
+
   suspend fun calculateBitrate(size: Long) {
     bitrate += size
     val timeDiff = TimeUtils.getCurrentTimeMillis() - timeStamp
@@ -44,7 +101,9 @@ open class BitrateManager(private val bitrateChecker: BitrateChecker) {
   }
 
   fun reset() {
-    bitrate = 0
+    byterate = 0
     bitrateOld = 0
+    queuedBytes = 0
+    previousQueueBytesOut.clear()
   }
 }

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

@@ -59,4 +59,20 @@ class StreamBlockingQueue(size: Int) {
     }
 
     fun getSize() = queue.size
+
+
+    /**
+     * Gets the total size of all items in the queue by summing the individual sizes.
+     *
+     * This method calculates the cumulative size of all [MediaFrame] objects currently
+     * in the queue by summing their [MediaFrame.info.size] values. This is useful for
+     * monitoring memory usage, bandwidth calculations, or determining the total data
+     * volume in the queue.
+     *
+     * @return The total size in bytes as a [Long] value, representing the sum of all
+     *         individual frame sizes in the queue.
+     *
+     * @see getSize for getting the count of items in the queue
+     */
+    fun getTotalSize(): Long = queue.sumOf { it.info.size.toLong() }
 }

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

@@ -0,0 +1,7 @@
+package com.pedro.common
+
+public enum class Throughput {
+    Unknown,
+    Sufficient,
+    Insufficient
+}

common/src/main/java/com/pedro/common/base/BaseSender.kt

@@ -47,16 +47,21 @@ abstract class BaseSender(
     protected abstract suspend fun stopImp(clear: Boolean = true)
 
     fun sendMediaFrame(mediaFrame: MediaFrame) {
-        if (running && !queue.trySend(mediaFrame)) {
-            when (mediaFrame.type) {
-                MediaFrame.Type.VIDEO -> {
-                    Log.i(TAG, "Video frame discarded")
-                    droppedVideoFrames++
-                }
-                MediaFrame.Type.AUDIO -> {
-                    Log.i(TAG, "Audio frame discarded")
-                    droppedAudioFrames++
+        if (running){
+            if(!queue.trySend(mediaFrame)) {
+                when (mediaFrame.type) {
+                    MediaFrame.Type.VIDEO -> {
+                        Log.i(TAG, "Video frame discarded")
+                        droppedVideoFrames++
+                    }
+
+                    MediaFrame.Type.AUDIO -> {
+                        Log.i(TAG, "Audio frame discarded")
+                        droppedAudioFrames++
+                    }
                 }
+            } else {
+                bitrateManager.queueBytes(mediaFrame.info.size.toLong())
             }
         }
     }
@@ -68,8 +73,7 @@ abstract class BaseSender(
         job = scope.launch {
             val bitrateTask = async {
                 while (scope.isActive && running) {
-                    //bytes to bits
-                    bitrateManager.calculateBitrate(bytesSendPerSecond * 8)
+                    bitrateManager.calculateBitrateAndBandwidth(bytesSendPerSecond, queue.getTotalSize())
                     bytesSendPerSecond = 0
                     delay(timeMillis = 1000)
                 }

common/src/test/java/com/pedro/common/BitrateManagerTest.kt

@@ -20,6 +20,7 @@ import com.pedro.common.util.MainDispatcherRule
 import com.pedro.common.util.Utils
 import kotlinx.coroutines.test.runTest
 import org.junit.After
+import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Rule
@@ -76,4 +77,56 @@ class BitrateManagerTest {
       assertTrue(expectedResult - marginError <= resultValue.firstValue && resultValue.firstValue <= expectedResult + marginError)
     }
   }
+
+  @Test
+  fun `WHEN calculateBitrateAndBandwidth called multiple times THEN return total bitrate and bandwidth stats`() = runTest {
+    Utils.useStatics(listOf(timeUtilsMocked)) {
+      val bitrateManager = BitrateManager(connectChecker)
+      val fakeValues = arrayOf(100L, 200L, 300L, 400L, 500L)
+      val queueValues = arrayOf(50L, 100L, 150L, 200L, 250L)
+      var expectedBitrate = 0L
+      
+      // Call calculateBitrateAndBandwidth multiple times within the same second
+      fakeValues.forEachIndexed { index, value ->
+        bitrateManager.calculateBitrateAndBandwidth(value, queueValues[index])
+        expectedBitrate += (value * 8) // Convert bytes to bits
+      }
+      
+      // Advance time by 1 second to trigger the calculation
+      fakeTime += 1000
+      
+      // Call one more time to trigger the final calculation
+      val finalValue = 100L
+      val finalQueueValue = 300L
+      bitrateManager.calculateBitrateAndBandwidth(finalValue, finalQueueValue)
+      expectedBitrate += (finalValue * 8)
+      
+      // Verify onNewBitrate was called
+      val bitrateCaptor = argumentCaptor<Long>()
+      verify(connectChecker, times(1)).onNewBitrate(bitrateCaptor.capture())
+      
+      // Verify onStreamingStats was called
+      val bitrateStatsCaptor = argumentCaptor<Long>()
+      val bytesSentCaptor = argumentCaptor<Long>()
+      val bytesQueuedCaptor = argumentCaptor<Long>()
+      val throughputCaptor = argumentCaptor<Throughput>()
+      
+      verify(connectChecker, times(1)).onStreamingStats(
+        bitrateStatsCaptor.capture(),
+        bytesSentCaptor.capture(),
+        bytesQueuedCaptor.capture(),
+        throughputCaptor.capture()
+      )
+      
+      // Check bitrate calculation (with margin for exponential moving average)
+      val marginError = 100
+      assertTrue(expectedBitrate - marginError <= bitrateCaptor.firstValue && bitrateCaptor.firstValue <= expectedBitrate + marginError)
+      
+      // Check streaming stats
+      assertTrue(bitrateStatsCaptor.firstValue > 0)
+      assertTrue(bytesSentCaptor.firstValue > 0)
+      assertEquals(finalQueueValue, bytesQueuedCaptor.firstValue)
+      assertEquals(Throughput.Unknown, throughputCaptor.firstValue) // Should be Unknown for first measurement
+    }
+  }
 }