Properly emit upon track subscription for audio/video flows (#84)

* upgrade gradle

* Properly emit upon track subscription for audio/video flows

* Move VideoItem into SDK

@@ -2,8 +2,8 @@
 
 buildscript {
     ext {
-        compose_version = '1.1.0-rc01'
-        compose_compiler_version = '1.1.0-rc02'
+        compose_version = '1.1.1'
+        compose_compiler_version = '1.1.1'
         kotlin_version = '1.6.10'
         java_version = JavaVersion.VERSION_1_8
         dokka_version = '1.5.0'
@@ -14,7 +14,7 @@ buildscript {
         jcenter()
     }
     dependencies {
-        classpath 'com.android.tools.build:gradle:7.0.4'
+        classpath 'com.android.tools.build:gradle:7.1.2'
         classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
         classpath "org.jetbrains.kotlin:kotlin-serialization:$kotlin_version"
         classpath "org.jetbrains.dokka:dokka-gradle-plugin:$dokka_version"

 #Thu Apr 29 14:50:17 JST 2021
 distributionBase=GRADLE_USER_HOME
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.0.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.2-bin.zip
 distributionPath=wrapper/dists
 zipStorePath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME

@@ -42,6 +42,12 @@ android {
         sourceCompatibility java_version
         targetCompatibility java_version
     }
+    buildFeatures {
+        compose true
+    }
+    composeOptions {
+        kotlinCompilerExtensionVersion compose_compiler_version
+    }
     kotlinOptions {
         freeCompilerArgs = ["-Xinline-classes", "-Xopt-in=kotlin.RequiresOptIn"]
         jvmTarget = java_version

+package io.livekit.android.compose
+
+import androidx.compose.runtime.*
+import androidx.compose.ui.Modifier
+import androidx.compose.ui.layout.onGloballyPositioned
+import androidx.compose.ui.viewinterop.AndroidView
+import io.livekit.android.renderer.TextureViewRenderer
+import io.livekit.android.room.Room
+import io.livekit.android.room.track.RemoteVideoTrack
+import io.livekit.android.room.track.VideoTrack
+import io.livekit.android.room.track.video.ComposeVisibility
+
+/**
+ * Widget for displaying a VideoTrack. Handles the Compose <-> AndroidView interop needed to use
+ * [TextureViewRenderer].
+ */
+@Composable
+fun VideoRenderer(
+    room: Room,
+    videoTrack: VideoTrack,
+    modifier: Modifier = Modifier
+) {
+
+    val videoSinkVisibility = remember(room, videoTrack) { ComposeVisibility() }
+    var boundVideoTrack by remember { mutableStateOf<VideoTrack?>(null) }
+    var view: TextureViewRenderer? by remember { mutableStateOf(null) }
+
+    fun cleanupVideoTrack() {
+        view?.let { boundVideoTrack?.removeRenderer(it) }
+        boundVideoTrack = null
+    }
+
+    fun setupVideoIfNeeded(videoTrack: VideoTrack, view: TextureViewRenderer) {
+        if (boundVideoTrack == videoTrack) {
+            return
+        }
+
+        cleanupVideoTrack()
+
+        boundVideoTrack = videoTrack
+        if (videoTrack is RemoteVideoTrack) {
+            videoTrack.addRenderer(view, videoSinkVisibility)
+        } else {
+            videoTrack.addRenderer(view)
+        }
+    }
+
+    DisposableEffect(room, videoTrack) {
+        onDispose {
+            videoSinkVisibility.onDispose()
+            cleanupVideoTrack()
+        }
+    }
+
+    DisposableEffect(currentCompositeKeyHash.toString()) {
+        onDispose {
+            view?.release()
+        }
+    }
+
+    AndroidView(
+        factory = { context ->
+            TextureViewRenderer(context).apply {
+                room.initVideoRenderer(this)
+                setupVideoIfNeeded(videoTrack, this)
+
+                view = this
+            }
+        },
+        update = { v -> setupVideoIfNeeded(videoTrack, v) },
+        modifier = modifier
+            .onGloballyPositioned { videoSinkVisibility.onGloballyPositioned(it) },
+    )
+}
\ No newline at end of file

@@ -9,17 +9,13 @@ import io.livekit.android.room.track.RemoteTrackPublication
 import io.livekit.android.room.track.Track
 import io.livekit.android.room.track.TrackPublication
 import io.livekit.android.util.FlowObservable
-import io.livekit.android.util.LKLog
 import io.livekit.android.util.flow
 import io.livekit.android.util.flowDelegate
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.flow.SharingStarted
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.stateIn
+import kotlinx.coroutines.flow.*
 import livekit.LivekitModels
-import livekit.LivekitRtc
 import javax.inject.Named
 
 open class Participant(
@@ -139,6 +135,21 @@ open class Participant(
     var tracks by flowDelegate(emptyMap<String, TrackPublication>())
         protected set
 
+    private fun Flow<Map<String, TrackPublication>>.trackUpdateFlow(): Flow<List<Pair<TrackPublication, Track?>>> {
+        return flatMapLatest { videoTracks ->
+            combine(
+                videoTracks.values
+                    .map { trackPublication ->
+                        // Re-emit when track changes
+                        trackPublication::track.flow
+                            .map { trackPublication to trackPublication.track }
+                    }
+            ) { trackPubs ->
+                trackPubs.toList()
+            }
+        }
+    }
+
     /**
      * Changes can be observed by using [io.livekit.android.util.flow]
      */
@@ -147,7 +158,8 @@ open class Participant(
     val audioTracks by flowDelegate(
         stateFlow = ::tracks.flow
             .map { it.filterValues { publication -> publication.kind == Track.Kind.AUDIO } }
-            .stateIn(scope, SharingStarted.Eagerly, emptyMap())
+            .trackUpdateFlow()
+            .stateIn(scope, SharingStarted.Eagerly, emptyList())
     )
 
     /**
@@ -157,10 +169,9 @@ open class Participant(
     @get:FlowObservable
     val videoTracks by flowDelegate(
         stateFlow = ::tracks.flow
-            .map {
-                it.filterValues { publication -> publication.kind == Track.Kind.VIDEO }
-            }
-            .stateIn(scope, SharingStarted.Eagerly, emptyMap())
+            .map { it.filterValues { publication -> publication.kind == Track.Kind.VIDEO } }
+            .trackUpdateFlow()
+            .stateIn(scope, SharingStarted.Eagerly, emptyList())
     )
 
     /**

@@ -3,6 +3,7 @@ package io.livekit.android.room.track
 import android.view.View
 import io.livekit.android.dagger.InjectionNames
 import io.livekit.android.events.TrackEvent
+import io.livekit.android.room.track.video.ComposeVisibility
 import io.livekit.android.room.track.video.VideoSinkVisibility
 import io.livekit.android.room.track.video.ViewVisibility
 import io.livekit.android.util.LKLog
@@ -28,6 +29,13 @@ class RemoteVideoTrack(
     internal var lastDimensions: Dimensions = Dimensions(0, 0)
         private set
 
+    /**
+     * If `autoManageVideo` is enabled, a VideoSinkVisibility should be passed, using
+     * [ViewVisibility] if using a traditional View layout, or [ComposeVisibility]
+     * if using Jetpack Compose.
+     *
+     * By default, any Views passed to this method will be added with a [ViewVisibility].
+     */
     override fun addRenderer(renderer: VideoSink) {
         if (autoManageVideo && renderer is View) {
             addRenderer(renderer, ViewVisibility(renderer))

@@ -11,8 +11,8 @@ open class FlowCollector<T>(
     private val flow: Flow<T>,
     coroutineScope: CoroutineScope
 ) {
-    val signal = MutableStateFlow<Unit?>(null)
-    val collectEventsDeferred = coroutineScope.async {
+    private val signal = MutableStateFlow<Unit?>(null)
+    private val collectEventsDeferred = coroutineScope.async {
         flow.toListUntilSignal(signal)
     }
 

@@ -113,8 +113,8 @@ class ParticipantTest {
 
         assertEquals(1, participant.tracks.values.size)
         assertEquals(audioPublication, participant.tracks.values.first())
-        assertEquals(1, participant.audioTracks.values.size)
-        assertEquals(audioPublication, participant.audioTracks.values.first())
+        assertEquals(1, participant.audioTracks.size)
+        assertEquals(audioPublication, participant.audioTracks.first().first)
     }
 
     companion object {

 package io.livekit.android.room.participant
 
 import io.livekit.android.BaseTest
+import io.livekit.android.events.FlowCollector
 import io.livekit.android.room.SignalClient
 import io.livekit.android.room.track.TrackPublication
 import io.livekit.android.util.flow
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.launch
 import livekit.LivekitModels
 import org.junit.Assert.*
 import org.junit.Before
@@ -65,20 +65,12 @@ class RemoteParticipantTest : BaseTest() {
             .addTracks(TRACK_INFO)
             .build()
 
-        val emissions = mutableListOf<Map<String, TrackPublication>>()
-        val job = launch {
-            participant::tracks.flow.collect {
-                emissions.add(it)
-            }
-        }
-
-        assertEquals(1, emissions.size)
-        assertEquals(emptyMap<String, TrackPublication>(), emissions.first())
-
+        val collector = FlowCollector(participant::tracks.flow, coroutineRule.scope)
         participant.updateFromInfo(newTrackInfo)
 
-        job.cancel()
+        val emissions = collector.stopCollecting()
         assertEquals(2, emissions.size)
+        assertEquals(emptyMap<String, TrackPublication>(), emissions[0])
         assertEquals(1, emissions[1].size)
     }
 
@@ -89,20 +81,12 @@ class RemoteParticipantTest : BaseTest() {
             .addTracks(TRACK_INFO)
             .build()
 
-        val emissions = mutableListOf<Map<String, TrackPublication>>()
-        val job = launch {
-            participant::audioTracks.flow.collect {
-                emissions.add(it)
-            }
-        }
-
-        assertEquals(1, emissions.size)
-        assertEquals(emptyMap<String, TrackPublication>(), emissions.first())
-
+        val collector = FlowCollector(participant::tracks.flow, coroutineRule.scope)
         participant.updateFromInfo(newTrackInfo)
 
-        job.cancel()
+        val emissions = collector.stopCollecting()
         assertEquals(2, emissions.size)
+        assertEquals(emptyMap<String, TrackPublication>(), emissions[0])
         assertEquals(1, emissions[1].size)
     }
 

@@ -9,8 +9,6 @@ android {
     defaultConfig {
         minSdk androidSdk.minVersion
         targetSdk androidSdk.targetVersion
-        versionCode 1
-        versionName "1.0"
         consumerProguardFiles 'consumer-rules.pro'
 
         testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"

@@ -34,7 +34,6 @@ fun ParticipantItem(
 ) {
 
     val identity by participant::identity.flow.collectAsState()
-    val videoTracks by participant::videoTracks.flow.collectAsState()
     val audioTracks by participant::audioTracks.flow.collectAsState()
     val identityBarPadding = 4.dp
     ConstraintLayout(
@@ -86,7 +85,7 @@ fun ParticipantItem(
             },
         )
 
-        val isMuted = audioTracks.values.none { it.track != null && !it.muted }
+        val isMuted = audioTracks.none { (pub) -> pub.track != null && !pub.muted }
 
         if (isMuted) {
             Icon(

@@ -6,86 +6,16 @@ import androidx.compose.runtime.*
 import androidx.compose.ui.Alignment
 import androidx.compose.ui.Modifier
 import androidx.compose.ui.graphics.Color
-import androidx.compose.ui.layout.onGloballyPositioned
 import androidx.compose.ui.res.painterResource
-import androidx.compose.ui.viewinterop.AndroidView
-import com.github.ajalt.timberkt.Timber
-import io.livekit.android.renderer.TextureViewRenderer
+import io.livekit.android.compose.VideoRenderer
 import io.livekit.android.room.Room
 import io.livekit.android.room.participant.Participant
-import io.livekit.android.room.track.RemoteVideoTrack
 import io.livekit.android.room.track.Track
 import io.livekit.android.room.track.VideoTrack
-import io.livekit.android.room.track.video.ComposeVisibility
 import io.livekit.android.util.flow
 import kotlinx.coroutines.flow.*
 
 /**
- * Widget for displaying a VideoTrack. Handles the Compose <-> AndroidView interop needed to use
- * [TextureViewRenderer].
- */
-@Composable
-fun VideoItem(
-    room: Room,
-    videoTrack: VideoTrack,
-    modifier: Modifier = Modifier
-) {
-
-    val videoSinkVisibility = remember(room, videoTrack) { ComposeVisibility() }
-    var boundVideoTrack by remember { mutableStateOf<VideoTrack?>(null) }
-    var view: TextureViewRenderer? by remember { mutableStateOf(null) }
-
-    fun cleanupVideoTrack() {
-        view?.let { boundVideoTrack?.removeRenderer(it) }
-        boundVideoTrack = null
-    }
-
-    fun setupVideoIfNeeded(videoTrack: VideoTrack, view: TextureViewRenderer) {
-        if (boundVideoTrack == videoTrack) {
-            return
-        }
-
-        cleanupVideoTrack()
-
-        boundVideoTrack = videoTrack
-        if (videoTrack is RemoteVideoTrack) {
-            videoTrack.addRenderer(view, videoSinkVisibility)
-        } else {
-            videoTrack.addRenderer(view)
-        }
-    }
-
-    DisposableEffect(room, videoTrack) {
-        onDispose {
-            videoSinkVisibility.onDispose()
-            cleanupVideoTrack()
-        }
-    }
-
-    DisposableEffect(currentCompositeKeyHash.toString()) {
-        onDispose {
-            view?.release()
-        }
-    }
-
-    AndroidView(
-        factory = { context ->
-            TextureViewRenderer(context).apply {
-                room.initVideoRenderer(this)
-                setupVideoIfNeeded(videoTrack, this)
-
-                view = this
-            }
-        },
-        update = { view ->
-            setupVideoIfNeeded(videoTrack, view)
-        },
-        modifier = modifier
-            .onGloballyPositioned { videoSinkVisibility.onGloballyPositioned(it) },
-    )
-}
-
-/**
  * This widget primarily serves as a way to observe changes in [videoTracks].
  */
 @Composable
@@ -94,32 +24,15 @@ fun VideoItemTrackSelector(
     participant: Participant,
     modifier: Modifier = Modifier
 ) {
-
-    val subscribedVideoTracksFlow by remember(participant) {
-        mutableStateOf(
-            participant::videoTracks.flow
-                .flatMapLatest { videoTracks ->
-                    combine(
-                        videoTracks.values
-                            .map { trackPublication ->
-                                // Re-emit when track changes
-                                trackPublication::track.flow
-                                    .map { trackPublication }
-                            }
-                    ) { trackPubs ->
-                        trackPubs.toList().filter { trackPublication -> trackPublication.track != null }
-                    }
-                }
-        )
-    }
-
-    val videoTracks by subscribedVideoTracksFlow.collectAsState(initial = emptyList())
-    val videoTrack = videoTracks.firstOrNull { pub -> pub.source == Track.Source.SCREEN_SHARE }?.track as? VideoTrack
-        ?: videoTracks.firstOrNull { pub -> pub.source == Track.Source.CAMERA }?.track as? VideoTrack
-        ?: videoTracks.firstOrNull()?.track as? VideoTrack
+    val videoTrackMap by participant::videoTracks.flow.collectAsState(initial = emptyList())
+    val videoPubs = videoTrackMap.filter { (pub) -> pub.subscribed }
+        .map { (pub) -> pub }
+    val videoTrack = videoPubs.firstOrNull { pub -> pub.source == Track.Source.SCREEN_SHARE }?.track as? VideoTrack
+        ?: videoPubs.firstOrNull { pub -> pub.source == Track.Source.CAMERA }?.track as? VideoTrack
+        ?: videoPubs.firstOrNull()?.track as? VideoTrack
 
     if (videoTrack != null) {
-        VideoItem(
+        VideoRenderer(
             room = room,
             videoTrack = videoTrack,
             modifier = modifier

 package io.livekit.android.sample
 
 import android.app.Activity
-import android.content.DialogInterface
 import android.media.AudioManager
 import android.media.projection.MediaProjectionManager
 import android.os.Bundle
@@ -97,13 +96,10 @@ class CallActivity : AppCompatActivity() {
                     }
                 }.flatMapLatest { (participant, videoTracks) ->
 
-                    for (videoTrack in videoTracks.values) {
-                        Timber.e { "videoTrack is ${videoTrack.track}" }
-                    }
                     // Prioritize any screenshare streams.
                     val trackPublication = participant.getTrackPublication(Track.Source.SCREEN_SHARE)
                         ?: participant.getTrackPublication(Track.Source.CAMERA)
-                        ?: videoTracks.values.firstOrNull()
+                        ?: videoTracks.firstOrNull()?.first
                         ?: return@flatMapLatest emptyFlow()
 
                     trackPublication::track.flow

@@ -49,7 +49,7 @@ class ParticipantItem(
         coroutineScope?.launch {
             participant::audioTracks.flow
                 .flatMapLatest { tracks ->
-                    val audioTrack = tracks.values.firstOrNull()
+                    val audioTrack = tracks.firstOrNull()?.first
                     if (audioTrack != null) {
                         audioTrack::muted.flow
                     } else {

@@ -86,7 +86,7 @@ fun ParticipantItem(
             },
         )
 
-        val isMuted = audioTracks.values.none { it.track != null && !it.muted }
+        val isMuted = audioTracks.none { (pub) -> pub.track != null && !pub.muted }
 
         if (isMuted) {
             Icon(

@@ -121,29 +121,12 @@ fun VideoItemTrackSelector(
     participant: Participant,
     modifier: Modifier = Modifier
 ) {
-
-    val subscribedVideoTracksFlow by remember(participant) {
-        mutableStateOf(
-            participant::videoTracks.flow
-                .flatMapLatest { videoTracks ->
-                    combine(
-                        videoTracks.values
-                            .map { trackPublication ->
-                                // Re-emit when track changes
-                                trackPublication::track.flow
-                                    .map { trackPublication }
-                            }
-                    ) { trackPubs ->
-                        trackPubs.toList().filter { trackPublication -> trackPublication.track != null }
-                    }
-                }
-        )
-    }
-
-    val videoTracks by subscribedVideoTracksFlow.collectAsState(initial = emptyList())
-    val videoTrack = videoTracks.firstOrNull { pub -> pub.source == Track.Source.SCREEN_SHARE }?.track as? VideoTrack
-        ?: videoTracks.firstOrNull { pub -> pub.source == Track.Source.CAMERA }?.track as? VideoTrack
-        ?: videoTracks.firstOrNull()?.track as? VideoTrack
+    val videoTrackMap by participant::videoTracks.flow.collectAsState(initial = emptyList())
+    val videoPubs = videoTrackMap.filter { (pub) -> pub.subscribed }
+        .map { (pub) -> pub }
+    val videoTrack = videoPubs.firstOrNull { pub -> pub.source == Track.Source.SCREEN_SHARE }?.track as? VideoTrack
+        ?: videoPubs.firstOrNull { pub -> pub.source == Track.Source.CAMERA }?.track as? VideoTrack
+        ?: videoPubs.firstOrNull()?.track as? VideoTrack
 
     if (videoTrack != null) {
         VideoItem(