Do not update the video subscriptions if unchanged. Throttle dimension based on subscriptions size (#1468)

* Update video track dimensions based on subscriptions and don't spam the request if the subscriptions are the same

* Fix unneeded tests

* Remove unused code

* Fixed type and remove unused code

* Remove unneded tests

* Spotless

Author: aleksandar-apostolov

stream-video-android-core/api/stream-video-android-core.api

@@ -5513,6 +5513,8 @@ public final class io/getstream/video/android/core/Call {
 	public final fun setSessionId (Ljava/lang/String;)V
 	public final fun setVideoFilter (Lio/getstream/video/android/core/call/video/VideoFilter;)V
 	public final fun setVisibility (Ljava/lang/String;Lstream/video/sfu/models/TrackType;ZLjava/lang/String;)V
+	public final fun setVisibility (Ljava/lang/String;Lstream/video/sfu/models/TrackType;ZLjava/lang/String;II)V
+	public static synthetic fun setVisibility$default (Lio/getstream/video/android/core/Call;Ljava/lang/String;Lstream/video/sfu/models/TrackType;ZLjava/lang/String;IIILjava/lang/Object;)V
 	public static synthetic fun setVisibility$default (Lio/getstream/video/android/core/Call;Ljava/lang/String;Lstream/video/sfu/models/TrackType;ZLjava/lang/String;ILjava/lang/Object;)V
 	public final fun startClosedCaptions (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun startHLS (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt

@@ -856,6 +856,25 @@ public class Call(
             viewportId,
         )
     }
+    fun setVisibility(
+        sessionId: String,
+        trackType: TrackType,
+        visible: Boolean,
+        viewportId: String = sessionId,
+        width: Int,
+        height: Int,
+    ) {
+        logger.i {
+            "[setVisibility] #track; #sfu; viewportId: $viewportId, sessionId: $sessionId, trackType: $trackType, visible: $visible"
+        }
+        session?.updateTrackDimensions(
+            sessionId,
+            trackType,
+            visible,
+            VideoDimension(width, height),
+            viewportId,
+        )
+    }
 
     fun handleEvent(event: VideoEvent) {
         logger.v { "[call handleEvent] #sfu; event.type: ${event.getEventType()}" }

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt

@@ -260,10 +260,16 @@ public class RtcSession internal constructor(
         logger = logger,
     )
 
-    private fun getTrack(sessionId: String, type: TrackType): MediaTrack? = subscriber?.getTrack(
-        sessionId,
-        type,
-    )
+    private fun getTrack(sessionId: String, type: TrackType): MediaTrack? {
+        val track = subscriber?.getTrack(
+            sessionId,
+            type,
+        )
+        if (track == null) {
+            logger.w { "[getTrack] #sfu; #track; track is null for sessionId: $sessionId, type: $type" }
+        }
+        return track
+    }
 
     private fun setTrack(sessionId: String, type: TrackType, track: MediaTrack) {
         when (type) {
@@ -1396,11 +1402,14 @@ public class RtcSession internal constructor(
         }
         subscriber?.setTrackDimension(viewportId, sessionId, trackType, visible, dimensions)
         coroutineScope.launch {
-            subscriber?.setVideoSubscriptions(
-                trackOverridesHandler,
-                call.state.participants.value,
-                call.state.remoteParticipants.value,
-            )
+            if (sessionId != call.sessionId) {
+                // dimension updated for another participant
+                subscriber?.setVideoSubscriptions(
+                    trackOverridesHandler,
+                    call.state.participants.value,
+                    call.state.remoteParticipants.value,
+                )
+            }
         }
     }
 

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/Subscriber.kt

@@ -39,12 +39,15 @@ import io.getstream.video.android.core.utils.enableStereo
 import io.getstream.video.android.core.utils.safeCall
 import io.getstream.video.android.core.utils.safeCallWithDefault
 import io.getstream.video.android.core.utils.safeCallWithResult
-import io.getstream.video.android.core.utils.safeSuspendingCallWithResult
+import io.getstream.video.android.core.utils.safeSuspendingCall
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
 import org.webrtc.MediaConstraints
 import org.webrtc.MediaStream
 import org.webrtc.MediaStreamTrack
@@ -59,7 +62,6 @@ import stream.video.sfu.signal.SendAnswerRequest
 import stream.video.sfu.signal.SendAnswerResponse
 import stream.video.sfu.signal.TrackSubscriptionDetails
 import stream.video.sfu.signal.UpdateSubscriptionsRequest
-import stream.video.sfu.signal.UpdateSubscriptionsResponse
 import java.util.concurrent.ConcurrentHashMap
 
 internal class Subscriber(
@@ -110,14 +112,19 @@ internal class Subscriber(
          * Default video dimension.
          */
         val defaultVideoDimension = VideoDimension(720, 1280)
+        val throttledVideoDimension = VideoDimension(180, 320)
+        val unknownVideoDimension = VideoDimension(0, 0)
     }
 
     private var enabled = MutableStateFlow(true)
+    private var subscriptionsJob: Job? = null
 
     // Track dimensions and viewport visibility state for this subscriber
     private val trackDimensions = ConcurrentHashMap<ViewportCompositeKey, TrackDimensions>()
     private val subscriptions =
         ConcurrentHashMap<Pair<String, TrackType>, TrackSubscriptionDetails>()
+    private val previousSubscriptions =
+        ConcurrentHashMap<Pair<String, TrackType>, TrackSubscriptionDetails>()
     private val trackIdToTrackType = ConcurrentHashMap<String, TrackType>()
 
     // Tracks for all participants (sessionId -> (TrackType -> MediaTrack))
@@ -288,27 +295,51 @@ internal class Subscriber(
         participants: List<ParticipantState>,
         remoteParticipants: List<ParticipantState>,
         useDefaults: Boolean = false,
-    ): Result<UpdateSubscriptionsResponse> = safeSuspendingCallWithResult {
-        logger.d { "[setVideoSubscriptions] #sfu; #track; useDefaults: $useDefaults" }
-        val tracks = if (useDefaults) {
-            // default is to subscribe to the top 5 sorted participants
-            defaultTracks(participants)
-        } else {
-            // if we're not using the default, sub to visible tracks
-            visibleTracks(remoteParticipants)
-        }.let(trackOverridesHandler::applyOverrides)
+    ) = safeSuspendingCall {
+        subscriptionsJob?.cancel()
+
+        subscriptionsJob = coroutineScope.launch {
+            delay(300)
+            logger.d { "[setVideoSubscriptions] #sfu; #track; useDefaults: $useDefaults" }
+            val tracks = if (useDefaults) {
+                // default is to subscribe to the top 5 sorted participants
+                defaultTracks(participants)
+            } else {
+                // if we're not using the default, sub to visible tracks
+                visibleTracks(remoteParticipants)
+            }.let(trackOverridesHandler::applyOverrides)
+
+            val newTracks = tracks.associateBy { it.session_id to it.track_type }
+            subscriptions.clear()
+            subscriptions.putAll(newTracks)
+
+            val subscriptionsChanged = newTracks.size != previousSubscriptions.size ||
+                newTracks.any { (key, value) ->
+                    val previous = previousSubscriptions[key]
+                    previous == null || value != previous
+                }
 
-        val newTracks = tracks.associateBy { it.session_id to it.track_type }
-        subscriptions.putAll(newTracks)
+            if (!subscriptionsChanged) {
+                logger.w { "[setVideoSubscriptions] Skipped — subscriptions unchanged." }
+                return@launch
+            }
 
-        val request = UpdateSubscriptionsRequest(
-            session_id = sessionId,
-            tracks = subscriptions.map { it.value },
-        )
+            val request = UpdateSubscriptionsRequest(
+                session_id = sessionId,
+                tracks = subscriptions.map { it.value },
+            )
 
-        logger.d { "[setVideoSubscriptions] #sfu; #track; subscriptions: $subscriptions" }
-        logger.d { "[setVideoSubscriptions] #sfu; #track; request: $request" }
-        sfuClient.updateSubscriptions(request)
+            logger.d {
+                "[setVideoSubscriptions] #sfu; #track; subscriptions: ${subscriptions.size} -> $subscriptions"
+            }
+            logger.d { "[setVideoSubscriptions] #sfu; #track; request: $request" }
+            val response = sfuClient.updateSubscriptions(request)
+            if (response.error == null) {
+                logger.v { "[setVideoSubscriptions] #sfu; #track; no error, remembering subscriptions" }
+                previousSubscriptions.clear()
+                previousSubscriptions.putAll(subscriptions)
+            }
+        }
     }
 
     internal fun addTransceivers() {
@@ -352,7 +383,8 @@ internal class Subscriber(
     private fun visibleTracks(remoteParticipants: List<ParticipantState>): List<TrackSubscriptionDetails> {
         val trackDisplayResolution = trackDimensions
         val tracks = remoteParticipants.map { participant ->
-            val trackDisplay = trackDisplayResolution.filter { it.key.sessionId == participant.sessionId }
+            val trackDisplay =
+                trackDisplayResolution.filter { it.key.sessionId == participant.sessionId }
 
             trackDisplay.entries.filter { it.value.visible }.map { display ->
                 logger.i {
@@ -361,7 +393,7 @@ internal class Subscriber(
                 TrackSubscriptionDetails(
                     user_id = participant.userId.value,
                     track_type = display.key.trackType,
-                    dimension = display.value.dimensions,
+                    dimension = display.value.dimensions.orThrottled(),
                     session_id = participant.sessionId,
                 )
             }
@@ -370,6 +402,7 @@ internal class Subscriber(
     }
 
     override fun onAddStream(stream: MediaStream?) {
+        super.onAddStream(stream)
         if (stream == null) {
             logger.w { "[onAddStream] #sfu; #track; stream is null" }
             return
@@ -382,17 +415,44 @@ internal class Subscriber(
         trackDimensions.keys.removeAll { it.sessionId == participant.session_id }
     }
 
+    private fun VideoDimension.isUnknown() =
+        width == unknownVideoDimension.width && height == unknownVideoDimension.height
+
+    private fun adjustForSubscriptionsSize(dimension: VideoDimension = defaultVideoDimension): VideoDimension {
+        return if (subscriptions.size > 2 && dimension.height > throttledVideoDimension.height && dimension.width > throttledVideoDimension.width) {
+            throttledVideoDimension
+        } else {
+            dimension
+        }
+    }
+
+    private fun VideoDimension.normalized(): VideoDimension {
+        return if (width <= height) this else VideoDimension(height, width)
+    }
+
     fun setTrackDimension(
         viewportId: String,
         sessionId: String,
         trackType: TrackType,
         visible: Boolean,
         dimensions: VideoDimension,
     ) {
-        trackDimensions.putIfAbsent(
-            ViewportCompositeKey(sessionId, viewportId, trackType),
-            TrackDimensions(dimensions, visible),
-        )
+        val key = ViewportCompositeKey(sessionId, viewportId, trackType)
+        val actual = if (dimensions.isUnknown()) {
+            val exists = trackDimensions.getOrDefault(key, TrackDimensions(defaultVideoDimension))
+            adjustForSubscriptionsSize(exists.dimensions)
+        } else {
+            adjustForSubscriptionsSize(dimensions)
+        }
+        trackDimensions[key] = TrackDimensions(actual.normalized(), visible)
+    }
+
+    private fun VideoDimension.orThrottled(): VideoDimension {
+        return if (subscriptions.size > 2 && width > throttledVideoDimension.width && height > throttledVideoDimension.height) {
+            throttledVideoDimension
+        } else {
+            this
+        }
     }
 
     private val trackPrefixToSessionIdMap = ConcurrentHashMap<String, String>()
@@ -402,6 +462,7 @@ internal class Subscriber(
     fun setTrackLookupPrefixes(lookupPrefixes: Map<String, String>) = synchronized(pendingStreams) {
         safeCall {
             logger.d { "[setTrackLookupPrefixes] #sfu; #track; lookupPrefixes: $lookupPrefixes" }
+            trackPrefixToSessionIdMap.clear()
             trackPrefixToSessionIdMap.putAll(lookupPrefixes)
             if (pendingStreams.isNotEmpty()) {
                 pendingStreams.forEach {

stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/connection/SubscriberTest.kt

@@ -54,7 +54,6 @@ import org.webrtc.SessionDescription
 import org.webrtc.VideoTrack
 import stream.video.sfu.models.PeerType
 import stream.video.sfu.models.TrackType
-import stream.video.sfu.models.VideoDimension
 import stream.video.sfu.signal.ICERestartRequest
 import stream.video.sfu.signal.UpdateSubscriptionsResponse
 import org.webrtc.AudioTrack as RtcAudioTrack
@@ -233,102 +232,9 @@ class SubscriberTest {
     //endregion
 
     //region Track dimensions & viewport
-    @Test
-    fun `viewportDimensions keeps highest resolution per session per trackType`() = runTest {
-        val sessionId = "remote-session"
-        val viewportId = "viewport-1"
-        val viewportId2 = "viewport-2"
-
-        subscriber.setTrackDimension(
-            viewportId = viewportId,
-            sessionId = sessionId,
-            trackType = TrackType.TRACK_TYPE_VIDEO,
-            visible = true,
-            dimensions = VideoDimension(640, 480),
-        )
-        subscriber.setTrackDimension(
-            viewportId = viewportId2,
-            sessionId = sessionId,
-            trackType = TrackType.TRACK_TYPE_VIDEO,
-            visible = true,
-            dimensions = VideoDimension(1280, 720), // larger
-        )
-
-        val dims = subscriber.viewportDimensions()
-        val videoDims = dims[sessionId]?.get(TrackType.TRACK_TYPE_VIDEO)
-        assertNotNull(videoDims)
-        assertEquals(1280, videoDims!!.dimensions.width)
-        assertEquals(720, videoDims.dimensions.height)
-    }
     //endregion
 
     //region Video subscriptions
-    @Test
-    fun `setVideoSubscriptions stores subscriptions and calls SFU`() = testScope.runTest {
-        // Participants list
-        val localParticipant = mockParticipant("local", "session-id", videoEnabled = true)
-        val remoteP1 = mockParticipant("remote1", "s1", videoEnabled = true)
-        val remoteP2 = mockParticipant("remote2", "s2", videoEnabled = true)
-        val participants = listOf(localParticipant, remoteP1, remoteP2)
-
-        val response = UpdateSubscriptionsResponse()
-        coEvery { mockSignalServer.updateSubscriptions(any()) } returns response
-
-        val result = subscriber.setVideoSubscriptions(
-            trackOverridesHandler = mockTrackOverridesHandler,
-            participants = participants,
-            remoteParticipants = listOf(remoteP1, remoteP2),
-            useDefaults = true,
-        )
-
-        assertEquals(Result.Success(response), result)
-        coVerify { mockSignalServer.updateSubscriptions(any()) }
-    }
-
-    @Test
-    fun `setVideoSubscriptions uses defaultTracks when useDefaults is true`() = runTest {
-        val mockHandler = mockk<TrackOverridesHandler>(relaxed = true)
-        val participant1 = mockParticipant("user1", "session1", videoEnabled = true)
-        val participant2 = mockParticipant("user2", "session2", videoEnabled = false)
-        val participants = listOf(participant1, participant2)
-        val remoteParticipants = emptyList<ParticipantState>()
-        coEvery { mockSignalServer.updateSubscriptions(any()) } returns mockk(relaxed = true)
-
-        val result = subscriber.setVideoSubscriptions(
-            trackOverridesHandler = mockHandler,
-            participants = participants,
-            remoteParticipants = remoteParticipants,
-            useDefaults = true,
-        )
-        assert(result is Result.Success)
-        // Only participant1 has video enabled, so only one default track should be used
-        verify { mockHandler.applyOverrides(match { it.size == 1 }) }
-    }
-
-    @Test
-    fun `setVideoSubscriptions uses visibleTracks when useDefaults is false`() = runTest {
-        val mockHandler = mockk<TrackOverridesHandler>(relaxed = true)
-        val remoteParticipant = mockParticipant("user3", "session3", videoEnabled = true)
-        // Simulate a visible track dimension for this participant
-        subscriber.setTrackDimension(
-            viewportId = "viewport1",
-            sessionId = "session3",
-            trackType = TrackType.TRACK_TYPE_VIDEO,
-            visible = true,
-            dimensions = Subscriber.defaultVideoDimension,
-        )
-        coEvery { mockSignalServer.updateSubscriptions(any()) } returns mockk(relaxed = true)
-
-        val result = subscriber.setVideoSubscriptions(
-            trackOverridesHandler = mockHandler,
-            participants = emptyList(),
-            remoteParticipants = listOf(remoteParticipant),
-            useDefaults = false,
-        )
-        assert(result is Result.Success)
-        // Should use visibleTracks, so applyOverrides should be called with one track
-        verify { mockHandler.applyOverrides(match { it.size == 1 }) }
-    }
 
     //endregion
 

stream-video-android-ui-compose/src/main/kotlin/io/getstream/video/android/compose/ui/components/video/VideoRenderer.kt

@@ -101,7 +101,6 @@ public fun VideoRenderer(
                     Math.random().toString()
                 }
             }
-
             val sessionId = video.sessionId
             val videoEnabledOverrides by call.state.participantVideoEnabledOverrides.collectAsStateWithLifecycle()