Stream stats

app/src/main/java/com/pedro/streamer/rotation/CameraFragment.kt

@@ -30,6 +30,8 @@ import android.widget.TextView
 import androidx.annotation.RequiresApi
 import androidx.fragment.app.Fragment
 import com.pedro.common.ConnectChecker
+import com.pedro.common.StreamingStatsReport
+import com.pedro.common.Throughput
 import com.pedro.common.onMainThreadHandler
 import com.pedro.encoder.input.sources.video.Camera1Source
 import com.pedro.encoder.input.sources.video.Camera2Source
@@ -219,10 +221,18 @@ class CameraFragment: Fragment(), ConnectChecker {
     }
   }
 
-  override fun onNewBitrate(bitrate: Long) {
+  override fun onStreamingStats(report: StreamingStatsReport) {
     onMainThreadHandler {
-      bitrateAdapter.adaptBitrate(bitrate, genericStream.getStreamClient().hasCongestion())
-      txtBitrate.text = String.format(Locale.getDefault(), "%.1f mb/s", bitrate / 1000_000f)
+      bitrateAdapter.adaptBitrate(report.smoothedBitrate, genericStream.getStreamClient().hasCongestion())
+      if (report.throughput != Throughput.UNKNOWN) {
+        txtBitrate.text = String.format(
+          Locale.getDefault(),
+          "%.1f mb/s [%s, queue %d KB]",
+          report.smoothedBitrate / 1000_000f,
+          report.throughput.name.lowercase(),
+          report.queueBytesOut / 1024,
+        )
+      }
     }
   }
 

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

@@ -21,4 +21,6 @@
  */
 public interface BitrateChecker {
   default void onNewBitrate(long bitrate) {}
+
+  default void onStreamingStats(StreamingStatsReport report) {}
 }

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

@@ -37,12 +37,14 @@ open class BitrateManager(private val bitrateChecker: BitrateChecker) {
       val currentValue = (bitrate / (timeDiff / 1000f)).toLong()
       if (bitrateOld == 0L) { bitrateOld = currentValue }
       bitrateOld = (bitrateOld + exponentialFactor * (currentValue - bitrateOld)).toLong()
-      onMainThread { bitrateChecker.onNewBitrate(bitrateOld) }
       timeStamp = TimeUtils.getCurrentTimeMillis()
       bitrate = 0
+      onMainThread { bitrateChecker.onNewBitrate(bitrateOld) }
     }
   }
 
+  fun getSmoothedBitrate(): Long = bitrateOld
+
   fun reset() {
     bitrate = 0
     bitrateOld = 0

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

@@ -72,4 +72,10 @@ class StreamBlockingQueue(var capacity: Int) {
     }
 
     fun getSize() = queue.size
+
+    /**
+     * Sum of [MediaFrame.info.size] for all frames in the main send queue.
+     * Delay/cache queue bytes are excluded.
+     */
+    fun getTotalSize(): Long = queue.sumOf { it.info.size.toLong() }
 }

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright (C) 2024 pedroSG94.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.pedro.common
+
+/**
+ * Collects send-queue and throughput statistics each second and classifies bandwidth trends
+ * over a sliding window, mirroring HaishinKit NetworkMonitor behavior.
+ */
+class StreamingStatsMonitor(private val bitrateChecker: BitrateChecker) {
+
+  private val measureInterval = 3
+  private val congestionThreshold = 95f
+  private val previousQueueBytesOut: MutableList<Long> = mutableListOf()
+
+  fun reset() {
+    previousQueueBytesOut.clear()
+  }
+
+  suspend fun collect(
+    queueBytesOut: Long,
+    bytesOutPerSecond: Long,
+    totalBytesOut: Long,
+    smoothedBitrate: Long,
+    queueCongestionPercent: Float,
+  ) {
+    var throughput = Throughput.UNKNOWN
+    previousQueueBytesOut.add(queueBytesOut)
+    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)
+    }
+    // This library caps the queue at a fixed capacity,
+    // so a saturated queue stops growing and will return Insufficient
+    if (queueCongestionPercent >= congestionThreshold) {
+      throughput = Throughput.INSUFFICIENT
+    }
+
+    val report = StreamingStatsReport(
+      bytesOutPerSecond = bytesOutPerSecond,
+      queueBytesOut = queueBytesOut,
+      totalBytesOut = totalBytesOut,
+      queueCongestionPercent = queueCongestionPercent,
+      throughput = throughput,
+      bitrate = bytesOutPerSecond * 8,
+      smoothedBitrate = smoothedBitrate,
+    )
+    onMainThread { bitrateChecker.onStreamingStats(report) }
+  }
+}

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2024 pedroSG94.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.pedro.common
+
+/**
+ * Per-second streaming statistics report.
+ *
+ * Mirrors HaishinKit [NetworkMonitorReport](https://github.com/HaishinKit/HaishinKit.swift/blob/main/HaishinKit/Sources/Network/NetworkMonitorReport.swift).
+ */
+data class StreamingStatsReport(
+  val bytesOutPerSecond: Long,
+  val queueBytesOut: Long,
+  val totalBytesOut: Long,
+  /** Percent (0-100) of the send queue's fixed capacity currently in use. */
+  val queueCongestionPercent: Float,
+  val throughput: Throughput,
+  val bitrate: Long,
+  val smoothedBitrate: Long,
+)

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

@@ -0,0 +1,23 @@
+/*
+ * Copyright (C) 2024 pedroSG94.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.pedro.common
+
+enum class Throughput {
+  UNKNOWN,
+  SUFFICIENT,
+  INSUFFICIENT,
+}

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

@@ -6,7 +6,9 @@ import com.pedro.common.BufferPool
 import com.pedro.common.ConnectChecker
 import com.pedro.common.StreamBlockingQueue
 import com.pedro.common.clone
+import com.pedro.common.StreamingStatsMonitor
 import com.pedro.common.frame.MediaFrame
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -36,6 +38,7 @@ abstract class BaseSender(
     private val droppedVideoFrames = AtomicLong(0)
 
     private val bitrateManager: BitrateManager = BitrateManager(connectChecker)
+    private val streamingStatsMonitor: StreamingStatsMonitor = StreamingStatsMonitor(connectChecker)
     protected var isEnableLogs = true
     private var job: Job? = null
     protected val scope = CoroutineScope(Dispatchers.IO)
@@ -81,16 +84,34 @@ abstract class BaseSender(
         }
     }
 
-    fun start() {
+    suspend fun start() {
+        running = false
+        job?.cancelAndJoin()
         bitrateManager.reset()
         queue.clear { bufferPool.release(it.data) }
+        streamingStatsMonitor.reset()
         running = true
         job = scope.launch {
             val bitrateTask = async {
                 while (scope.isActive && running) {
-                    //bytes to bits
-                    bitrateManager.calculateBitrate(bytesSendPerSecond.get() * 8)
-                    bytesSendPerSecond.set(0)
+                    try {
+                        val bytesThisSecond = bytesSendPerSecond.getAndSet(0)
+                        //bytes to bits
+                        bitrateManager.calculateBitrate(bytesThisSecond * 8)
+                        streamingStatsMonitor.collect(
+                            queueBytesOut = queue.getTotalSize(),
+                            bytesOutPerSecond = bytesThisSecond,
+                            totalBytesOut = bytesSend.get(),
+                            smoothedBitrate = bitrateManager.getSmoothedBitrate(),
+                            queueCongestionPercent = queueUsagePercent(),
+                        )
+                    } catch (e: CancellationException) {
+                        throw e
+                    } catch (e: Exception) {
+                        //never let a reporting failure (e.g. no Main dispatcher, checker callback
+                        //throwing) take down the send loop
+                        Log.e(TAG, "bitrate/stats reporting failed", e)
+                    }
                     delay(timeMillis = 1000)
                 }
             }
@@ -115,10 +136,14 @@ abstract class BaseSender(
     @Throws(IllegalArgumentException::class)
     fun hasCongestion(percentUsed: Float = 20f): Boolean {
         if (percentUsed !in 0.0..100.0) throw IllegalArgumentException("the value must be in range 0 to 100")
+        return queueUsagePercent() >= percentUsed
+    }
+
+    private fun queueUsagePercent(): Float {
         val size = queue.getSize().toFloat()
         val remaining = queue.remainingCapacity().toFloat()
         val capacity = size + remaining
-        return size >= capacity * (percentUsed / 100f)
+        return if (capacity <= 0f) 0f else (size / capacity) * 100f
     }
 
     fun resizeCache(newSize: Int) {
@@ -132,6 +157,8 @@ abstract class BaseSender(
 
     fun getItemsInCache(): Int = queue.getSize()
 
+    fun getQueueBytesOut(): Long = queue.getTotalSize()
+
     fun clearCache() {
         queue.clear { bufferPool.release(it.data) }
     }