Permanent local participant to fix NPEs when not connected (#108)

* Permanent local participant to fix npe

* fix tests

@@ -135,12 +135,9 @@ constructor(
      */
     var videoTrackPublishDefaults: VideoTrackPublishDefaults by defaultsManager::videoTrackPublishDefaults
 
-    var _localParticipant: LocalParticipant? = null
-    val localParticipant: LocalParticipant
-        get() {
-            return _localParticipant
-                ?: throw UninitializedPropertyAccessException("localParticipant has not been initialized yet.")
-        }
+    val localParticipant: LocalParticipant = localParticipantFactory.create(dynacast = dynacast).apply {
+        internalListener = this@Room
+    }
 
     private var mutableRemoteParticipants by flowDelegate(emptyMap<String, RemoteParticipant>())
 
@@ -174,6 +171,44 @@ constructor(
             coroutineScope.cancel()
         }
         coroutineScope = CoroutineScope(defaultDispatcher + SupervisorJob())
+
+        // Setup local participant.
+        localParticipant.reinitialize()
+        coroutineScope.launch {
+            localParticipant.events.collect {
+                when (it) {
+                    is ParticipantEvent.TrackPublished -> emitWhenConnected(
+                        RoomEvent.TrackPublished(
+                            room = this@Room,
+                            publication = it.publication,
+                            participant = it.participant,
+                        )
+                    )
+                    is ParticipantEvent.ParticipantPermissionsChanged -> emitWhenConnected(
+                        RoomEvent.ParticipantPermissionsChanged(
+                            room = this@Room,
+                            participant = it.participant,
+                            newPermissions = it.newPermissions,
+                            oldPermissions = it.oldPermissions,
+                        )
+                    )
+                    is ParticipantEvent.MetadataChanged -> {
+                        listener?.onMetadataChanged(it.participant, it.prevMetadata, this@Room)
+                        emitWhenConnected(
+                            RoomEvent.ParticipantMetadataChanged(
+                                this@Room,
+                                it.participant,
+                                it.prevMetadata
+                            )
+                        )
+                    }
+                    else -> {
+                        /* do nothing */
+                    }
+                }
+            }
+        }
+
         state = State.CONNECTING
         connectOptions = options
         engine.join(url, token, options, getCurrentRoomOptions())
@@ -214,37 +249,7 @@ constructor(
             return
         }
 
-        if (_localParticipant == null) {
-            val lp = localParticipantFactory.create(response.participant, dynacast)
-            lp.internalListener = this
-            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,
-                                participant = it.participant,
-                                newPermissions = it.newPermissions,
-                                oldPermissions = it.oldPermissions,
-                            )
-                        )
-                        else -> {
-                            /* do nothing */
-                        }
-                    }
-                }
-            }
-            _localParticipant = lp
-        } else {
-            localParticipant.updateFromInfo(response.participant)
-        }
+        localParticipant.updateFromInfo(response.participant)
 
         if (response.otherParticipantsList.isNotEmpty()) {
             response.otherParticipantsList.forEach {
@@ -319,6 +324,16 @@ constructor(
                             it.subscriptionAllowed
                         )
                     )
+                    is ParticipantEvent.MetadataChanged -> {
+                        listener?.onMetadataChanged(it.participant, it.prevMetadata, this@Room)
+                        emitWhenConnected(
+                            RoomEvent.ParticipantMetadataChanged(
+                                this@Room,
+                                it.participant,
+                                it.prevMetadata
+                            )
+                        )
+                    }
                     is ParticipantEvent.ParticipantPermissionsChanged -> eventBus.postEvent(
                         RoomEvent.ParticipantPermissionsChanged(
                             room = this@Room,
@@ -413,7 +428,7 @@ constructor(
      * Removes all participants and tracks from the room.
      */
     private fun cleanupRoom() {
-        _localParticipant?.cleanup()
+        localParticipant.cleanup()
         remoteParticipants.keys.toMutableSet()  // copy keys to avoid concurrent modifications.
             .forEach { sid -> handleParticipantDisconnect(sid) }
     }
@@ -436,8 +451,7 @@ constructor(
 
         listener?.onDisconnect(this, null)
         listener = null
-        _localParticipant?.dispose()
-        _localParticipant = null
+        localParticipant.dispose()
 
         // Ensure all observers see the disconnected before closing scope.
         runBlocking {
@@ -742,8 +756,6 @@ constructor(
      * @suppress
      */
     override fun onMetadataChanged(participant: Participant, prevMetadata: String?) {
-        listener?.onMetadataChanged(participant, prevMetadata, this)
-        eventBus.postEvent(RoomEvent.ParticipantMetadataChanged(this, participant, prevMetadata), coroutineScope)
     }
 
     /** @suppress */
@@ -826,6 +838,11 @@ constructor(
         viewRenderer.setEnableHardwareScaler(false /* enabled */)
     }
 
+    private suspend fun emitWhenConnected(event: RoomEvent) {
+        if (state == State.CONNECTED) {
+            eventBus.postEvent(event)
+        }
+    }
 }
 
 /**

@@ -16,7 +16,6 @@ import io.livekit.android.room.track.*
 import io.livekit.android.room.util.EncodingUtils
 import io.livekit.android.util.LKLog
 import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.cancel
 import livekit.LivekitModels
 import livekit.LivekitRtc
 import org.webrtc.*
@@ -27,8 +26,6 @@ class LocalParticipant
 @AssistedInject
 internal constructor(
     @Assisted
-    info: LivekitModels.ParticipantInfo,
-    @Assisted
     private val dynacast: Boolean,
     internal val engine: RTCEngine,
     private val peerConnectionFactory: PeerConnectionFactory,
@@ -39,17 +36,13 @@ internal constructor(
     private val defaultsManager: DefaultsManager,
     @Named(InjectionNames.DISPATCHER_DEFAULT)
     coroutineDispatcher: CoroutineDispatcher,
-) : Participant(info.sid, info.identity, coroutineDispatcher) {
+) : Participant("", null, coroutineDispatcher) {
 
     var audioTrackCaptureDefaults: LocalAudioTrackOptions by defaultsManager::audioTrackCaptureDefaults
     var audioTrackPublishDefaults: AudioTrackPublishDefaults by defaultsManager::audioTrackPublishDefaults
     var videoTrackCaptureDefaults: LocalVideoTrackOptions by defaultsManager::videoTrackCaptureDefaults
     var videoTrackPublishDefaults: VideoTrackPublishDefaults by defaultsManager::videoTrackPublishDefaults
 
-    init {
-        updateFromInfo(info)
-    }
-
     private val localTrackPublications
         get() = tracks.values
             .mapNotNull { it as? LocalTrackPublication }
@@ -563,9 +556,9 @@ internal constructor(
         }
     }
 
-    fun dispose() {
+    override fun dispose() {
         cleanup()
-        scope.cancel()
+        super.dispose()
     }
 
     interface PublishListener {
@@ -575,7 +568,7 @@ internal constructor(
 
     @AssistedFactory
     interface Factory {
-        fun create(info: LivekitModels.ParticipantInfo, dynacast: Boolean): LocalParticipant
+        fun create(dynacast: Boolean): LocalParticipant
     }
 
     companion object {

@@ -11,9 +11,7 @@ import io.livekit.android.room.track.TrackPublication
 import io.livekit.android.util.FlowObservable
 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.*
 import kotlinx.coroutines.flow.*
 import livekit.LivekitModels
 import javax.inject.Named
@@ -22,9 +20,17 @@ open class Participant(
     var sid: String,
     identity: String? = null,
     @Named(InjectionNames.DISPATCHER_DEFAULT)
-    coroutineDispatcher: CoroutineDispatcher,
+    private val coroutineDispatcher: CoroutineDispatcher,
 ) {
-    protected val scope = CoroutineScope(coroutineDispatcher + SupervisorJob())
+
+    /**
+     * To only be used for flow delegate scoping, and should not be cancelled.
+     **/
+    private val delegateScope = createScope()
+    protected var scope: CoroutineScope = createScope()
+        private set
+
+    private fun createScope() = CoroutineScope(coroutineDispatcher + SupervisorJob())
 
     protected val eventBus = BroadcastEventBus<ParticipantEvent>()
     val events = eventBus.readOnly()
@@ -159,7 +165,7 @@ open class Participant(
         stateFlow = ::tracks.flow
             .map { it.filterValues { publication -> publication.kind == Track.Kind.AUDIO } }
             .trackUpdateFlow()
-            .stateIn(scope, SharingStarted.Eagerly, emptyList())
+            .stateIn(delegateScope, SharingStarted.Eagerly, emptyList())
     )
 
     /**
@@ -171,7 +177,7 @@ open class Participant(
         stateFlow = ::tracks.flow
             .map { it.filterValues { publication -> publication.kind == Track.Kind.VIDEO } }
             .trackUpdateFlow()
-            .stateIn(scope, SharingStarted.Eagerly, emptyList())
+            .stateIn(delegateScope, SharingStarted.Eagerly, emptyList())
     )
 
     /**
@@ -294,6 +300,16 @@ open class Participant(
             scope
         )
     }
+
+    internal fun reinitialize() {
+        if (!scope.isActive) {
+            scope = createScope()
+        }
+    }
+
+    internal open fun dispose() {
+        scope.cancel()
+    }
 }
 
 @Deprecated("Use Participant.events instead.")

@@ -43,7 +43,7 @@ class RoomMockE2ETest : MockE2ETest() {
         val collector = EventCollector(room.events, coroutineRule.scope)
         connect()
         val events = collector.stopCollecting()
-        assertEquals(0, events.size)
+        assertEquals(emptyList<RoomEvent>(), events)
     }
 
     @Test
@@ -59,7 +59,7 @@ class RoomMockE2ETest : MockE2ETest() {
         val collector = EventCollector(room.events, coroutineRule.scope)
         connect(joinResponse)
         val events = collector.stopCollecting()
-        assertEquals(0, events.size)
+        assertEquals(emptyList<RoomEvent>(), events)
     }
 
     @Test

@@ -6,12 +6,15 @@ import androidx.test.core.app.ApplicationProvider
 import io.livekit.android.audio.NoAudioHandler
 import io.livekit.android.coroutines.TestCoroutineRule
 import io.livekit.android.events.EventCollector
+import io.livekit.android.events.EventListenable
+import io.livekit.android.events.ParticipantEvent
 import io.livekit.android.events.RoomEvent
 import io.livekit.android.mock.*
 import io.livekit.android.room.participant.LocalParticipant
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.test.runTest
-import livekit.LivekitModels
 import org.junit.Assert
 import org.junit.Before
 import org.junit.Rule
@@ -42,8 +45,13 @@ class RoomTest {
     var eglBase: EglBase = MockEglBase()
 
     val localParticipantFactory = object : LocalParticipant.Factory {
-        override fun create(info: LivekitModels.ParticipantInfo, dynacast: Boolean): LocalParticipant {
+        override fun create(dynacast: Boolean): LocalParticipant {
             return Mockito.mock(LocalParticipant::class.java)
+                .apply {
+                    whenever(this.events).thenReturn(object : EventListenable<ParticipantEvent> {
+                        override val events: SharedFlow<ParticipantEvent> = MutableSharedFlow()
+                    })
+                }
         }
     }
 
@@ -137,14 +145,6 @@ class RoomTest {
         Assert.assertEquals(true, events[1] is RoomEvent.TrackUnpublished)
         Assert.assertEquals(true, events[2] is RoomEvent.ParticipantDisconnected)
         Assert.assertEquals(true, events[3] is RoomEvent.Disconnected)
-
-        var localParticipantEmpty = false
-        try{
-            room.localParticipant // should throw
-        } catch (e: Exception) {
-            localParticipantEmpty = true
-        }
-        Assert.assertTrue(localParticipantEmpty)
         Assert.assertTrue(room.remoteParticipants.isEmpty())
     }
 }
\ No newline at end of file