Fire TrackPublished events when participants connect (#99)

@@ -210,6 +210,13 @@ constructor(
             coroutineScope.launch {
                 lp.events.collect {
                     when (it) {
+                        is ParticipantEvent.TrackPublished -> eventBus.postEvent(
+                            RoomEvent.TrackPublished(
+                                room = this@Room,
+                                publication = it.publication,
+                                participant = it.participant,
+                            )
+                        )
                         is ParticipantEvent.ParticipantPermissionsChanged -> eventBus.postEvent(
                             RoomEvent.ParticipantPermissionsChanged(
                                 room = this@Room,
@@ -276,6 +283,17 @@ constructor(
         coroutineScope.launch {
             participant.events.collect {
                 when (it) {
+                    is ParticipantEvent.TrackPublished -> {
+                        if (state == State.CONNECTED) {
+                            eventBus.postEvent(
+                                RoomEvent.TrackPublished(
+                                    room = this@Room,
+                                    publication = it.publication,
+                                    participant = it.participant,
+                                )
+                            )
+                        }
+                    }
                     is ParticipantEvent.TrackStreamStateChanged -> eventBus.postEvent(
                         RoomEvent.TrackStreamStateChanged(
                             this@Room,
@@ -306,6 +324,10 @@ constructor(
             }
         }
 
+        if (info != null) {
+            participant.updateFromInfo(info)
+        }
+
         val newRemoteParticipants = mutableRemoteParticipants.toMutableMap()
         newRemoteParticipants[sid] = participant
         mutableRemoteParticipants = newRemoteParticipants
@@ -729,14 +751,6 @@ constructor(
     /**
      * @suppress
      */
-    override fun onTrackPublished(publication: RemoteTrackPublication, participant: RemoteParticipant) {
-        listener?.onTrackPublished(publication, participant, this)
-        eventBus.postEvent(RoomEvent.TrackPublished(this, publication, participant), coroutineScope)
-    }
-
-    /**
-     * @suppress
-     */
     override fun onTrackUnpublished(publication: RemoteTrackPublication, participant: RemoteParticipant) {
         listener?.onTrackUnpublished(publication, participant, this)
         eventBus.postEvent(RoomEvent.TrackUnpublished(this, publication, participant), coroutineScope)
@@ -745,14 +759,6 @@ constructor(
     /**
      * @suppress
      */
-    override fun onTrackPublished(publication: LocalTrackPublication, participant: LocalParticipant) {
-        listener?.onTrackPublished(publication, participant, this)
-        eventBus.postEvent(RoomEvent.TrackPublished(this, publication, participant), coroutineScope)
-    }
-
-    /**
-     * @suppress
-     */
     override fun onTrackUnpublished(publication: LocalTrackPublication, participant: LocalParticipant) {
         listener?.onTrackUnpublished(publication, participant, this)
         eventBus.postEvent(RoomEvent.TrackUnpublished(this, publication, participant), coroutineScope)

@@ -23,6 +23,10 @@ class RemoteParticipant(
     defaultDispatcher: CoroutineDispatcher,
 ) : Participant(sid, identity, defaultDispatcher) {
     /**
+     * Note: This constructor does not update all info due to event listener race conditions.
+     *
+     * Callers are responsible for calling through to [updateFromInfo] once ready.
+     *
      * @suppress
      */
     constructor(
@@ -37,7 +41,7 @@ class RemoteParticipant(
         ioDispatcher,
         defaultDispatcher
     ) {
-        updateFromInfo(info)
+        super.updateFromInfo(info)
     }
 
     private val coroutineScope = CloseableCoroutineScope(defaultDispatcher + SupervisorJob())
@@ -48,7 +52,6 @@ class RemoteParticipant(
      * @suppress
      */
     override fun updateFromInfo(info: LivekitModels.ParticipantInfo) {
-        val hadInfo = hasInfo
         super.updateFromInfo(info)
 
         val validTrackPublication = mutableMapOf<String, RemoteTrackPublication>()
@@ -74,13 +77,11 @@ class RemoteParticipant(
             validTrackPublication[trackSid] = publication
         }
 
-        if (hadInfo) {
         for (publication in newTrackPublications.values) {
             internalListener?.onTrackPublished(publication, this)
             listener?.onTrackPublished(publication, this)
             eventBus.postEvent(ParticipantEvent.TrackPublished(this, publication), scope)
         }
-        }
 
         val invalidKeys = tracks.keys - validTrackPublication.keys
         for (invalidKey in invalidKeys) {

@@ -14,6 +14,7 @@ import io.livekit.android.room.SignalClientTest
 import io.livekit.android.util.toOkioByteString
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.launch
+import livekit.LivekitRtc
 import okhttp3.Protocol
 import okhttp3.Request
 import okhttp3.Response
@@ -42,12 +43,12 @@ abstract class MockE2ETest : BaseTest() {
         wsFactory = component.websocketFactory()
     }
 
-    suspend fun connect() {
-        connectSignal()
+    suspend fun connect(joinResponse: LivekitRtc.SignalResponse = SignalClientTest.JOIN) {
+        connectSignal(joinResponse)
         connectPeerConnection()
     }
 
-    suspend fun connectSignal() {
+    suspend fun connectSignal(joinResponse: LivekitRtc.SignalResponse) {
         val job = coroutineRule.scope.launch {
             room.connect(
                 url = SignalClientTest.EXAMPLE_URL,
@@ -55,14 +56,14 @@ abstract class MockE2ETest : BaseTest() {
             )
         }
         wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
-        wsFactory.listener.onMessage(wsFactory.ws, SignalClientTest.JOIN.toOkioByteString())
+        simulateMessageFromServer(joinResponse)
 
         job.join()
     }
 
     suspend fun connectPeerConnection() {
         subscriber = component.rtcEngine().subscriber
-        wsFactory.listener.onMessage(wsFactory.ws, SignalClientTest.OFFER.toOkioByteString())
+        simulateMessageFromServer(SignalClientTest.OFFER)
         val subPeerConnection = subscriber.peerConnection as MockPeerConnection
         subPeerConnection.moveToIceConnectionState(PeerConnection.IceConnectionState.CONNECTED)
     }

@@ -20,6 +20,7 @@ object TestData {
         sid = "local_participant_sid"
         identity = "local_participant_identity"
         state = LivekitModels.ParticipantInfo.State.ACTIVE
+        metadata = "local_metadata"
         permission = LivekitModels.ParticipantPermission.newBuilder()
             .setCanPublish(true)
             .setCanSubscribe(true)
@@ -34,6 +35,14 @@ object TestData {
         sid = "remote_participant_sid"
         identity = "remote_participant_identity"
         state = LivekitModels.ParticipantInfo.State.ACTIVE
+        metadata = "remote_metadata"
+        isPublisher = true
+        permission = with(LivekitModels.ParticipantPermission.newBuilder()) {
+            canPublish = true
+            canSubscribe = true
+            canPublishData
+            build()
+        }
         addTracks(REMOTE_AUDIO_TRACK)
         build()
     }

@@ -14,6 +14,7 @@ import io.livekit.android.room.track.LocalAudioTrack
 import io.livekit.android.room.track.Track
 import io.livekit.android.util.flow
 import io.livekit.android.util.toOkioByteString
+import junit.framework.Assert.assertEquals
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.launch
 import org.junit.Assert
@@ -38,6 +39,30 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
+    fun connectNoEvents() = runTest {
+        val collector = EventCollector(room.events, coroutineRule.scope)
+        connect()
+        val events = collector.stopCollecting()
+        assertEquals(0, events.size)
+    }
+
+    @Test
+    fun connectNoEventsWithRemoteParticipant() = runTest {
+        val joinResponse = with(SignalClientTest.JOIN.toBuilder()) {
+            join = with(SignalClientTest.JOIN.join.toBuilder()) {
+                addOtherParticipants(TestData.REMOTE_PARTICIPANT)
+                build()
+            }
+            build()
+        }
+
+        val collector = EventCollector(room.events, coroutineRule.scope)
+        connect(joinResponse)
+        val events = collector.stopCollecting()
+        assertEquals(0, events.size)
+    }
+
+    @Test
     fun connectFailureProperlyContinues() = runTest {
 
         var didThrowException = false
@@ -94,14 +119,12 @@ class RoomMockE2ETest : MockE2ETest() {
         connect()
 
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
-        wsFactory.listener.onMessage(
-            wsFactory.ws,
-            SignalClientTest.PARTICIPANT_JOIN.toOkioByteString()
-        )
+        simulateMessageFromServer(SignalClientTest.PARTICIPANT_JOIN)
         val events = eventCollector.stopCollecting()
 
-        Assert.assertEquals(1, events.size)
+        Assert.assertEquals(2, events.size)
         Assert.assertEquals(true, events[0] is RoomEvent.ParticipantConnected)
+        Assert.assertEquals(true, events[1] is RoomEvent.TrackPublished)
     }
 
     @Test

 package io.livekit.android.room.participant
 
 import io.livekit.android.BaseTest
+import io.livekit.android.events.EventCollector
 import io.livekit.android.events.FlowCollector
+import io.livekit.android.events.ParticipantEvent
 import io.livekit.android.room.SignalClient
 import io.livekit.android.room.track.TrackPublication
 import io.livekit.android.util.flow
@@ -31,31 +33,20 @@ class RemoteParticipantTest : BaseTest() {
     }
 
     @Test
-    fun constructorAddsTrack() {
-        val info = LivekitModels.ParticipantInfo.newBuilder(INFO)
-            .addTracks(TRACK_INFO)
-            .build()
-
-        participant = RemoteParticipant(
-            info,
-            signalClient,
-            ioDispatcher = coroutineRule.dispatcher,
-            defaultDispatcher = coroutineRule.dispatcher,
-        )
-
-        assertEquals(1, participant.tracks.values.size)
-        assertNotNull(participant.getTrackPublication(TRACK_INFO.sid))
-    }
-
-    @Test
-    fun updateFromInfoAddsTrack() {
+    fun updateFromInfoAddsTrack() = runTest {
         val newTrackInfo = LivekitModels.ParticipantInfo.newBuilder(INFO)
             .addTracks(TRACK_INFO)
             .build()
+
+        val collector = EventCollector(participant.events, coroutineRule.scope)
         participant.updateFromInfo(newTrackInfo)
+        val events = collector.stopCollecting()
 
         assertEquals(1, participant.tracks.values.size)
         assertNotNull(participant.getTrackPublication(TRACK_INFO.sid))
+
+        val publishes = events.filterIsInstance<ParticipantEvent.TrackPublished>()
+        assertEquals(1, publishes.size)
     }
 
     @Test