Ensure that rtc engine is setup prior to any other further messages being consumed.

@@ -2,9 +2,24 @@ package io.livekit.android.dagger
 
 
 object InjectionNames {
 object InjectionNames {
 
 
+    /**
+     * @see [kotlinx.coroutines.Dispatchers.Default]
+     */
     internal const val DISPATCHER_DEFAULT = "dispatcher_default"
     internal const val DISPATCHER_DEFAULT = "dispatcher_default"
+
+    /**
+     * @see [kotlinx.coroutines.Dispatchers.IO]
+     */
     internal const val DISPATCHER_IO = "dispatcher_io";
     internal const val DISPATCHER_IO = "dispatcher_io";
+
+    /**
+     * @see [kotlinx.coroutines.Dispatchers.Main]
+     */
     internal const val DISPATCHER_MAIN = "dispatcher_main"
     internal const val DISPATCHER_MAIN = "dispatcher_main"
+
+    /**
+     * @see [kotlinx.coroutines.Dispatchers.Unconfined]
+     */
     internal const val DISPATCHER_UNCONFINED = "dispatcher_unconfined"
     internal const val DISPATCHER_UNCONFINED = "dispatcher_unconfined"
 
 
     internal const val SIGNAL_JSON_ENABLED = "signal_json_enabled"
     internal const val SIGNAL_JSON_ENABLED = "signal_json_enabled"

@@ -106,6 +106,7 @@ internal constructor(
         if (!this.isSubscriberPrimary) {
         if (!this.isSubscriberPrimary) {
             negotiate()
             negotiate()
         }
         }
+        client.onReady()
         return joinResponse
         return joinResponse
     }
     }
 
 

@@ -5,11 +5,14 @@ import io.livekit.android.ConnectOptions
 import io.livekit.android.Version
 import io.livekit.android.Version
 import io.livekit.android.dagger.InjectionNames
 import io.livekit.android.dagger.InjectionNames
 import io.livekit.android.room.track.Track
 import io.livekit.android.room.track.Track
+import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.Either
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.safe
 import io.livekit.android.util.safe
-import kotlinx.coroutines.CancellableContinuation
-import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.Json
@@ -40,6 +43,8 @@ constructor(
     private val okHttpClient: OkHttpClient,
     private val okHttpClient: OkHttpClient,
     @Named(InjectionNames.SIGNAL_JSON_ENABLED)
     @Named(InjectionNames.SIGNAL_JSON_ENABLED)
     private val useJson: Boolean,
     private val useJson: Boolean,
+    @Named(InjectionNames.DISPATCHER_IO)
+    ioDispatcher: CoroutineDispatcher,
 ) : WebSocketListener() {
 ) : WebSocketListener() {
     var isConnected = false
     var isConnected = false
         private set
         private set
@@ -49,7 +54,9 @@ constructor(
     private var lastUrl: String? = null
     private var lastUrl: String? = null
 
 
     private var joinContinuation: CancellableContinuation<Either<LivekitRtc.JoinResponse, Unit>>? = null
     private var joinContinuation: CancellableContinuation<Either<LivekitRtc.JoinResponse, Unit>>? = null
+    private val coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher)
 
 
+    private val responseFlow = MutableSharedFlow<LivekitRtc.SignalResponse>(Int.MAX_VALUE)
     suspend fun join(
     suspend fun join(
         url: String,
         url: String,
         token: String,
         token: String,
@@ -114,6 +121,15 @@ constructor(
         }
         }
     }
     }
 
 
+    @ExperimentalCoroutinesApi
+    fun onReady(){
+        coroutineScope.launch {
+            responseFlow.collect {
+                responseFlow.resetReplayCache()
+                handleSignalResponseImpl(it)
+            }
+        }
+    }
     //--------------------------------- WebSocket Listener --------------------------------------//
     //--------------------------------- WebSocket Listener --------------------------------------//
     override fun onOpen(webSocket: WebSocket, response: Response) {
     override fun onOpen(webSocket: WebSocket, response: Response) {
         if (isReconnecting) {
         if (isReconnecting) {
@@ -335,7 +351,11 @@ constructor(
             }
             }
             return
             return
         }
         }
-
+        coroutineScope.launch {
+            responseFlow.tryEmit(response)
+        }
+    }
+    private fun handleSignalResponseImpl(response: LivekitRtc.SignalResponse) {
         LKLog.v { "response: $response" }
         LKLog.v { "response: $response" }
         when (response.messageCase) {
         when (response.messageCase) {
             LivekitRtc.SignalResponse.MessageCase.ANSWER -> {
             LivekitRtc.SignalResponse.MessageCase.ANSWER -> {
@@ -386,6 +406,7 @@ constructor(
 
 
     fun close() {
     fun close() {
         isConnected = false
         isConnected = false
+        coroutineScope.close()
         currentWs?.close(1000, "Normal Closure")
         currentWs?.close(1000, "Normal Closure")
     }
     }