Implement Track getStats method (#227)

* Implement Track getStats method

* Fix tests

@@ -70,12 +70,14 @@ internal constructor(
                         listener?.onEngineReconnected()
                     }
                 }
+
                 ConnectionState.DISCONNECTED -> {
                     LKLog.d { "primary ICE disconnected" }
                     if (oldVal == ConnectionState.CONNECTED) {
                         reconnect()
                     }
                 }
+
                 else -> {
                 }
             }
@@ -588,6 +590,7 @@ internal constructor(
                     null
                 }
             }
+
             is Either.Right -> {
                 if (serverResponse.value.hasClientConfiguration()) {
                     serverResponse.value.clientConfiguration
@@ -612,7 +615,7 @@ internal constructor(
         fun onEngineDisconnected(reason: DisconnectReason)
         fun onFailToConnect(error: Throwable)
         fun onJoinResponse(response: JoinResponse)
-        fun onAddTrack(track: MediaStreamTrack, streams: Array<out MediaStream>)
+        fun onAddTrack(receiver: RtpReceiver, track: MediaStreamTrack, streams: Array<out MediaStream>)
         fun onUpdateParticipants(updates: List<LivekitModels.ParticipantInfo>)
         fun onActiveSpeakersUpdate(speakers: List<LivekitModels.SpeakerInfo>)
         fun onRemoteMuteChanged(trackSid: String, muted: Boolean)
@@ -654,6 +657,7 @@ internal constructor(
                 is Either.Left -> {
                     // do nothing.
                 }
+
                 is Either.Right -> {
                     LKLog.e { "error setting remote description for answer: ${outcome.value} " }
                 }
@@ -671,6 +675,7 @@ internal constructor(
                         LKLog.e { "error setting remote description for answer: ${outcome.value} " }
                         return@launch
                     }
+
                     else -> {}
                 }
             }
@@ -691,6 +696,7 @@ internal constructor(
                         LKLog.e { "error setting local description for answer: ${outcome.value}" }
                         return@launch
                     }
+
                     else -> {}
                 }
             }
@@ -709,6 +715,7 @@ internal constructor(
                     LKLog.w { "received candidate for publisher when we don't have one. ignoring." }
                 }
             }
+
             LivekitRtc.SignalTarget.SUBSCRIBER -> subscriber.addIceCandidate(candidate)
             else -> LKLog.i { "unknown ice candidate target?" }
         }
@@ -814,9 +821,11 @@ internal constructor(
             LivekitModels.DataPacket.ValueCase.SPEAKER -> {
                 listener?.onActiveSpeakersUpdate(dp.speaker.speakersList)
             }
+
             LivekitModels.DataPacket.ValueCase.USER -> {
                 listener?.onUserPacket(dp.user, dp.kind)
             }
+
             LivekitModels.DataPacket.ValueCase.VALUE_NOT_SET,
             null -> {
                 LKLog.v { "invalid value for data packet" }

@@ -25,6 +25,7 @@ import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.flowDelegate
 import io.livekit.android.util.invoke
+import io.livekit.android.webrtc.createStatsGetter
 import io.livekit.android.webrtc.getFilteredStats
 import kotlinx.coroutines.*
 import livekit.LivekitModels
@@ -202,6 +203,7 @@ constructor(
                             participant = it.participant,
                         )
                     )
+
                     is ParticipantEvent.ParticipantPermissionsChanged -> emitWhenConnected(
                         RoomEvent.ParticipantPermissionsChanged(
                             room = this@Room,
@@ -210,6 +212,7 @@ constructor(
                             oldPermissions = it.oldPermissions,
                         )
                     )
+
                     is ParticipantEvent.MetadataChanged -> {
                         listener?.onMetadataChanged(it.participant, it.prevMetadata, this@Room)
                         emitWhenConnected(
@@ -220,6 +223,7 @@ constructor(
                             )
                         )
                     }
+
                     is ParticipantEvent.NameChanged -> {
                         emitWhenConnected(
                             RoomEvent.ParticipantNameChanged(
@@ -229,6 +233,7 @@ constructor(
                             )
                         )
                     }
+
                     else -> {
                         /* do nothing */
                     }
@@ -355,6 +360,7 @@ constructor(
                             )
                         }
                     }
+
                     is ParticipantEvent.TrackStreamStateChanged -> eventBus.postEvent(
                         RoomEvent.TrackStreamStateChanged(
                             this@Room,
@@ -362,6 +368,7 @@ constructor(
                             it.streamState
                         )
                     )
+
                     is ParticipantEvent.TrackSubscriptionPermissionChanged -> eventBus.postEvent(
                         RoomEvent.TrackSubscriptionPermissionChanged(
                             this@Room,
@@ -370,6 +377,7 @@ constructor(
                             it.subscriptionAllowed
                         )
                     )
+
                     is ParticipantEvent.MetadataChanged -> {
                         listener?.onMetadataChanged(it.participant, it.prevMetadata, this@Room)
                         emitWhenConnected(
@@ -380,6 +388,7 @@ constructor(
                             )
                         )
                     }
+
                     is ParticipantEvent.NameChanged -> {
                         emitWhenConnected(
                             RoomEvent.ParticipantNameChanged(
@@ -389,6 +398,7 @@ constructor(
                             )
                         )
                     }
+
                     is ParticipantEvent.ParticipantPermissionsChanged -> eventBus.postEvent(
                         RoomEvent.ParticipantPermissionsChanged(
                             room = this@Room,
@@ -397,6 +407,7 @@ constructor(
                             oldPermissions = it.oldPermissions,
                         )
                     )
+
                     else -> {
                         /* do nothing */
                     }
@@ -641,7 +652,7 @@ constructor(
     /**
      * @suppress
      */
-    override fun onAddTrack(track: MediaStreamTrack, streams: Array<out MediaStream>) {
+    override fun onAddTrack(receiver: RtpReceiver, track: MediaStreamTrack, streams: Array<out MediaStream>) {
         if (streams.count() < 0) {
             LKLog.i { "add track with empty streams?" }
             return
@@ -652,7 +663,13 @@ constructor(
             trackSid = track.id()
         }
         val participant = getOrCreateRemoteParticipant(participantSid)
-        participant.addSubscribedMediaTrack(track, trackSid!!, adaptiveStream)
+        val statsGetter = createStatsGetter(engine.subscriber.peerConnection, receiver)
+        participant.addSubscribedMediaTrack(
+            track,
+            trackSid!!,
+            autoManageVideo = adaptiveStream,
+            statsGetter = statsGetter
+        )
     }
 
     /**

@@ -2,7 +2,14 @@ package io.livekit.android.room
 
 import io.livekit.android.util.LKLog
 import livekit.LivekitRtc
-import org.webrtc.*
+import org.webrtc.CandidatePairChangeEvent
+import org.webrtc.DataChannel
+import org.webrtc.IceCandidate
+import org.webrtc.MediaStream
+import org.webrtc.MediaStreamTrack
+import org.webrtc.PeerConnection
+import org.webrtc.RtpReceiver
+import org.webrtc.RtpTransceiver
 
 /**
  * @suppress
@@ -23,7 +30,7 @@ class SubscriberTransportObserver(
     override fun onAddTrack(receiver: RtpReceiver, streams: Array<out MediaStream>) {
         val track = receiver.track() ?: return
         LKLog.v { "onAddTrack: ${track.kind()}, ${track.id()}, ${streams.fold("") { sum, it -> "$sum, $it" }}" }
-        engine.listener?.onAddTrack(track, streams)
+        engine.listener?.onAddTrack(receiver, track, streams)
     }
 
     override fun onTrack(transceiver: RtpTransceiver) {

@@ -16,6 +16,7 @@ import io.livekit.android.room.RTCEngine
 import io.livekit.android.room.track.*
 import io.livekit.android.room.util.EncodingUtils
 import io.livekit.android.util.LKLog
+import io.livekit.android.webrtc.createStatsGetter
 import kotlinx.coroutines.CoroutineDispatcher
 import livekit.LivekitModels
 import livekit.LivekitRtc
@@ -180,10 +181,12 @@ internal constructor(
                         track.startCapture()
                         publishVideoTrack(track)
                     }
+
                     Track.Source.MICROPHONE -> {
                         val track = createAudioTrack()
                         publishAudioTrack(track)
                     }
+
                     Track.Source.SCREEN_SHARE -> {
                         if (mediaProjectionPermissionResultData == null) {
                             throw IllegalArgumentException("Media Projection permission result data is required to create a screen share track.")
@@ -192,6 +195,7 @@ internal constructor(
                             createScreencastTrack(mediaProjectionPermissionResultData = mediaProjectionPermissionResultData)
                         publishVideoTrack(track)
                     }
+
                     else -> {
                         LKLog.w { "Attempting to enable an unknown source, ignoring." }
                     }
@@ -307,6 +311,7 @@ internal constructor(
             return false
         }
 
+        track.statsGetter = createStatsGetter(engine.publisher.peerConnection, transceiver.sender)
 
         if (options is VideoTrackPublishOptions && options.videoCodec != null) {
             val targetCodec = options.videoCodec.lowercase()

@@ -5,6 +5,7 @@ import io.livekit.android.room.SignalClient
 import io.livekit.android.room.track.*
 import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.LKLog
+import io.livekit.android.webrtc.RTCStatsGetter
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
@@ -96,6 +97,7 @@ class RemoteParticipant(
     fun addSubscribedMediaTrack(
         mediaTrack: MediaStreamTrack,
         sid: String,
+        statsGetter: RTCStatsGetter,
         autoManageVideo: Boolean = false,
         triesLeft: Int = 20
     ) {
@@ -114,23 +116,26 @@ class RemoteParticipant(
             } else {
                 coroutineScope.launch {
                     delay(150)
-                    addSubscribedMediaTrack(mediaTrack, sid, autoManageVideo, triesLeft - 1)
+                    addSubscribedMediaTrack(mediaTrack, sid, statsGetter, autoManageVideo, triesLeft - 1)
                 }
             }
             return
         }
 
         val track: Track = when (val kind = mediaTrack.kind()) {
-            KIND_AUDIO -> AudioTrack(rtcTrack = mediaTrack as AudioTrack, name = "")
+            KIND_AUDIO -> RemoteAudioTrack(rtcTrack = mediaTrack as AudioTrack, name = "")
             KIND_VIDEO -> RemoteVideoTrack(
                 rtcTrack = mediaTrack as VideoTrack,
                 name = "",
                 autoManageVideo = autoManageVideo,
                 dispatcher = ioDispatcher
             )
+
             else -> throw TrackException.InvalidTrackTypeException("invalid track type: $kind")
         }
 
+        track.statsGetter = statsGetter
+
         publication.track = track
         publication.subscriptionAllowed = true
         track.name = publication.name

 package io.livekit.android.room.track
 
-import livekit.LivekitModels
 import org.webrtc.AudioTrack
 
-open class AudioTrack(name: String, override val rtcTrack: AudioTrack) :
+abstract class AudioTrack(name: String, override val rtcTrack: AudioTrack) :
     Track(name, Kind.AUDIO, rtcTrack) {
 }

+package io.livekit.android.room.track
+
+import org.webrtc.AudioTrack
+
+class RemoteAudioTrack(name: String, rtcTrack: AudioTrack) : io.livekit.android.room.track.AudioTrack(name, rtcTrack)

@@ -3,11 +3,15 @@ package io.livekit.android.room.track
 import io.livekit.android.events.BroadcastEventBus
 import io.livekit.android.events.TrackEvent
 import io.livekit.android.util.flowDelegate
+import io.livekit.android.webrtc.RTCStatsGetter
+import io.livekit.android.webrtc.getStats
 import livekit.LivekitModels
 import livekit.LivekitRtc
 import org.webrtc.MediaStreamTrack
+import org.webrtc.RTCStatsCollectorCallback
+import org.webrtc.RTCStatsReport
 
-open class Track(
+abstract class Track(
     name: String,
     kind: Kind,
     open val rtcTrack: MediaStreamTrack
@@ -28,6 +32,20 @@ open class Track(
     }
         internal set
 
+    var statsGetter: RTCStatsGetter? = null
+
+    /**
+     * Return the [RTCStatsReport] for this track, or null if none is available.
+     */
+    suspend fun getRTCStats(): RTCStatsReport? = statsGetter?.getStats()
+
+    /**
+     * Calls the [callback] with the [RTCStatsReport] for this track, or null if none is available.
+     */
+    fun getRTCStats(callback: RTCStatsCollectorCallback) {
+        statsGetter?.invoke(callback) ?: callback.onStatsDelivered(null)
+    }
+
     enum class Kind(val value: String) {
         AUDIO("audio"),
         VIDEO("video"),

@@ -3,7 +3,7 @@ package io.livekit.android.room.track
 import org.webrtc.VideoSink
 import org.webrtc.VideoTrack
 
-open class VideoTrack(name: String, override val rtcTrack: VideoTrack) :
+abstract class VideoTrack(name: String, override val rtcTrack: VideoTrack) :
     Track(name, Kind.VIDEO, rtcTrack) {
     protected val sinks: MutableList<VideoSink> = ArrayList();
 

 package io.livekit.android.webrtc
 
 import io.livekit.android.util.LKLog
+import kotlinx.coroutines.suspendCancellableCoroutine
 import org.webrtc.MediaStreamTrack
+import org.webrtc.PeerConnection
 import org.webrtc.RTCStats
+import org.webrtc.RTCStatsCollectorCallback
 import org.webrtc.RTCStatsReport
+import org.webrtc.RtpReceiver
+import org.webrtc.RtpSender
+import kotlin.coroutines.resume
 
 /**
  * Returns an RTCStatsReport with all the relevant information pertaining to a track.
@@ -143,4 +149,24 @@ private fun getExtraStats(
         }
     }
     return extraStats
-}
\ No newline at end of file
+}
+
+typealias RTCStatsGetter = (RTCStatsCollectorCallback) -> Unit
+
+suspend fun RTCStatsGetter.getStats(): RTCStatsReport = suspendCancellableCoroutine { cont ->
+    val listener = RTCStatsCollectorCallback { report ->
+        cont.resume(report)
+    }
+    this.invoke(listener)
+}
+
+fun createStatsGetter(peerConnection: PeerConnection, sender: RtpSender): RTCStatsGetter =
+    { statsCallback: RTCStatsCollectorCallback ->
+        peerConnection.getStats(statsCallback, sender)
+    }
+
+fun createStatsGetter(peerConnection: PeerConnection, receiver: RtpReceiver): RTCStatsGetter =
+    { statsCallback: RTCStatsCollectorCallback ->
+        peerConnection.getStats(statsCallback, receiver)
+    }
+

@@ -5,6 +5,7 @@ import org.webrtc.IceCandidate
 import org.webrtc.MediaConstraints
 import org.webrtc.MediaStream
 import org.webrtc.MediaStreamTrack
+import org.webrtc.MockRtpTransceiver
 import org.webrtc.NativePeerConnectionFactory
 import org.webrtc.PeerConnection
 import org.webrtc.RTCStatsCollectorCallback

+package io.livekit.android.mock
+
+import org.mockito.Mockito
+import org.webrtc.RtpReceiver
+
+object MockRtpReceiver {
+    fun create(): RtpReceiver {
+        return Mockito.mock(RtpReceiver::class.java)
+    }
+}
\ No newline at end of file

+package io.livekit.android.mock
+
+import org.mockito.Mockito
+import org.webrtc.RtpSender
+
+object MockRtpSender {
+    fun create(): RtpSender {
+        return Mockito.mock(RtpSender::class.java)
+    }
+}
\ No newline at end of file

@@ -9,6 +9,7 @@ import io.livekit.android.assert.assertIsClassList
 import io.livekit.android.events.*
 import io.livekit.android.mock.MockAudioStreamTrack
 import io.livekit.android.mock.MockMediaStream
+import io.livekit.android.mock.MockRtpReceiver
 import io.livekit.android.mock.TestData
 import io.livekit.android.mock.createMediaStreamId
 import io.livekit.android.room.participant.ConnectionQuality
@@ -198,7 +199,8 @@ class RoomMockE2ETest : MockE2ETest() {
         // We intentionally don't emit if the track isn't subscribed, so need to
         // add track.
         room.onAddTrack(
-            MockAudioStreamTrack(),
+            receiver = MockRtpReceiver.create(),
+            track = MockAudioStreamTrack(),
             arrayOf(
                 MockMediaStream(
                     id = createMediaStreamId(
@@ -231,6 +233,7 @@ class RoomMockE2ETest : MockE2ETest() {
             SignalClientTest.PARTICIPANT_JOIN.toOkioByteString()
         )
         room.onAddTrack(
+            MockRtpReceiver.create(),
             MockAudioStreamTrack(),
             arrayOf(
                 MockMediaStream(
@@ -317,6 +320,41 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
+    fun disconnectCleansUpParticipants() = runTest {
+        connect()
+
+        room.onUpdateParticipants(SignalClientTest.PARTICIPANT_JOIN.update.participantsList)
+        room.onAddTrack(
+            MockRtpReceiver.create(),
+            MockAudioStreamTrack(),
+            arrayOf(
+                MockMediaStream(
+                    id = createMediaStreamId(
+                        TestData.REMOTE_PARTICIPANT.sid,
+                        TestData.REMOTE_AUDIO_TRACK.sid
+                    )
+                )
+            )
+        )
+
+        val eventCollector = EventCollector(room.events, coroutineRule.scope)
+        room.onEngineDisconnected(DisconnectReason.CLIENT_INITIATED)
+        val events = eventCollector.stopCollecting()
+
+        assertIsClassList(
+            listOf(
+                RoomEvent.TrackUnsubscribed::class.java,
+                RoomEvent.TrackUnpublished::class.java,
+                RoomEvent.TrackUnpublished::class.java,
+                RoomEvent.ParticipantDisconnected::class.java,
+                RoomEvent.Disconnected::class.java
+            ),
+            events
+        )
+        Assert.assertTrue(room.remoteParticipants.isEmpty())
+    }
+
+    @Test
     fun serverDisconnectReason() = runTest {
         connect()
 

@@ -188,38 +188,4 @@ class RoomTest {
         assertNull(room.name)
         assertFalse(room.isRecording)
     }
-
-    @Test
-    fun disconnectCleansUpParticipants() = runTest {
-        connect()
-
-        room.onUpdateParticipants(SignalClientTest.PARTICIPANT_JOIN.update.participantsList)
-        room.onAddTrack(
-            MockAudioStreamTrack(),
-            arrayOf(
-                MockMediaStream(
-                    id = createMediaStreamId(
-                        TestData.REMOTE_PARTICIPANT.sid,
-                        TestData.REMOTE_AUDIO_TRACK.sid
-                    )
-                )
-            )
-        )
-
-        val eventCollector = EventCollector(room.events, coroutineRule.scope)
-        room.onEngineDisconnected(DisconnectReason.CLIENT_INITIATED)
-        val events = eventCollector.stopCollecting()
-
-        assertIsClassList(
-            listOf(
-                RoomEvent.TrackUnsubscribed::class.java,
-                RoomEvent.TrackUnpublished::class.java,
-                RoomEvent.TrackUnpublished::class.java,
-                RoomEvent.ParticipantDisconnected::class.java,
-                RoomEvent.Disconnected::class.java
-            ),
-            events
-        )
-        assertTrue(room.remoteParticipants.isEmpty())
-    }
 }
\ No newline at end of file

@@ -7,7 +7,6 @@ import io.livekit.android.util.toOkioByteString
 import io.livekit.android.util.toPBByteString
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.advanceUntilIdle
-import livekit.LivekitModels
 import livekit.LivekitModels.VideoQuality
 import livekit.LivekitRtc
 import org.junit.Assert.assertEquals
@@ -33,6 +32,7 @@ class RemoteTrackPublicationTest : MockE2ETest() {
         )
 
         room.onAddTrack(
+            MockRtpReceiver.create(),
             MockVideoStreamTrack(),
             arrayOf(
                 MockMediaStream(

-package io.livekit.android.mock
+package org.webrtc
 
+import io.livekit.android.mock.MockRtpReceiver
+import io.livekit.android.mock.MockRtpSender
 import org.mockito.Mockito
-import org.webrtc.MediaStreamTrack
-import org.webrtc.RtpTransceiver
+import org.webrtc.RtpTransceiver.RtpTransceiverDirection
 
 object MockRtpTransceiver {
     fun create(
@@ -19,6 +20,28 @@ object MockRtpTransceiver {
             }
         }
 
+        val direction = RtpTransceiverDirection.fromNativeIndex(init.directionNativeIndex)
+
+        when (direction) {
+            RtpTransceiverDirection.SEND_RECV, RtpTransceiverDirection.SEND_ONLY -> {
+                val sender = MockRtpSender.create()
+                Mockito.`when`(mock.sender)
+                    .then { sender }
+            }
+
+            else -> {}
+        }
+
+        when (direction) {
+            RtpTransceiverDirection.SEND_RECV, RtpTransceiverDirection.RECV_ONLY -> {
+                val receiver = MockRtpReceiver.create()
+                Mockito.`when`(mock.receiver)
+                    .then { receiver }
+            }
+
+            else -> {}
+        }
+
         return mock
     }
 }
\ No newline at end of file

@@ -25,7 +25,13 @@ import io.livekit.android.room.track.LocalVideoTrack
 import io.livekit.android.room.track.Track
 import io.livekit.android.sample.service.ForegroundService
 import io.livekit.android.util.flow
-import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 
 class CallViewModel(
@@ -109,6 +115,11 @@ class CallViewModel(
                             val message = it.data.toString(Charsets.UTF_8)
                             mutableDataReceived.emit("$identity: $message")
                         }
+
+                        is RoomEvent.TrackSubscribed -> {
+                            launch { collectTrackStats(it) }
+                        }
+
                         else -> {
                             Timber.e { "Room event: $it" }
                         }
@@ -130,6 +141,22 @@ class CallViewModel(
         }
     }
 
+    private suspend fun collectTrackStats(event: RoomEvent.TrackSubscribed) {
+        val pub = event.publication
+        while (true) {
+            delay(10000)
+            if (pub.subscribed) {
+                val statsReport = pub.track?.getRTCStats() ?: continue
+                Timber.e { "stats for ${pub.sid}:" }
+
+                for (entry in statsReport.statsMap) {
+                    Timber.e { "${entry.key} = ${entry.value}" }
+                }
+            }
+        }
+
+    }
+
     private suspend fun connectToRoom() {
         try {
             room.connect(