reconnect upon publisher failure (#77)

@@ -12,7 +12,7 @@ class PublisherTransportObserver(
     private val client: SignalClient,
 ) : PeerConnection.Observer, PeerConnectionTransport.Listener {
 
-    var connectionChangeListener: ((newState: PeerConnection.PeerConnectionState?) -> Unit)? = null
+    var connectionChangeListener: ((newState: PeerConnection.PeerConnectionState) -> Unit)? = null
 
     override fun onIceCandidate(iceCandidate: IceCandidate?) {
         val candidate = iceCandidate ?: return
@@ -35,7 +35,7 @@ class PublisherTransportObserver(
     override fun onStandardizedIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
     }
 
-    override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
+    override fun onConnectionChange(newState: PeerConnection.PeerConnectionState) {
         LKLog.v { "onConnection new state: $newState" }
         connectionChangeListener?.invoke(newState)
     }

@@ -210,13 +210,11 @@ internal constructor(
             null,
         )
 
-        val connectionStateListener: (PeerConnection.PeerConnectionState?) -> Unit = { newState ->
-            val state =
-                newState ?: throw NullPointerException("unexpected null new state, what do?")
+        val connectionStateListener: (PeerConnection.PeerConnectionState) -> Unit = { newState ->
             LKLog.v { "onIceConnection new state: $newState" }
-            if (state.isConnected()) {
+            if (newState.isConnected()) {
                 connectionState = ConnectionState.CONNECTED
-            } else if (state.isDisconnected()) {
+            } else if (newState.isDisconnected()) {
                 connectionState = ConnectionState.DISCONNECTED
             }
         }
@@ -233,6 +231,12 @@ internal constructor(
             }
 
             subscriberObserver.connectionChangeListener = connectionStateListener
+            // Also reconnect on publisher disconnect
+            publisherObserver.connectionChangeListener = { newState ->
+                if (newState.isDisconnected()) {
+                    reconnect()
+                }
+            }
         } else {
             publisherObserver.connectionChangeListener = connectionStateListener
         }
@@ -387,6 +391,16 @@ internal constructor(
                 }
                 // wait until ICE connected
                 val endTime = SystemClock.elapsedRealtime() + MAX_ICE_CONNECT_TIMEOUT_MS
+                if (hasPublished) {
+                    while (SystemClock.elapsedRealtime() < endTime) {
+                        if (publisher.peerConnection.connectionState().isConnected()) {
+                            LKLog.v { "publisher reconnected to ICE" }
+                            break
+                        }
+                        delay(100)
+                    }
+                }
+
                 while (SystemClock.elapsedRealtime() < endTime) {
                     if (connectionState == ConnectionState.CONNECTED) {
                         LKLog.v { "reconnected to ICE" }
@@ -395,7 +409,9 @@ internal constructor(
                     delay(100)
                 }
 
-                if (connectionState == ConnectionState.CONNECTED) {
+                if (connectionState == ConnectionState.CONNECTED &&
+                    (!hasPublished || publisher.peerConnection.connectionState().isConnected())
+                ) {
                     client.onPCConnected()
                     listener?.onPostReconnect(isFullReconnect)
                     return@launch

@@ -13,7 +13,7 @@ class SubscriberTransportObserver(
 ) : PeerConnection.Observer {
 
     var dataChannelListener: ((DataChannel) -> Unit)? = null
-    var connectionChangeListener: ((PeerConnection.PeerConnectionState?) -> Unit)? = null
+    var connectionChangeListener: ((PeerConnection.PeerConnectionState) -> Unit)? = null
 
     override fun onIceCandidate(candidate: IceCandidate) {
         LKLog.v { "onIceCandidate: $candidate" }
@@ -41,7 +41,7 @@ class SubscriberTransportObserver(
     override fun onStandardizedIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
     }
 
-    override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
+    override fun onConnectionChange(newState: PeerConnection.PeerConnectionState) {
         LKLog.v { "onConnectionChange new state: $newState" }
         connectionChangeListener?.invoke(newState)
     }

@@ -223,9 +223,7 @@ class MockPeerConnection(
         }
     }
 
-    override fun connectionState(): PeerConnectionState {
-        return super.connectionState()
-    }
+    override fun connectionState(): PeerConnectionState = connectionState
 
     override fun iceGatheringState(): IceGatheringState {
         return super.iceGatheringState()

@@ -2,7 +2,6 @@ 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.toOkioByteString
 import io.livekit.android.util.toPBByteString
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -12,6 +11,7 @@ import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.robolectric.RobolectricTestRunner
+import org.webrtc.PeerConnection
 
 
 @ExperimentalCoroutinesApi
@@ -57,6 +57,30 @@ class RTCEngineMockE2ETest : MockE2ETest() {
     }
 
     @Test
+    fun reconnectOnSubscriberFailure() = runTest {
+        connect()
+        val oldWs = wsFactory.ws
+
+        val subPeerConnection = rtcEngine.subscriber.peerConnection as MockPeerConnection
+        subPeerConnection.moveToIceConnectionState(PeerConnection.IceConnectionState.FAILED)
+
+        val newWs = wsFactory.ws
+        Assert.assertNotEquals(oldWs, newWs)
+    }
+
+    @Test
+    fun reconnectOnPublisherFailure() = runTest {
+        connect()
+        val oldWs = wsFactory.ws
+
+        val pubPeerConnection = rtcEngine.publisher.peerConnection as MockPeerConnection
+        pubPeerConnection.moveToIceConnectionState(PeerConnection.IceConnectionState.FAILED)
+
+        val newWs = wsFactory.ws
+        Assert.assertNotEquals(oldWs, newWs)
+    }
+
+    @Test
     fun refreshToken() = runTest {
         connect()