session migration (#40)

* update protocol repo * session migration * Simulate Scenario debug menu * some tests cleanup * fix reconnection implementation * fix not renegotiating publisher after migration * clean up dev url/token * bump ice connect timeout

session migration (#40)
* update protocol repo * session migration * Simulate Scenario debug menu * some tests cleanup * fix reconnection implementation * fix not renegotiating publisher after migration * clean up dev url/token * bump ice connect timeout
davidliu · GitHub
Commit eb2246d550a212bdfa30881a31938f5ae642a947 eb2246d5 1 parent f343e461
livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt
livekit-android-sdk/src/main/java/io/livekit/android/room/track/TrackPublication.kt
livekit-android-sdk/src/main/java/io/livekit/android/webrtc/PeerConnectionExt.kt
livekit-android-sdk/src/main/java/io/livekit/android/webrtc/SessionDescriptionExt.kt
livekit-android-sdk/src/test/java/io/livekit/android/ExampleUnitTest.kt
livekit-android-sdk/src/test/java/io/livekit/android/MockE2ETest.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/MockPeerConnection.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/MockPeerConnectionTransport.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebSocket.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebsocketFactory.kt → livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebSocketFactory.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestLiveKitComponent.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestRTCModule.kt
livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestWebModule.kt
livekit-android-sdk/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt
livekit-android-sdk/src/test/java/io/livekit/android/room/RoomMockE2ETest.kt
livekit-android-sdk/src/test/java/io/livekit/android/room/RoomTest.kt
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/PeerConnectionTransport.kt
查看文件 @eb2246d
@@ -11,6 +11,9 @@ import io.livekit.android.util.debounce
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import org.webrtc.*
 import javax.inject.Named
@@ -37,28 +40,36 @@ constructor(
     private var renegotiate = false
+    private val mutex = Mutex()
+
     interface Listener {
         fun onOffer(sd: SessionDescription)
     }
     fun addIceCandidate(candidate: IceCandidate) {
+        runBlocking {
+            mutex.withLock {
                 if (peerConnection.remoteDescription != null && !restartingIce) {
                     peerConnection.addIceCandidate(candidate)
                 } else {
                     pendingCandidates.add(candidate)
                 }
             }
+        }
+    }
     suspend fun setRemoteDescription(sd: SessionDescription): Either<Unit, String?> {
         val result = peerConnection.setRemoteDescription(sd)
         if (result is Either.Left) {
+            mutex.withLock {
                 pendingCandidates.forEach { pending ->
                     peerConnection.addIceCandidate(pending)
                 }
                 pendingCandidates.clear()
                 restartingIce = false
             }
+        }
         if (this.renegotiate) {
             this.renegotiate = false
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt
查看文件 @eb2246d
@@ -10,10 +10,8 @@ import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.LKLog
 import io.livekit.android.webrtc.isConnected
-import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
+import io.livekit.android.webrtc.toProtoSessionDescription
+import kotlinx.coroutines.*
 import livekit.LivekitModels
 import livekit.LivekitRtc
 import org.webrtc.*
@@ -48,22 +46,25 @@ internal constructor(
             when (value) {
                 IceState.CONNECTED -> {
                     if (oldVal == IceState.DISCONNECTED) {
-                        LKLog.d { "publisher ICE connected" }
+                        LKLog.d { "primary ICE connected" }
                         listener?.onIceConnected()
                     } else if (oldVal == IceState.RECONNECTING) {
-                        LKLog.d { "publisher ICE reconnected" }
+                        LKLog.d { "primary ICE reconnected" }
                         listener?.onIceReconnected()
                     }
                 }
                 IceState.DISCONNECTED -> {
-                    LKLog.d { "publisher ICE disconnected" }
-                    listener?.onDisconnect("Peer connection disconnected")
+                    LKLog.d { "primary ICE disconnected" }
+                    if (oldVal == IceState.CONNECTED) {
+                        reconnect()
+                    }
                 }
                 else -> {
                 }
             }
         }
-    private var wsRetries: Int = 0
+
+    private var reconnectingJob: Job? = null
     private val pendingTrackResolvers: MutableMap<String, Continuation<LivekitModels.TrackInfo>> =
         mutableMapOf()
     private var sessionUrl: String? = null
@@ -72,7 +73,7 @@ internal constructor(
     private val publisherObserver = PublisherTransportObserver(this, client)
     private val subscriberObserver = SubscriberTransportObserver(this, client)
     internal lateinit var publisher: PeerConnectionTransport
-    private lateinit var subscriber: PeerConnectionTransport
+    internal lateinit var subscriber: PeerConnectionTransport
     private var reliableDataChannel: DataChannel? = null
     private var reliableDataChannelSub: DataChannel? = null
     private var lossyDataChannel: DataChannel? = null
@@ -94,6 +95,7 @@ internal constructor(
         sessionToken = token
         val joinResponse = client.join(url, token, options)
         isClosed = false
+        listener?.onSignalConnected()
         isSubscriberPrimary = joinResponse.subscriberPrimary
@@ -169,12 +171,13 @@ internal constructor(
             val state =
                 newState ?: throw NullPointerException("unexpected null new state, what do?")
             LKLog.v { "onIceConnection new state: $newState" }
-            if (state == PeerConnection.IceConnectionState.CONNECTED) {
+            if (state.isConnected()) {
                 iceState = IceState.CONNECTED
-            } else if (state == PeerConnection.IceConnectionState.FAILED) {
+            } else if (state == PeerConnection.IceConnectionState.DISCONNECTED ||
+                state == PeerConnection.IceConnectionState.FAILED
+            ) {
                 // when we publish tracks, some WebRTC versions will send out disconnected events periodically
                 iceState = IceState.DISCONNECTED
-                listener?.onDisconnect("Peer connection disconnected")
             }
         }
@@ -242,6 +245,7 @@ internal constructor(
     }
     fun close() {
+        isClosed = true
         coroutineScope.close()
         publisher.close()
         subscriber.close()
@@ -252,53 +256,74 @@ internal constructor(
      * reconnect Signal and PeerConnections
      */
     internal fun reconnect() {
+        if (reconnectingJob != null) {
+            return
+        }
+        if (this.isClosed) {
+            return
+        }
         val url = sessionUrl
         val token = sessionToken
         if (url == null || token == null) {
             LKLog.w { "couldn't reconnect, no url or no token" }
             return
         }
-        if (iceState == IceState.DISCONNECTED || wsRetries >= MAX_SIGNAL_RETRIES) {
-            LKLog.w { "could not connect to signal after max attempts, giving up" }
-            close()
-            listener?.onDisconnect("could not reconnect after limit")
-            return
-        }
+        val job = coroutineScope.launch {
+            listener?.onReconnecting()
+
+            for (wsRetries in 0 until MAX_SIGNAL_RETRIES) {
                 var startDelay = wsRetries.toLong() * wsRetries * 500
                 if (startDelay > 5000) {
                     startDelay = 5000
                 }
-        coroutineScope.launch {
-            delay(startDelay)
-            if (iceState == IceState.DISCONNECTED) {
-                LKLog.e { "Ice is disconnected" }
-                return@launch
-            }
+                LKLog.i { "Reconnecting to signal, attempt ${wsRetries + 1}" }
+
+                delay(startDelay)
+                try {
                     client.reconnect(url, token)
+                } catch (e: Exception) {
+                    // ws reconnect failed, retry.
+                    continue
+                }
-            LKLog.v { "reconnected, restarting ICE" }
-            wsRetries = 0
+                LKLog.v { "ws reconnected, restarting ICE" }
+                listener?.onSignalConnected()
-            // trigger publisher reconnect
                 subscriber.prepareForIceRestart()
+                iceState = IceState.RECONNECTING
+                // trigger publisher reconnect
                 // only restart publisher if it's needed
                 if (hasPublished) {
-                publisher.negotiate(
-                    getPublisherOfferConstraints().apply {
-                        with(mandatory) {
-                            add(
-                                MediaConstraints.KeyValuePair(
-                                    MediaConstraintKeys.ICE_RESTART,
-                                    MediaConstraintKeys.TRUE
-                                )
-                            )
+                    negotiate()
                 }
+
+                // wait until ICE connected
+                val endTime = SystemClock.elapsedRealtime() + MAX_ICE_CONNECT_TIMEOUT_MS;
+                while (SystemClock.elapsedRealtime() < endTime) {
+                    if (iceState == IceState.CONNECTED) {
+                        LKLog.v { "reconnected to ICE" }
+                        break
                     }
-                )
+                    delay(100)
+                }
+
+                if (iceState == IceState.CONNECTED) {
+                    return@launch
+                }
+            }
+
+
+            listener?.onDisconnect("failed reconnecting.")
+            close()
         }
+        reconnectingJob = job
+        job.invokeOnCompletion {
+            if (reconnectingJob == job) {
+                reconnectingJob = null
+            }
         }
     }
@@ -306,6 +331,9 @@ internal constructor(
         if (!client.isConnected) {
             return
         }
+
+        hasPublished = true
+
         coroutineScope.launch {
             publisher.negotiate(getPublisherOfferConstraints())
         }
@@ -409,6 +437,8 @@ internal constructor(
         fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>)
         fun onSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate)
         fun onSubscriptionPermissionUpdate(subscriptionPermissionUpdate: LivekitRtc.SubscriptionPermissionUpdate)
+        fun onSignalConnected()
+        fun onReconnecting()
     }
     companion object {
@@ -416,7 +446,7 @@ internal constructor(
         private const val LOSSY_DATA_CHANNEL_LABEL = "_lossy"
         internal const val MAX_DATA_PACKET_SIZE = 15000
         private const val MAX_SIGNAL_RETRIES = 5
-        private const val MAX_ICE_CONNECT_TIMEOUT_MS = 5000
+        private const val MAX_ICE_CONNECT_TIMEOUT_MS = 20000
         internal val CONN_CONSTRAINTS = MediaConstraints().apply {
             with(optional) {
@@ -433,11 +463,7 @@ internal constructor(
             LKLog.i { sessionDescription.toString() }
             when (val outcome = publisher.setRemoteDescription(sessionDescription)) {
                 is Either.Left -> {
-                    // when reconnecting, ICE might not have disconnected and won't trigger
-                    // our connected callback, so we'll take a shortcut and set it to active
-                    if (iceState == IceState.RECONNECTING) {
-                        iceState = IceState.CONNECTED
-                    }
+                    // do nothing.
                 }
                 is Either.Right -> {
                     LKLog.e { "error setting remote description for answer: ${outcome.value} " }
@@ -520,9 +546,8 @@ internal constructor(
     }
     override fun onClose(reason: String, code: Int) {
-        // TODO: reconnect logic
         LKLog.i { "received close event: $reason, code: $code" }
-        listener?.onDisconnect(reason)
+        reconnect()
     }
     override fun onRemoteMuteChanged(trackSid: String, muted: Boolean) {
@@ -537,14 +562,17 @@ internal constructor(
         listener?.onConnectionQuality(updates)
     }
-    override fun onLeave() {
+    override fun onLeave(leave: LivekitRtc.LeaveRequest) {
         close()
-        listener?.onDisconnect("")
+        listener?.onDisconnect("server leave")
     }
+    // Signal error
     override fun onError(error: Throwable) {
+        if (isClosed) {
             listener?.onFailToConnect(error)
         }
+    }
     override fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>) {
         listener?.onStreamStateUpdate(streamStates)
@@ -584,6 +612,21 @@ internal constructor(
             }
         }
     }
+
+    fun sendSyncState(
+        subscription: LivekitRtc.UpdateSubscription,
+        publishedTracks: List<LivekitRtc.TrackPublishedResponse>
+    ) {
+        val answer = subscriber.peerConnection.localDescription.toProtoSessionDescription()
+
+        val syncState = LivekitRtc.SyncState.newBuilder()
+            .setAnswer(answer)
+            .setSubscription(subscription)
+            .addAllPublishTracks(publishedTracks)
+            .build()
+
+        client.sendSyncState(syncState)
+    }
 }
 internal enum class IceState {
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt
查看文件 @eb2246d
@@ -133,12 +133,16 @@ constructor(
         get() = mutableActiveSpeakers
     private var hasLostConnectivity: Boolean = false
+    private var connectOptions: ConnectOptions = ConnectOptions()
+
+
     suspend fun connect(url: String, token: String, options: ConnectOptions = ConnectOptions()) {
         if (this::coroutineScope.isInitialized) {
             coroutineScope.cancel()
         }
         coroutineScope = CoroutineScope(defaultDispatcher + SupervisorJob())
         state = State.CONNECTING
+        this.connectOptions = connectOptions
         val response = engine.join(url, token, options)
         LKLog.i { "Connected to server, server version: ${response.serverVersion}, client version: ${Version.CLIENT_VERSION}" }
@@ -174,6 +178,9 @@ constructor(
         }
     }
+    /**
+     * Disconnect from the room.
+     */
     fun disconnect() {
         engine.client.sendLeave()
         handleDisconnect()
@@ -306,10 +313,7 @@ constructor(
         if (state == State.RECONNECTING) {
             return
         }
-        state = State.RECONNECTING
         engine.reconnect()
-        listener?.onReconnecting(this)
-        eventBus.postEvent(RoomEvent.Reconnecting(this), coroutineScope)
     }
     private fun handleDisconnect() {
@@ -345,6 +349,43 @@ constructor(
         coroutineScope.cancel()
     }
+    fun sendSyncState() {
+        // Whether we're sending subscribed tracks or tracks to unsubscribe.
+        val sendUnsub = connectOptions.autoSubscribe
+        val participantTracksList = mutableListOf<LivekitModels.ParticipantTracks>()
+        for (participant in remoteParticipants.values) {
+            val builder = LivekitModels.ParticipantTracks.newBuilder()
+            builder.participantSid = participant.sid
+            for (trackPub in participant.tracks.values) {
+                val remoteTrackPub = (trackPub as? RemoteTrackPublication) ?: continue
+                if (remoteTrackPub.subscribed != sendUnsub) {
+                    builder.addTrackSids(remoteTrackPub.sid)
+                }
+            }
+
+            if (builder.trackSidsCount > 0) {
+                participantTracksList.add(builder.build())
+            }
+        }
+
+        val subscription = LivekitRtc.UpdateSubscription.newBuilder()
+            .setSubscribe(!sendUnsub)
+            .addAllParticipantTracks(participantTracksList)
+            .build()
+        val publishedTracks = localParticipant.publishTracksInfo()
+        engine.sendSyncState(subscription, publishedTracks)
+    }
+
+    /**
+     * Sends a simulated scenario for the server to use.
+     *
+     * To be used for internal testing purposes only.
+     * @suppress
+     */
+    fun sendSimulateScenario(scenario: LivekitRtc.SimulateScenario) {
+        engine.client.sendSimulateScenario(scenario)
+    }
+
     /**
      * @suppress
      */
@@ -388,6 +429,12 @@ constructor(
         eventBus.postEvent(RoomEvent.Reconnected(this), coroutineScope)
     }
+    override fun onReconnecting() {
+        state = State.RECONNECTING
+        listener?.onReconnecting(this)
+        eventBus.postEvent(RoomEvent.Reconnecting(this), coroutineScope)
+    }
+
     /**
      * @suppress
      */
@@ -513,6 +560,13 @@ constructor(
         eventBus.tryPostEvent(RoomEvent.FailedToConnect(this, error))
     }
+    override fun onSignalConnected() {
+        if (state == State.RECONNECTING) {
+            // during reconnection, need to send sync state upon signal connection.
+            sendSyncState()
+        }
+    }
+
     //------------------------------- ParticipantListener --------------------------------//
     /**
      * This is called for both Local and Remote participants
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt
查看文件 @eb2246d
@@ -11,6 +11,7 @@ import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.safe
+import io.livekit.android.webrtc.toProtoSessionDescription
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.collect
@@ -59,6 +60,10 @@ constructor(
     private val coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher)
     private val responseFlow = MutableSharedFlow<LivekitRtc.SignalResponse>(Int.MAX_VALUE)
+
+    /**
+     * @throws Exception if fails to connect.
+     */
     suspend fun join(
         url: String,
         token: String,
@@ -68,6 +73,9 @@ constructor(
         return (joinResponse as Either.Left).value
     }
+    /**
+     * @throws Exception if fails to connect.
+     */
     suspend fun reconnect(url: String, token: String) {
         connect(
             url,
@@ -141,7 +149,6 @@ constructor(
     }
     override fun onMessage(webSocket: WebSocket, text: String) {
-        LKLog.v { text }
         val signalResponseBuilder = LivekitRtc.SignalResponse.newBuilder()
         fromJsonProtobuf.merge(text, signalResponseBuilder)
         val response = signalResponseBuilder.build()
@@ -159,9 +166,7 @@ constructor(
     }
     override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
-        LKLog.v { "websocket closed" }
-
-        listener?.onClose(reason, code)
+        handleWebSocketClose(reason, code)
     }
     override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
@@ -193,6 +198,21 @@ constructor(
             listener?.onError(t)
             joinContinuation?.cancel(t)
         }
+
+        val wasConnected = isConnected
+        isConnected = false
+
+        if (wasConnected) {
+            handleWebSocketClose(
+                reason = reason ?: response?.toString() ?: t.localizedMessage ?: "websocket failure",
+                code = response?.code ?: 500
+            )
+        }
+    }
+
+    private fun handleWebSocketClose(reason: String, code: Int) {
+        LKLog.v { "websocket closed" }
+        listener?.onClose(reason, code)
     }
     //------------------------------- End WebSocket Listener ------------------------------------//
@@ -207,21 +227,8 @@ constructor(
         return SessionDescription(rtcSdpType, sd.sdp)
     }
-    private fun toProtoSessionDescription(sdp: SessionDescription): LivekitRtc.SessionDescription {
-        val sdBuilder = LivekitRtc.SessionDescription.newBuilder()
-        sdBuilder.sdp = sdp.description
-        sdBuilder.type = when (sdp.type) {
-            SessionDescription.Type.ANSWER -> SD_TYPE_ANSWER
-            SessionDescription.Type.OFFER -> SD_TYPE_OFFER
-            SessionDescription.Type.PRANSWER -> SD_TYPE_PRANSWER
-            else -> throw IllegalArgumentException("invalid RTC SdpType: ${sdp.type}")
-        }
-
-        return sdBuilder.build()
-    }
-
     fun sendOffer(offer: SessionDescription) {
-        val sd = toProtoSessionDescription(offer)
+        val sd = offer.toProtoSessionDescription()
         val request = LivekitRtc.SignalRequest.newBuilder()
             .setOffer(sd)
             .build()
@@ -230,7 +237,7 @@ constructor(
     }
     fun sendAnswer(answer: SessionDescription) {
-        val sd = toProtoSessionDescription(answer)
+        val sd = answer.toProtoSessionDescription()
         val request = LivekitRtc.SignalRequest.newBuilder()
             .setAnswer(sd)
             .build()
@@ -345,6 +352,22 @@ constructor(
         sendRequest(request)
     }
+    fun sendSyncState(syncState: LivekitRtc.SyncState) {
+        val request = LivekitRtc.SignalRequest.newBuilder()
+            .setSyncState(syncState)
+            .build()
+
+        sendRequest(request)
+    }
+
+    internal fun sendSimulateScenario(scenario: LivekitRtc.SimulateScenario) {
+        val request = LivekitRtc.SignalRequest.newBuilder()
+            .setSimulate(scenario)
+            .build()
+
+        sendRequest(request)
+    }
+
     fun sendLeave() {
         val request = LivekitRtc.SignalRequest.newBuilder()
             .setLeave(LivekitRtc.LeaveRequest.newBuilder().build())
@@ -373,6 +396,8 @@ constructor(
     }
     private fun handleSignalResponse(response: LivekitRtc.SignalResponse) {
+        LKLog.v { "response: $response" }
+
         if (!isConnected) {
             // Only handle joins if not connected.
             if (response.hasJoin()) {
@@ -394,7 +419,6 @@ constructor(
     }
     private fun handleSignalResponseImpl(response: LivekitRtc.SignalResponse) {
-        LKLog.v { "response: $response" }
         when (response.messageCase) {
             LivekitRtc.SignalResponse.MessageCase.ANSWER -> {
                 val sd = fromProtoSessionDescription(response.answer)
@@ -427,7 +451,7 @@ constructor(
                 LKLog.d { "received unexpected extra join message?" }
             }
             LivekitRtc.SignalResponse.MessageCase.LEAVE -> {
-                listener?.onLeave()
+                listener?.onLeave(response.leave)
             }
             LivekitRtc.SignalResponse.MessageCase.MUTE -> {
                 listener?.onRemoteMuteChanged(response.mute.sid, response.mute.muted)
@@ -475,7 +499,7 @@ constructor(
         fun onRemoteMuteChanged(trackSid: String, muted: Boolean)
         fun onRoomUpdate(update: LivekitModels.Room)
         fun onConnectionQuality(updates: List<LivekitRtc.ConnectionQualityInfo>)
-        fun onLeave()
+        fun onLeave(leave: LivekitRtc.LeaveRequest)
         fun onError(error: Throwable)
         fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>)
         fun onSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate)
@@ -486,7 +510,7 @@ constructor(
         const val SD_TYPE_ANSWER = "answer"
         const val SD_TYPE_OFFER = "offer"
         const val SD_TYPE_PRANSWER = "pranswer"
-        const val PROTOCOL_VERSION = 5
+        const val PROTOCOL_VERSION = 6
         const val SDK_TYPE = "android"
         private fun iceServer(url: String) =
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt
查看文件 @eb2246d
@@ -382,12 +382,13 @@ internal constructor(
                     height = trackHeight
                     quality = LivekitModels.VideoQuality.HIGH
                     bitrate = 0
+                    ssrc = 0
                 }.build()
             )
         } else {
-            encodings.map {
-                val scaleDownBy = it.scaleResolutionDownBy ?: 1.0
-                var videoQuality = videoQualityForRid(it.rid ?: "")
+            encodings.map { encoding ->
+                val scaleDownBy = encoding.scaleResolutionDownBy ?: 1.0
+                var videoQuality = videoQualityForRid(encoding.rid ?: "")
                 if (videoQuality == LivekitModels.VideoQuality.UNRECOGNIZED && encodings.size == 1) {
                     videoQuality = LivekitModels.VideoQuality.HIGH
                 }
@@ -395,7 +396,8 @@ internal constructor(
                     width = (trackWidth / scaleDownBy).roundToInt()
                     height = (trackHeight / scaleDownBy).roundToInt()
                     quality = videoQuality
-                    bitrate = it.maxBitrateBps ?: 0
+                    bitrate = encoding.maxBitrateBps ?: 0
+                    ssrc = 0
                 }.build()
             }
         }
@@ -587,6 +589,17 @@ internal constructor(
     }
 }
+internal fun LocalParticipant.publishTracksInfo(): List<LivekitRtc.TrackPublishedResponse> {
+    return tracks.values.mapNotNull { trackPub ->
+        val track = trackPub.track ?: return@mapNotNull null
+
+        LivekitRtc.TrackPublishedResponse.newBuilder()
+            .setCid(track.rtcTrack.id())
+            .setTrack(trackPub.trackInfo)
+            .build()
+    }
+}
+
 interface TrackPublishOptions {
     val name: String?
 }
--- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/TrackPublication.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/TrackPublication.kt
查看文件 @eb2246d
@@ -11,6 +11,7 @@ open class TrackPublication(
     track: Track?,
     participant: Participant
 ) {
+
     @get:FlowObservable
     open var track: Track? by flowDelegate(track)
         internal set
@@ -35,6 +36,8 @@ open class TrackPublication(
     var mimeType: String? = null
         internal set
+    internal var trackInfo: LivekitModels.TrackInfo? = null
+
     var participant: WeakReference<Participant>
     init {
@@ -56,5 +59,7 @@ open class TrackPublication(
             dimensions = Track.Dimensions(info.width, info.height)
         }
         mimeType = info.mimeType
+
+        trackInfo = info
     }
 }
--- a/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/PeerConnectionExt.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/PeerConnectionExt.kt
查看文件 @eb2246d
@@ -6,10 +6,13 @@ import org.webrtc.PeerConnection
  * Completed state is a valid state for a connected connection, so this should be used
  * when checking for a connected state
  */
-internal fun PeerConnection.isConnected(): Boolean {
-    return when (iceConnectionState()) {
+internal fun PeerConnection.isConnected(): Boolean = iceConnectionState().isConnected()
+
+internal fun PeerConnection.IceConnectionState.isConnected(): Boolean {
+    return when (this) {
         PeerConnection.IceConnectionState.CONNECTED,
         PeerConnection.IceConnectionState.COMPLETED -> true
         else -> false
     }
 }
+
--- a/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/SessionDescriptionExt.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/main/java/io/livekit/android/webrtc/SessionDescriptionExt.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android.webrtc
+
+import livekit.LivekitRtc
+import org.webrtc.SessionDescription
+
+fun SessionDescription.toProtoSessionDescription(): LivekitRtc.SessionDescription {
+    val sdBuilder = LivekitRtc.SessionDescription.newBuilder()
+    sdBuilder.sdp = description
+    sdBuilder.type = type.canonicalForm()
+
+    return sdBuilder.build()
+}
--- a/livekit-android-sdk/src/test/java/io/livekit/android/ExampleUnitTest.kt 已删除 100644 → 0
查看文件 @f343e46
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/ExampleUnitTest.kt 已删除 100644 → 0
查看文件 @f343e46
-package io.livekit.android
-
-import org.junit.Test
-
-import org.junit.Assert.*
-
-/**
- * Example local unit test, which will execute on the development machine (host).
- *
- * See [testing documentation](http://d.android.com/tools/testing).
- */
-class ExampleUnitTest {
-    @Test
-    fun addition_isCorrect() {
-        assertEquals(4, 2 + 2)
-    }
-}
--- a/livekit-android-sdk/src/test/java/io/livekit/android/MockE2ETest.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/MockE2ETest.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android
+
+import android.content.Context
+import androidx.test.core.app.ApplicationProvider
+import io.livekit.android.coroutines.TestCoroutineRule
+import io.livekit.android.mock.MockWebSocketFactory
+import io.livekit.android.mock.dagger.DaggerTestLiveKitComponent
+import io.livekit.android.mock.dagger.TestCoroutinesModule
+import io.livekit.android.mock.dagger.TestLiveKitComponent
+import io.livekit.android.room.Room
+import io.livekit.android.room.SignalClientTest
+import io.livekit.android.util.toOkioByteString
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runBlockingTest
+import okhttp3.Protocol
+import okhttp3.Request
+import okhttp3.Response
+import org.junit.Before
+import org.junit.Rule
+import org.mockito.junit.MockitoJUnit
+
+@ExperimentalCoroutinesApi
+abstract class MockE2ETest {
+
+    @get:Rule
+    var mockitoRule = MockitoJUnit.rule()
+
+    @get:Rule
+    var coroutineRule = TestCoroutineRule()
+
+    lateinit var component: TestLiveKitComponent
+    lateinit var context: Context
+    lateinit var room: Room
+    lateinit var wsFactory: MockWebSocketFactory
+
+    @Before
+    fun setup() {
+        context = ApplicationProvider.getApplicationContext()
+        component = DaggerTestLiveKitComponent
+            .factory()
+            .create(context, TestCoroutinesModule(coroutineRule.dispatcher))
+
+        room = component.roomFactory()
+            .create(context)
+        wsFactory = component.websocketFactory()
+    }
+
+    fun connect() {
+        val job = coroutineRule.scope.launch {
+            room.connect(
+                url = SignalClientTest.EXAMPLE_URL,
+                token = "",
+            )
+        }
+        wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
+        wsFactory.listener.onMessage(wsFactory.ws, SignalClientTest.JOIN.toOkioByteString())
+
+        // PeerTransport negotiation is on a debounce delay.
+        coroutineRule.dispatcher.advanceTimeBy(1000L)
+        runBlockingTest {
+            job.join()
+        }
+    }
+
+    fun createOpenResponse(request: Request): Response {
+        return Response.Builder()
+            .request(request)
+            .code(200)
+            .protocol(Protocol.HTTP_2)
+            .message("")
+            .build()
+    }
+}
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockPeerConnection.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockPeerConnection.kt
查看文件 @eb2246d
@@ -7,7 +7,7 @@ private class MockNativePeerConnectionFactory : NativePeerConnectionFactory {
 }
 class MockPeerConnection(
-    private val observer: PeerConnection.Observer?
+    val observer: PeerConnection.Observer?
 ) : PeerConnection(MockNativePeerConnectionFactory()) {
     var localDesc: SessionDescription? = null
@@ -33,12 +33,12 @@ class MockPeerConnection(
     }
     override fun createOffer(observer: SdpObserver?, constraints: MediaConstraints?) {
-        val sdp = SessionDescription(SessionDescription.Type.OFFER, "")
+        val sdp = SessionDescription(SessionDescription.Type.OFFER, "local_offer")
         observer?.onCreateSuccess(sdp)
     }
     override fun createAnswer(observer: SdpObserver?, constraints: MediaConstraints?) {
-        val sdp = SessionDescription(SessionDescription.Type.ANSWER, "")
+        val sdp = SessionDescription(SessionDescription.Type.ANSWER, "local_answer")
         observer?.onCreateSuccess(sdp)
     }
@@ -143,8 +143,41 @@ class MockPeerConnection(
         return super.signalingState()
     }
-    override fun iceConnectionState(): IceConnectionState {
-        return super.iceConnectionState()
+    private var iceConnectionState = IceConnectionState.NEW
+        set(value) {
+            if (field != value) {
+                field = value
+                observer?.onIceConnectionChange(field)
+            }
+        }
+
+    override fun iceConnectionState(): IceConnectionState = iceConnectionState
+
+    fun moveToIceConnectionState(newState: IceConnectionState) {
+        when (newState) {
+            IceConnectionState.NEW,
+            IceConnectionState.CHECKING,
+            IceConnectionState.CONNECTED,
+            IceConnectionState.COMPLETED -> {
+                val currentOrdinal = iceConnectionState.ordinal
+                val newOrdinal = newState.ordinal
+
+                if (currentOrdinal < newOrdinal) {
+                    // Ensure that we move through each state.
+                    for (ordinal in ((currentOrdinal + 1)..newOrdinal)) {
+                        iceConnectionState = IceConnectionState.values()[ordinal]
+                    }
+                } else {
+                    iceConnectionState = newState
+                }
+            }
+            IceConnectionState.FAILED,
+            IceConnectionState.DISCONNECTED,
+            IceConnectionState.CLOSED -> {
+                // jump to state directly.
+                iceConnectionState = newState
+            }
+        }
     }
     override fun connectionState(): PeerConnectionState {
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockPeerConnectionTransport.kt 已删除 100644 → 0
查看文件 @f343e46
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockPeerConnectionTransport.kt 已删除 100644 → 0
查看文件 @f343e46
-package io.livekit.android.mock
-
-import io.livekit.android.room.PeerConnectionTransport
-import kotlinx.coroutines.CoroutineDispatcher
-import org.webrtc.PeerConnection
-import org.webrtc.PeerConnectionFactory
-
-internal class MockPeerConnectionTransportFactory(
-    private val dispatcher: CoroutineDispatcher,
-) : PeerConnectionTransport.Factory {
-    override fun create(
-        config: PeerConnection.RTCConfiguration,
-        pcObserver: PeerConnection.Observer,
-        listener: PeerConnectionTransport.Listener?
-    ): PeerConnectionTransport {
-        return PeerConnectionTransport(
-            config,
-            pcObserver,
-            listener,
-            dispatcher,
-            PeerConnectionFactory.builder()
-                .createPeerConnectionFactory()
-        )
-    }
-}
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebSocket.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebSocket.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android.mock
+
+import okhttp3.Request
+import okhttp3.WebSocket
+import okio.ByteString
+
+class MockWebSocket(private val request: Request) : WebSocket {
+
+    var isClosed = false
+        private set
+
+    private val mutableSentRequests = mutableListOf<ByteString>()
+    val sentRequests: List<ByteString>
+        get() = mutableSentRequests
+
+    override fun cancel() {
+        isClosed = true
+    }
+
+    override fun close(code: Int, reason: String?): Boolean {
+        val willClose = !isClosed
+        isClosed = true
+        return willClose
+    }
+
+    override fun queueSize(): Long = 0
+
+    override fun request(): Request = request
+
+    override fun send(text: String): Boolean = !isClosed
+
+    override fun send(bytes: ByteString): Boolean {
+        mutableSentRequests.add(bytes)
+        return !isClosed
+    }
+
+
+}
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebsocketFactory.kt → livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebSocketFactory.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebsocketFactory.kt → livekit-android-sdk/src/test/java/io/livekit/android/mock/MockWebSocketFactory.kt
查看文件 @eb2246d
@@ -3,14 +3,25 @@ package io.livekit.android.mock
 import okhttp3.Request
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
-import org.mockito.Mockito
-class MockWebsocketFactory : WebSocket.Factory {
+class MockWebSocketFactory : WebSocket.Factory {
+    /**
+     * The most recently created [WebSocket].
+     */
     lateinit var ws: WebSocket
+
+    /**
+     * The request used to create [ws]
+     */
     lateinit var request: Request
+
+    /**
+     * The listener associated with [ws]
+     */
     lateinit var listener: WebSocketListener
     override fun newWebSocket(request: Request, listener: WebSocketListener): WebSocket {
-        this.ws = Mockito.mock(WebSocket::class.java)
+        this.ws = MockWebSocket(request)
+
         this.listener = listener
         this.request = request
         return ws
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestLiveKitComponent.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestLiveKitComponent.kt
查看文件 @eb2246d
@@ -5,7 +5,8 @@ import dagger.BindsInstance
 import dagger.Component
 import io.livekit.android.dagger.JsonFormatModule
 import io.livekit.android.dagger.LiveKitComponent
-import io.livekit.android.mock.MockWebsocketFactory
+import io.livekit.android.mock.MockWebSocketFactory
+import io.livekit.android.room.RTCEngine
 import javax.inject.Singleton
 @Singleton
@@ -19,10 +20,15 @@ import javax.inject.Singleton
 )
 internal interface TestLiveKitComponent : LiveKitComponent {
-    fun websocketFactory(): MockWebsocketFactory
+    fun websocketFactory(): MockWebSocketFactory
+
+    fun rtcEngine(): RTCEngine
     @Component.Factory
     interface Factory {
-        fun create(@BindsInstance appContext: Context, coroutinesModule: TestCoroutinesModule = TestCoroutinesModule()): TestLiveKitComponent
+        fun create(
+            @BindsInstance appContext: Context,
+            coroutinesModule: TestCoroutinesModule = TestCoroutinesModule()
+        ): TestLiveKitComponent
     }
 }
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestRTCModule.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestRTCModule.kt
查看文件 @eb2246d
@@ -28,12 +28,7 @@ object TestRTCModule {
     fun peerConnectionFactory(
         appContext: Context
     ): PeerConnectionFactory {
-        try {
-            ContextUtils.initialize(appContext)
-            NativeLibraryLoaderTestHelper.initialize()
-        } catch (e: Throwable) {
-            // do nothing. this is expected.
-        }
+        WebRTCInitializer.initialize(appContext)
         return MockPeerConnectionFactory()
     }
--- a/livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestWebModule.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/mock/dagger/TestWebModule.kt
查看文件 @eb2246d
@@ -2,7 +2,7 @@ package io.livekit.android.mock.dagger
 import dagger.Module
 import dagger.Provides
-import io.livekit.android.mock.MockWebsocketFactory
+import io.livekit.android.mock.MockWebSocketFactory
 import okhttp3.OkHttpClient
 import okhttp3.Response
 import okhttp3.WebSocket
@@ -26,13 +26,13 @@ object TestWebModule {
     @Provides
     @Singleton
-    fun websocketFactory(websocketFactory: MockWebsocketFactory): WebSocket.Factory {
-        return websocketFactory
+    fun websocketFactory(webSocketFactory: MockWebSocketFactory): WebSocket.Factory {
+        return webSocketFactory
     }
     @Provides
     @Singleton
-    fun mockWebsocketFactory(): MockWebsocketFactory {
-        return MockWebsocketFactory()
+    fun mockWebsocketFactory(): MockWebSocketFactory {
+        return MockWebSocketFactory()
     }
 }
--- a/livekit-android-sdk/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/room/RTCEngineMockE2ETest.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android.room
+
+import io.livekit.android.MockE2ETest
+import io.livekit.android.mock.MockPeerConnection
+import io.livekit.android.mock.MockWebSocket
+import io.livekit.android.util.LoggingRule
+import io.livekit.android.util.toPBByteString
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runBlockingTest
+import livekit.LivekitRtc
+import org.junit.Assert
+import org.junit.Before
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import org.webrtc.PeerConnection
+import org.webrtc.SessionDescription
+
+
+@ExperimentalCoroutinesApi
+@RunWith(RobolectricTestRunner::class)
+class RTCEngineMockE2ETest : MockE2ETest() {
+
+
+    @get:Rule
+    var loggingRule = LoggingRule()
+
+    lateinit var rtcEngine: RTCEngine
+
+    @Before
+    fun setupRTCEngine() {
+        rtcEngine = component.rtcEngine()
+    }
+
+    @Test
+    fun iceSubscriberConnect() = runBlockingTest {
+        connect()
+
+        val remoteOffer = SessionDescription(SessionDescription.Type.OFFER, "remote_offer")
+        rtcEngine.onOffer(remoteOffer)
+
+        Assert.assertEquals(remoteOffer, rtcEngine.subscriber.peerConnection.remoteDescription)
+
+        val ws = wsFactory.ws as MockWebSocket
+        val sentRequest = LivekitRtc.SignalRequest.newBuilder()
+            .mergeFrom(ws.sentRequests[0].toPBByteString())
+            .build()
+
+        val subPeerConnection = rtcEngine.subscriber.peerConnection as MockPeerConnection
+        val localAnswer = subPeerConnection.localDescription ?: throw IllegalStateException("no answer was created.")
+        Assert.assertTrue(sentRequest.hasAnswer())
+        Assert.assertEquals(localAnswer.description, sentRequest.answer.sdp)
+        Assert.assertEquals(localAnswer.type.canonicalForm(), sentRequest.answer.type)
+
+        subPeerConnection.moveToIceConnectionState(PeerConnection.IceConnectionState.CONNECTED)
+
+        Assert.assertEquals(IceState.CONNECTED, rtcEngine.iceState)
+    }
+
+    @Test
+    fun reconnectOnFailure() = runBlockingTest {
+        connect()
+        val oldWs = wsFactory.ws
+        wsFactory.listener.onFailure(oldWs, Exception(), null)
+
+        val newWs = wsFactory.ws
+        Assert.assertNotEquals(oldWs, newWs)
+    }
+}
--- a/livekit-android-sdk/src/test/java/io/livekit/android/room/RoomMockE2ETest.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/room/RoomMockE2ETest.kt
查看文件 @eb2246d
 package io.livekit.android.room
-import android.content.Context
-import androidx.test.core.app.ApplicationProvider
-import io.livekit.android.coroutines.TestCoroutineRule
+import android.net.Network
+import io.livekit.android.MockE2ETest
 import io.livekit.android.events.EventCollector
-import io.livekit.android.events.ParticipantEvent
 import io.livekit.android.events.RoomEvent
-import io.livekit.android.mock.*
-import io.livekit.android.mock.dagger.DaggerTestLiveKitComponent
-import io.livekit.android.mock.dagger.TestCoroutinesModule
+import io.livekit.android.mock.MockAudioStreamTrack
+import io.livekit.android.mock.MockMediaStream
+import io.livekit.android.mock.TestData
+import io.livekit.android.mock.createMediaStreamId
 import io.livekit.android.room.participant.ConnectionQuality
 import io.livekit.android.room.track.Track
 import io.livekit.android.util.toOkioByteString
@@ -16,54 +15,14 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runBlockingTest
 import org.junit.Assert
-import org.junit.Before
-import org.junit.Rule
 import org.junit.Test
 import org.junit.runner.RunWith
-import org.mockito.junit.MockitoJUnit
+import org.mockito.Mockito
 import org.robolectric.RobolectricTestRunner
 @ExperimentalCoroutinesApi
 @RunWith(RobolectricTestRunner::class)
-class RoomMockE2ETest {
-
-    @get:Rule
-    var mockitoRule = MockitoJUnit.rule()
-
-    @get:Rule
-    var coroutineRule = TestCoroutineRule()
-
-    lateinit var context: Context
-    lateinit var room: Room
-    lateinit var wsFactory: MockWebsocketFactory
-
-    @Before
-    fun setup() {
-        context = ApplicationProvider.getApplicationContext()
-        val component = DaggerTestLiveKitComponent
-            .factory()
-            .create(context, TestCoroutinesModule(coroutineRule.dispatcher))
-
-        room = component.roomFactory()
-            .create(context)
-        wsFactory = component.websocketFactory()
-    }
-
-    fun connect() {
-        val job = coroutineRule.scope.launch {
-            room.connect(
-                url = "http://www.example.com",
-                token = "",
-            )
-        }
-        wsFactory.listener.onMessage(wsFactory.ws, SignalClientTest.JOIN.toOkioByteString())
-
-        // PeerTransport negotiation is on a debounce delay.
-        coroutineRule.dispatcher.advanceTimeBy(1000L)
-        runBlockingTest {
-            job.join()
-        }
-    }
+class RoomMockE2ETest : MockE2ETest() {
     @Test
     fun connectTest() {
@@ -77,7 +36,7 @@ class RoomMockE2ETest {
         val job = coroutineRule.scope.launch {
             try {
                 room.connect(
-                    url = "http://www.example.com",
+                    url = SignalClientTest.EXAMPLE_URL,
                     token = "",
                 )
             } catch (e: Throwable) {
@@ -265,6 +224,19 @@ class RoomMockE2ETest {
     }
     @Test
+    fun onConnectionAvailableWillReconnect() {
+        connect()
+        val eventCollector = EventCollector(room.events, coroutineRule.scope)
+        val network = Mockito.mock(Network::class.java)
+        room.onLost(network)
+        room.onAvailable(network)
+        val events = eventCollector.stopCollecting()
+
+        Assert.assertEquals(1, events.size)
+        Assert.assertEquals(true, events[0] is RoomEvent.Reconnecting)
+    }
+
+    @Test
     fun leave() {
         connect()
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
--- a/livekit-android-sdk/src/test/java/io/livekit/android/room/RoomTest.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/room/RoomTest.kt
查看文件 @eb2246d
@@ -75,7 +75,7 @@ class RoomTest {
         }
         val job = coroutineRule.scope.launch {
             room.connect(
-                url = "http://www.example.com",
+                url = SignalClientTest.EXAMPLE_URL,
                 token = "",
             )
         }
@@ -93,15 +93,10 @@ class RoomTest {
     fun onConnectionAvailableWillReconnect() {
         connect()
-        val eventCollector = EventCollector(room.events, coroutineRule.scope)
         val network = Mockito.mock(Network::class.java)
         room.onLost(network)
         room.onAvailable(network)
-
-        val events = eventCollector.stopCollecting()
-
-        Assert.assertEquals(1, events.size)
-        Assert.assertEquals(true, events[0] is RoomEvent.Reconnecting)
+        Mockito.verify(rtcEngine).reconnect()
     }
     @Test
--- a/livekit-android-sdk/src/test/java/io/livekit/android/room/SignalClientTest.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/room/SignalClientTest.kt
查看文件 @eb2246d
 package io.livekit.android.room
 import com.google.protobuf.util.JsonFormat
-import io.livekit.android.mock.MockWebsocketFactory
+import io.livekit.android.mock.MockWebSocketFactory
 import io.livekit.android.mock.TestData
 import io.livekit.android.util.toOkioByteString
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -12,22 +12,20 @@ import kotlinx.coroutines.test.runBlockingTest
 import kotlinx.serialization.json.Json
 import livekit.LivekitModels
 import livekit.LivekitRtc
-import okhttp3.OkHttpClient
-import okhttp3.Protocol
-import okhttp3.Request
-import okhttp3.Response
+import okhttp3.*
 import org.junit.After
 import org.junit.Assert
 import org.junit.Before
 import org.junit.Test
 import org.mockito.Mockito
+import org.mockito.kotlin.any
 import org.mockito.kotlin.argThat
 import org.webrtc.SessionDescription
 @ExperimentalCoroutinesApi
 class SignalClientTest {
-    lateinit var wsFactory: MockWebsocketFactory
+    lateinit var wsFactory: MockWebSocketFactory
     lateinit var client: SignalClient
     lateinit var listener: SignalClient.Listener
     lateinit var okHttpClient: OkHttpClient
@@ -39,7 +37,7 @@ class SignalClientTest {
     fun setup() {
         coroutineDispatcher = TestCoroutineDispatcher()
         coroutineScope = TestCoroutineScope(coroutineDispatcher)
-        wsFactory = MockWebsocketFactory()
+        wsFactory = MockWebSocketFactory()
         okHttpClient = Mockito.mock(OkHttpClient::class.java)
         client = SignalClient(
             wsFactory,
@@ -68,14 +66,21 @@ class SignalClientTest {
             .build()
     }
+    /**
+     * Supply the needed websocket messages to finish a join call.
+     */
+    private fun connectWebsocketAndJoin() {
+        client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
+        client.onMessage(wsFactory.ws, JOIN.toOkioByteString())
+    }
+
     @Test
     fun joinAndResponse() {
         val job = coroutineScope.async {
             client.join(EXAMPLE_URL, "")
         }
-        client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
-        client.onMessage(wsFactory.ws, JOIN.toOkioByteString())
+        connectWebsocketAndJoin()
         runBlockingTest {
             val response = job.await()
@@ -97,15 +102,29 @@ class SignalClientTest {
     }
     @Test
-    fun listenerNotCalledUntilOnReady() {
-        val listener = Mockito.mock(SignalClient.Listener::class.java)
-        client.listener = listener
+    fun joinFailure() {
+        var failed = false
+        val job = coroutineScope.async {
+            try {
+                client.join(EXAMPLE_URL, "")
+            } catch (e: Exception) {
+                failed = true
+            }
+        }
+
+        client.onFailure(wsFactory.ws, Exception(), null)
+        runBlockingTest { job.await() }
+        Assert.assertTrue(failed)
+    }
+
+    @Test
+    fun listenerNotCalledUntilOnReady() {
         val job = coroutineScope.async {
             client.join(EXAMPLE_URL, "")
         }
-        client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
-        client.onMessage(wsFactory.ws, JOIN.toOkioByteString())
+
+        connectWebsocketAndJoin()
         client.onMessage(wsFactory.ws, OFFER.toOkioByteString())
         runBlockingTest { job.await() }
@@ -115,14 +134,10 @@ class SignalClientTest {
     @Test
     fun listenerCalledAfterOnReady() {
-        val listener = Mockito.mock(SignalClient.Listener::class.java)
-        client.listener = listener
-
         val job = coroutineScope.async {
             client.join(EXAMPLE_URL, "")
         }
-        client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
-        client.onMessage(wsFactory.ws, JOIN.toOkioByteString())
+        connectWebsocketAndJoin()
         client.onMessage(wsFactory.ws, OFFER.toOkioByteString())
         runBlockingTest { job.await() }
@@ -131,9 +146,27 @@ class SignalClientTest {
             .onOffer(argThat { type == SessionDescription.Type.OFFER && description == OFFER.offer.sdp })
     }
+    /**
+     * [WebSocketListener.onFailure] does not call through to
+     * [WebSocketListener.onClosed]. Ensure that listener is called properly.
+     */
+    @Test
+    fun listenerNotifiedAfterFailure() {
+        val job = coroutineScope.async {
+            client.join(EXAMPLE_URL, "")
+        }
+        connectWebsocketAndJoin()
+        runBlockingTest { job.await() }
+
+        client.onFailure(wsFactory.ws, Exception(), null)
+
+        Mockito.verify(listener)
+            .onClose(any(), any())
+    }
+
     // mock data
     companion object {
-        private const val EXAMPLE_URL = "http://www.example.com"
+        const val EXAMPLE_URL = "ws://www.example.com"
         val JOIN = with(LivekitRtc.SignalResponse.newBuilder()) {
             join = with(joinBuilder) {
@@ -143,6 +176,8 @@ class SignalClientTest {
                     build()
                 }
                 participant = TestData.LOCAL_PARTICIPANT
+                subscriberPrimary = true
+                serverVersion = "0.15.2"
                 build()
             }
             build()
--- a/livekit-android-sdk/src/test/java/io/livekit/android/util/ByteString.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/util/ByteString.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android.util
+
+import com.google.protobuf.ByteString
+import okio.ByteString.Companion.toByteString
+
+fun com.google.protobuf.ByteString.toOkioByteString() = toByteArray().toByteString()
+
+fun okio.ByteString.toPBByteString() = ByteString.copyFrom(toByteArray())
--- a/livekit-android-sdk/src/test/java/io/livekit/android/util/TimberLogRule.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/io/livekit/android/util/TimberLogRule.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android.util
+
+import android.util.Log
+import io.livekit.android.LiveKit
+import org.junit.rules.TestRule
+import org.junit.runner.Description
+import org.junit.runners.model.Statement
+import timber.log.Timber
+
+class LoggingRule : TestRule {
+
+    val logTree = object : Timber.Tree() {
+        override fun log(priority: Int, tag: String?, message: String, t: Throwable?) {
+            val priorityChar = when (priority) {
+                Log.VERBOSE -> "v"
+                Log.DEBUG -> "d"
+                Log.INFO -> "i"
+                Log.WARN -> "w"
+                Log.ERROR -> "e"
+                Log.ASSERT -> "a"
+                else -> "?"
+            }
+
+            println("$priorityChar: $tag: $message")
+            if (t != null) {
+                println(t.toString())
+            }
+        }
+
+    }
+
+    override fun apply(base: Statement, description: Description?) = object : Statement() {
+        override fun evaluate() {
+            val oldLoggingLevel = LiveKit.loggingLevel
+            LiveKit.loggingLevel = LoggingLevel.VERBOSE
+            Timber.plant(logTree)
+            base.evaluate()
+            Timber.uproot(logTree)
+            LiveKit.loggingLevel = oldLoggingLevel
+        }
+    }
+}
--- a/livekit-android-sdk/src/test/java/org/webrtc/NativeLibraryLoaderHelper.kt
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/org/webrtc/NativeLibraryLoaderHelper.kt
查看文件 @eb2246d
@@ -2,6 +2,8 @@ package org.webrtc
 object NativeLibraryLoaderTestHelper {
     fun initialize() {
+        if (!NativeLibrary.isLoaded()) {
             NativeLibrary.initialize({ true }, "")
         }
+    }
 }
--- a/livekit-android-sdk/src/test/java/org/webrtc/WebRTCInitializer.kt 0 → 100644
查看文件 @eb2246d
+++ b/livekit-android-sdk/src/test/java/org/webrtc/WebRTCInitializer.kt 0 → 100644
查看文件 @eb2246d
+package org.webrtc
+
+import android.content.Context
+import org.mockito.Mockito
+
+object WebRTCInitializer {
+    fun initialize(context: Context = Mockito.mock(Context::class.java)) {
+        try {
+            ContextUtils.initialize(context)
+            NativeLibraryLoaderTestHelper.initialize()
+        } catch (e: Throwable) {
+            // do nothing. this is expected.
+        }
+    }
+}
--- a/protocol @ a4208afd
+++ b/protocol @ a4208afd
-Subproject commit 2c4c8d7764edf02818d1af686acc89165a5128bc
+Subproject commit a4208afda1fd87c5e57efa14242c1fa94e34ec07
--- a/sample-app-common/build.gradle
查看文件 @eb2246d
+++ b/sample-app-common/build.gradle
查看文件 @eb2246d
@@ -40,6 +40,7 @@ dependencies {
     api "androidx.lifecycle:lifecycle-runtime-ktx:${versions.androidx_lifecycle}"
     api "androidx.lifecycle:lifecycle-viewmodel-ktx:${versions.androidx_lifecycle}"
     api "androidx.lifecycle:lifecycle-common-java8:${versions.androidx_lifecycle}"
+    api "com.google.protobuf:protobuf-java:${versions.protobuf}"
     api project(":livekit-android-sdk")
     testImplementation 'junit:junit:4.+'
     androidTestImplementation 'androidx.test.ext:junit:1.1.3'
--- a/sample-app-common/src/main/java/io/livekit/android/sample/CallViewModel.kt
查看文件 @eb2246d
+++ b/sample-app-common/src/main/java/io/livekit/android/sample/CallViewModel.kt
查看文件 @eb2246d
@@ -17,6 +17,7 @@ import io.livekit.android.util.flow
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.launch
+import livekit.LivekitRtc
 @OptIn(ExperimentalCoroutinesApi::class)
 class CallViewModel(
@@ -246,6 +247,14 @@ class CallViewModel(
         mutablePermissionAllowed.value = !mutablePermissionAllowed.value
         room.value?.localParticipant?.setTrackSubscriptionPermissions(mutablePermissionAllowed.value)
     }
+
+    fun simulateMigration(){
+        room.value?.sendSimulateScenario(
+            LivekitRtc.SimulateScenario.newBuilder()
+                .setMigration(true)
+                .build()
+        )
+    }
 }
 private fun <T> LiveData<T>.hide(): LiveData<T> = this
--- a/sample-app-common/src/main/res/drawable/dots_horizontal_circle_outline.xml 0 → 100644
查看文件 @eb2246d
+++ b/sample-app-common/src/main/res/drawable/dots_horizontal_circle_outline.xml 0 → 100644
查看文件 @eb2246d
+<!-- drawable/dots_horizontal_circle_outline.xml -->
+<vector xmlns:android="http://schemas.android.com/apk/res/android"
+    android:height="24dp"
+    android:width="24dp"
+    android:viewportWidth="24"
+    android:viewportHeight="24">
+    <path android:fillColor="#000" android:pathData="M12,2A10,10 0 0,1 22,12A10,10 0 0,1 12,22A10,10 0 0,1 2,12A10,10 0 0,1 12,2M12,4A8,8 0 0,0 4,12A8,8 0 0,0 12,20A8,8 0 0,0 20,12A8,8 0 0,0 12,4M12,10.5A1.5,1.5 0 0,1 13.5,12A1.5,1.5 0 0,1 12,13.5A1.5,1.5 0 0,1 10.5,12A1.5,1.5 0 0,1 12,10.5M7.5,10.5A1.5,1.5 0 0,1 9,12A1.5,1.5 0 0,1 7.5,13.5A1.5,1.5 0 0,1 6,12A1.5,1.5 0 0,1 7.5,10.5M16.5,10.5A1.5,1.5 0 0,1 18,12A1.5,1.5 0 0,1 16.5,13.5A1.5,1.5 0 0,1 15,12A1.5,1.5 0 0,1 16.5,10.5Z" />
+</vector>
--- a/sample-app-compose/src/main/java/io/livekit/android/composesample/CallActivity.kt
查看文件 @eb2246d
+++ b/sample-app-compose/src/main/java/io/livekit/android/composesample/CallActivity.kt
查看文件 @eb2246d
@@ -27,6 +27,7 @@ import androidx.constraintlayout.compose.Dimension
 import androidx.lifecycle.lifecycleScope
 import com.github.ajalt.timberkt.Timber
 import com.google.accompanist.pager.ExperimentalPagerApi
+import io.livekit.android.composesample.ui.DebugMenuDialog
 import io.livekit.android.composesample.ui.theme.AppTheme
 import io.livekit.android.room.Room
 import io.livekit.android.room.participant.Participant
@@ -108,7 +109,8 @@ class CallActivity : AppCompatActivity() {
                 screencastEnabled,
                 permissionAllowed = permissionAllowed,
                 onExitClick = { finish() },
-                onSendMessage = { viewModel.sendData(it) }
+                onSendMessage = { viewModel.sendData(it) },
+                onSimulateMigration = { viewModel.simulateMigration() },
             )
         }
     }
@@ -156,6 +158,7 @@ class CallActivity : AppCompatActivity() {
         error: Throwable? = null,
         onSnackbarDismiss: () -> Unit = {},
         onSendMessage: (String) -> Unit = {},
+        onSimulateMigration: () -> Unit = {},
     ) {
         AppTheme(darkTheme = true) {
             ConstraintLayout(
@@ -229,6 +232,7 @@ class CallActivity : AppCompatActivity() {
                     val controlSize = 40.dp
                     val controlPadding = 4.dp
                     Row(
+                        modifier = Modifier.fillMaxWidth(),
                         horizontalArrangement = Arrangement.SpaceEvenly,
                         verticalAlignment = Alignment.Bottom,
                     ) {
@@ -368,6 +372,7 @@ class CallActivity : AppCompatActivity() {
                     Spacer(modifier = Modifier.height(10.dp))
                     Row(
+                        modifier = Modifier.fillMaxWidth(),
                         horizontalArrangement = Arrangement.SpaceEvenly,
                         verticalAlignment = Alignment.Bottom,
                     ) {
@@ -386,6 +391,28 @@ class CallActivity : AppCompatActivity() {
                                 tint = Color.White,
                             )
                         }
+
+                        var showDebugDialog by remember { mutableStateOf(false) }
+                        Surface(
+                            onClick = { showDebugDialog = true },
+                            indication = rememberRipple(false),
+                            modifier = Modifier
+                                .size(controlSize)
+                                .padding(controlPadding)
+                        ) {
+                            val resource = R.drawable.dots_horizontal_circle_outline
+                            Icon(
+                                painterResource(id = resource),
+                                contentDescription = "Permissions",
+                                tint = Color.White,
+                            )
+                        }
+                        if (showDebugDialog) {
+                            DebugMenuDialog(
+                                onDismissRequest = { showDebugDialog = false },
+                                simulateMigration = { onSimulateMigration() }
+                            )
+                        }
                     }
                 }
--- a/sample-app-compose/src/main/java/io/livekit/android/composesample/MainActivity.kt
查看文件 @eb2246d
+++ b/sample-app-compose/src/main/java/io/livekit/android/composesample/MainActivity.kt
查看文件 @eb2246d
@@ -4,7 +4,6 @@ import android.Manifest
 import android.content.Intent
 import android.content.pm.PackageManager
 import android.os.Bundle
-import android.widget.Space
 import android.widget.Toast
 import androidx.activity.ComponentActivity
 import androidx.activity.compose.setContent
@@ -175,8 +174,7 @@ class MainActivity : ComponentActivity() {
         const val PREFERENCES_KEY_URL = "url"
         const val PREFERENCES_KEY_TOKEN = "token"
-        const val URL = "wss://livekit.watercooler.fm"
-        const val TOKEN =
-            "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5ODQyMzE0OTgsImlzcyI6IkFQSU1teGlMOHJxdUt6dFpFb1pKVjlGYiIsImp0aSI6ImZvcnRoIiwibmJmIjoxNjI0MjMxNDk4LCJ2aWRlbyI6eyJyb29tIjoibXlyb29tIiwicm9vbUpvaW4iOnRydWV9fQ.PVx_lXAIGxcD2VRslosrbkigc777GXbu-DQME8hjJKI"
+        const val URL = "wss://www.example.com"
+        const val TOKEN = ""
     }
 }
--- a/sample-app-compose/src/main/java/io/livekit/android/composesample/ui/DebugMenuDialog.kt 0 → 100644
查看文件 @eb2246d
+++ b/sample-app-compose/src/main/java/io/livekit/android/composesample/ui/DebugMenuDialog.kt 0 → 100644
查看文件 @eb2246d
+package io.livekit.android.composesample.ui
+
+import androidx.compose.foundation.background
+import androidx.compose.foundation.layout.*
+import androidx.compose.foundation.shape.RoundedCornerShape
+import androidx.compose.material.Button
+import androidx.compose.material.Text
+import androidx.compose.runtime.Composable
+import androidx.compose.ui.Alignment
+import androidx.compose.ui.Modifier
+import androidx.compose.ui.graphics.Color
+import androidx.compose.ui.tooling.preview.Preview
+import androidx.compose.ui.unit.dp
+import androidx.compose.ui.window.Dialog
+
+@Preview
+@Composable
+fun DebugMenuDialog(
+    onDismissRequest: () -> Unit = {},
+    simulateMigration: () -> Unit = {}
+) {
+    Dialog(onDismissRequest = onDismissRequest) {
+        Column(
+            horizontalAlignment = Alignment.CenterHorizontally,
+            modifier = Modifier
+                .background(Color.DarkGray, shape = RoundedCornerShape(3.dp))
+                .fillMaxWidth()
+                .padding(10.dp)
+        ) {
+            Text("Debug Menu", color = Color.White)
+            Spacer(modifier = Modifier.height(10.dp))
+
+            Button(onClick = {
+                simulateMigration()
+                onDismissRequest()
+            }) {
+                Text("Simulate Migration")
+            }
+        }
+    }
+}