Manually set connection state after reconnection (#130)

subscriber may not have disconnected and won't fire a connection state update

@@ -21,7 +21,7 @@ class PublisherTransportObserver(
     }
     }
 
 
     override fun onRenegotiationNeeded() {
     override fun onRenegotiationNeeded() {
-        engine.negotiate()
+        engine.negotiatePublisher()
     }
     }
 
 
     override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
     override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {

@@ -150,7 +150,7 @@ internal constructor(
 
 
         // create offer
         // create offer
         if (!this.isSubscriberPrimary) {
         if (!this.isSubscriberPrimary) {
-            negotiate()
+            negotiatePublisher()
         }
         }
         client.onReadyForResponses()
         client.onReadyForResponses()
         return joinResponse
         return joinResponse
@@ -392,7 +392,7 @@ internal constructor(
                     // trigger publisher reconnect
                     // trigger publisher reconnect
                     // only restart publisher if it's needed
                     // only restart publisher if it's needed
                     if (hasPublished) {
                     if (hasPublished) {
-                        negotiate()
+                        negotiatePublisher()
                     }
                     }
                 }
                 }
                 // wait until ICE connected
                 // wait until ICE connected
@@ -408,8 +408,9 @@ internal constructor(
                 }
                 }
 
 
                 while (SystemClock.elapsedRealtime() < endTime) {
                 while (SystemClock.elapsedRealtime() < endTime) {
-                    if (connectionState == ConnectionState.CONNECTED) {
+                    if (subscriber.peerConnection.connectionState().isConnected()) {
                         LKLog.v { "reconnected to ICE" }
                         LKLog.v { "reconnected to ICE" }
+                        connectionState = ConnectionState.CONNECTED
                         break
                         break
                     }
                     }
                     delay(100)
                     delay(100)
@@ -442,7 +443,7 @@ internal constructor(
         }
         }
     }
     }
 
 
-    internal fun negotiate() {
+    internal fun negotiatePublisher() {
         if (!client.isConnected) {
         if (!client.isConnected) {
             return
             return
         }
         }
@@ -481,7 +482,7 @@ internal constructor(
             publisher.peerConnection.iceConnectionState() != PeerConnection.IceConnectionState.CHECKING
             publisher.peerConnection.iceConnectionState() != PeerConnection.IceConnectionState.CHECKING
         ) {
         ) {
             // start negotiation
             // start negotiation
-            this.negotiate()
+            this.negotiatePublisher()
         }
         }
 
 
 
 

@@ -62,6 +62,7 @@ constructor(
 
 
     private val responseFlow = MutableSharedFlow<LivekitRtc.SignalResponse>(Int.MAX_VALUE)
     private val responseFlow = MutableSharedFlow<LivekitRtc.SignalResponse>(Int.MAX_VALUE)
 
 
+    var connectionState: ConnectionState = ConnectionState.DISCONNECTED
 
 
     /**
     /**
      * @throws Exception if fails to connect.
      * @throws Exception if fails to connect.

@@ -28,7 +28,6 @@ import org.mockito.Mockito
 import org.robolectric.RobolectricTestRunner
 import org.robolectric.RobolectricTestRunner
 import org.robolectric.Shadows.shadowOf
 import org.robolectric.Shadows.shadowOf
 import org.robolectric.shadows.ShadowConnectivityManager
 import org.robolectric.shadows.ShadowConnectivityManager
-import org.robolectric.shadows.ShadowNetworkInfo
 
 
 @ExperimentalCoroutinesApi
 @ExperimentalCoroutinesApi
 @RunWith(RobolectricTestRunner::class)
 @RunWith(RobolectricTestRunner::class)
@@ -346,4 +345,21 @@ class RoomMockE2ETest : MockE2ETest() {
         assertTrue(events[0] is RoomEvent.Reconnecting)
         assertTrue(events[0] is RoomEvent.Reconnecting)
         assertTrue(events[1] is RoomEvent.Reconnected)
         assertTrue(events[1] is RoomEvent.Reconnected)
     }
     }
+
+    @Test
+    fun reconnectFromWebSocketFailure() = runTest {
+        connect()
+
+        val eventCollector = EventCollector(room.events, coroutineRule.scope)
+        wsFactory.onOpen = {
+            wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
+            connectPeerConnection()
+        }
+        wsFactory.ws.cancel()
+        val events = eventCollector.stopCollecting()
+
+        assertEquals(2, events.size)
+        assertTrue(events[0] is RoomEvent.Reconnecting)
+        assertTrue(events[1] is RoomEvent.Reconnected)
+    }
 }
 }