cancel join continuation if websocket fails to connect

@@ -144,4 +144,6 @@ sealed class RoomEvent(val room: Room) : Event() {
     class ConnectionQualityChanged(room: Room, val participant: Participant, val quality: ConnectionQuality) :
     class ConnectionQualityChanged(room: Room, val participant: Participant, val quality: ConnectionQuality) :
         RoomEvent(room)
         RoomEvent(room)
 
 
+    class FailedToConnect(room: Room, val error: Throwable) : RoomEvent(room)
+
 }
 }

@@ -385,7 +385,7 @@ internal constructor(
         fun onConnectionQuality(updates: List<LivekitRtc.ConnectionQualityInfo>)
         fun onConnectionQuality(updates: List<LivekitRtc.ConnectionQualityInfo>)
         fun onSpeakersChanged(speakers: List<LivekitModels.SpeakerInfo>)
         fun onSpeakersChanged(speakers: List<LivekitModels.SpeakerInfo>)
         fun onDisconnect(reason: String)
         fun onDisconnect(reason: String)
-        fun onFailToConnect(error: Exception)
+        fun onFailToConnect(error: Throwable)
         fun onUserPacket(packet: LivekitModels.UserPacket, kind: LivekitModels.DataPacket.Kind)
         fun onUserPacket(packet: LivekitModels.UserPacket, kind: LivekitModels.DataPacket.Kind)
         fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>)
         fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>)
     }
     }
@@ -521,7 +521,7 @@ internal constructor(
         listener?.onDisconnect("")
         listener?.onDisconnect("")
     }
     }
 
 
-    override fun onError(error: Exception) {
+    override fun onError(error: Throwable) {
         listener?.onFailToConnect(error)
         listener?.onFailToConnect(error)
     }
     }
 
 

@@ -20,7 +20,6 @@ import io.livekit.android.room.participant.*
 import io.livekit.android.room.track.*
 import io.livekit.android.room.track.*
 import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.LKLog
-import io.livekit.android.util.flow
 import io.livekit.android.util.flowDelegate
 import io.livekit.android.util.flowDelegate
 import kotlinx.coroutines.*
 import kotlinx.coroutines.*
 import livekit.LivekitModels
 import livekit.LivekitModels
@@ -465,8 +464,10 @@ constructor(
     /**
     /**
      * @suppress
      * @suppress
      */
      */
-    override fun onFailToConnect(error: Exception) {
+    override fun onFailToConnect(error: Throwable) {
         listener?.onFailedToConnect(this, error)
         listener?.onFailedToConnect(this, error)
+        // scope will likely be closed already here, so force it out of scope.
+        eventBus.tryPostEvent(RoomEvent.FailedToConnect(this, error))
     }
     }
 
 
     //------------------------------- ParticipantListener --------------------------------//
     //------------------------------- ParticipantListener --------------------------------//
@@ -613,7 +614,7 @@ interface RoomListener {
     /**
     /**
      * Could not connect to the room
      * Could not connect to the room
      */
      */
-    fun onFailedToConnect(room: Room, error: Exception) {}
+    fun onFailedToConnect(room: Room, error: Throwable) {}
 //        fun onReconnecting(room: Room, error: Exception) {}
 //        fun onReconnecting(room: Room, error: Exception) {}
 //        fun onReconnect(room: Room) {}
 //        fun onReconnect(room: Room) {}
 
 

@@ -12,7 +12,6 @@ import io.livekit.android.util.safe
 import kotlinx.coroutines.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.collect
 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
@@ -27,8 +26,6 @@ import org.webrtc.SessionDescription
 import javax.inject.Inject
 import javax.inject.Inject
 import javax.inject.Named
 import javax.inject.Named
 import javax.inject.Singleton
 import javax.inject.Singleton
-import kotlin.coroutines.Continuation
-import kotlin.coroutines.suspendCoroutine
 
 
 /**
 /**
  * SignalClient to LiveKit WS servers
  * SignalClient to LiveKit WS servers
@@ -186,10 +183,13 @@ constructor(
 
 
         if (reason != null) {
         if (reason != null) {
             LKLog.e(t) { "websocket failure: $reason" }
             LKLog.e(t) { "websocket failure: $reason" }
-            listener?.onError(Exception(reason))
+            val error = Exception(reason)
+            listener?.onError(error)
+            joinContinuation?.cancel(error)
         } else {
         } else {
             LKLog.e(t) { "websocket failure: $response" }
             LKLog.e(t) { "websocket failure: $response" }
-            listener?.onError(t as Exception)
+            listener?.onError(t)
+            joinContinuation?.cancel(t)
         }
         }
     }
     }
 
 
@@ -443,7 +443,7 @@ constructor(
         fun onRoomUpdate(update: LivekitModels.Room)
         fun onRoomUpdate(update: LivekitModels.Room)
         fun onConnectionQuality(updates: List<LivekitRtc.ConnectionQualityInfo>)
         fun onConnectionQuality(updates: List<LivekitRtc.ConnectionQualityInfo>)
         fun onLeave()
         fun onLeave()
-        fun onError(error: Exception)
+        fun onError(error: Throwable)
         fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>)
         fun onStreamStateUpdate(streamStates: List<LivekitRtc.StreamStateInfo>)
     }
     }
 
 

@@ -69,6 +69,29 @@ class RoomMockE2ETest {
     }
     }
 
 
     @Test
     @Test
+    fun connectFailureProperlyContinues(){
+
+        var didThrowException = false
+        val job = coroutineRule.scope.launch {
+            try {
+                room.connect(
+                    url = "http://www.example.com",
+                    token = "",
+                )
+            } catch (e: Throwable) {
+                didThrowException = true
+            }
+        }
+
+        wsFactory.listener.onFailure(wsFactory.ws, Exception(), null)
+
+        runBlockingTest {
+            job.join()
+        }
+
+        Assert.assertTrue(didThrowException)
+    }
+    @Test
     fun roomUpdateTest() {
     fun roomUpdateTest() {
         connect()
         connect()
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
         val eventCollector = EventCollector(room.events, coroutineRule.scope)

@@ -4,9 +4,10 @@ import android.app.Application
 import android.content.Intent
 import android.content.Intent
 import androidx.lifecycle.*
 import androidx.lifecycle.*
 import com.github.ajalt.timberkt.Timber
 import com.github.ajalt.timberkt.Timber
-import io.livekit.android.ConnectOptions
 import io.livekit.android.LiveKit
 import io.livekit.android.LiveKit
 import io.livekit.android.RoomOptions
 import io.livekit.android.RoomOptions
+import io.livekit.android.events.RoomEvent
+import io.livekit.android.events.collect
 import io.livekit.android.room.Room
 import io.livekit.android.room.Room
 import io.livekit.android.room.RoomListener
 import io.livekit.android.room.RoomListener
 import io.livekit.android.room.participant.Participant
 import io.livekit.android.room.participant.Participant
@@ -40,6 +41,9 @@ class CallViewModel(
         }
         }
     }
     }
 
 
+    private val mutableError = MutableStateFlow<Throwable?>(null)
+    val error = mutableError.hide()
+
     private val mutablePrimarySpeaker = MutableStateFlow<Participant?>(null)
     private val mutablePrimarySpeaker = MutableStateFlow<Participant?>(null)
     val primarySpeaker: StateFlow<Participant?> = mutablePrimarySpeaker
     val primarySpeaker: StateFlow<Participant?> = mutablePrimarySpeaker
 
 
@@ -67,24 +71,36 @@ class CallViewModel(
 
 
     init {
     init {
         viewModelScope.launch {
         viewModelScope.launch {
-            val room = LiveKit.connect(
-                application,
-                url,
-                token,
-                roomOptions = RoomOptions(autoManageVideo = true),
-                listener = this@CallViewModel
-            )
-
-            // Create and publish audio/video tracks
-            val localParticipant = room.localParticipant
-            localParticipant.setMicrophoneEnabled(true)
-            mutableMicEnabled.postValue(localParticipant.isMicrophoneEnabled())
-
-            localParticipant.setCameraEnabled(true)
-            mutableCameraEnabled.postValue(localParticipant.isCameraEnabled())
-            mutableRoom.value = room
-
-            mutablePrimarySpeaker.value = room.remoteParticipants.values.firstOrNull() ?: localParticipant
+            try {
+                val room = LiveKit.connect(
+                    application,
+                    url,
+                    token,
+                    roomOptions = RoomOptions(autoManageVideo = true),
+                    listener = this@CallViewModel
+                )
+
+                // Create and publish audio/video tracks
+                val localParticipant = room.localParticipant
+                localParticipant.setMicrophoneEnabled(true)
+                mutableMicEnabled.postValue(localParticipant.isMicrophoneEnabled())
+
+                localParticipant.setCameraEnabled(true)
+                mutableCameraEnabled.postValue(localParticipant.isCameraEnabled())
+                mutableRoom.value = room
+
+                mutablePrimarySpeaker.value = room.remoteParticipants.values.firstOrNull() ?: localParticipant
+
+                viewModelScope.launch {
+                    room.events.collect {
+                        when (it) {
+                            is RoomEvent.FailedToConnect -> mutableError.value = it.error
+                        }
+                    }
+                }
+            } catch (e: Throwable) {
+                mutableError.value = e
+            }
         }
         }
     }
     }
 
 
@@ -166,6 +182,12 @@ class CallViewModel(
             videoTrack.restartTrack(newOptions)
             videoTrack.restartTrack(newOptions)
         }
         }
     }
     }
+
+    fun dismissError() {
+        mutableError.value = null
+    }
 }
 }
 
 
 private fun <T> LiveData<T>.hide(): LiveData<T> = this
 private fun <T> LiveData<T>.hide(): LiveData<T> = this
+
+private fun <T> MutableStateFlow<T>.hide(): StateFlow<T> = this

@@ -5,6 +5,7 @@ import android.media.AudioManager
 import android.media.projection.MediaProjectionManager
 import android.media.projection.MediaProjectionManager
 import android.os.Bundle
 import android.os.Bundle
 import android.os.Parcelable
 import android.os.Parcelable
+import android.widget.Toast
 import androidx.activity.compose.setContent
 import androidx.activity.compose.setContent
 import androidx.activity.result.contract.ActivityResultContracts
 import androidx.activity.result.contract.ActivityResultContracts
 import androidx.appcompat.app.AppCompatActivity
 import androidx.appcompat.app.AppCompatActivity
@@ -23,6 +24,7 @@ import androidx.compose.ui.tooling.preview.Preview
 import androidx.compose.ui.unit.dp
 import androidx.compose.ui.unit.dp
 import androidx.constraintlayout.compose.ConstraintLayout
 import androidx.constraintlayout.compose.ConstraintLayout
 import androidx.constraintlayout.compose.Dimension
 import androidx.constraintlayout.compose.Dimension
+import androidx.lifecycle.lifecycleScope
 import com.github.ajalt.timberkt.Timber
 import com.github.ajalt.timberkt.Timber
 import com.google.accompanist.pager.ExperimentalPagerApi
 import com.google.accompanist.pager.ExperimentalPagerApi
 import io.livekit.android.composesample.ui.theme.AppTheme
 import io.livekit.android.composesample.ui.theme.AppTheme
@@ -30,6 +32,8 @@ import io.livekit.android.room.Room
 import io.livekit.android.room.participant.Participant
 import io.livekit.android.room.participant.Participant
 import io.livekit.android.sample.CallViewModel
 import io.livekit.android.sample.CallViewModel
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
 import kotlinx.parcelize.Parcelize
 import kotlinx.parcelize.Parcelize
 
 
 @OptIn(ExperimentalPagerApi::class)
 @OptIn(ExperimentalPagerApi::class)
@@ -82,6 +86,15 @@ class CallActivity : AppCompatActivity() {
             Timber.v { "Audio focus request failed" }
             Timber.v { "Audio focus request failed" }
         }
         }
 
 
+        lifecycleScope.launchWhenStarted {
+            viewModel.error.collect {
+                if(it != null){
+                    Toast.makeText(this@CallActivity, "Error: $it", Toast.LENGTH_LONG).show()
+                    viewModel.dismissError()
+                }
+            }
+        }
+
         // Setup compose view.
         // Setup compose view.
         setContent {
         setContent {
             val room by viewModel.room.collectAsState()
             val room by viewModel.room.collectAsState()
@@ -127,6 +140,8 @@ class CallActivity : AppCompatActivity() {
         flipButtonEnabled: Boolean = true,
         flipButtonEnabled: Boolean = true,
         screencastEnabled: Boolean = false,
         screencastEnabled: Boolean = false,
         onExitClick: () -> Unit = {},
         onExitClick: () -> Unit = {},
+        error: Throwable? = null,
+        onSnackbarDismiss: () -> Unit = {}
     ) {
     ) {
         AppTheme(darkTheme = true) {
         AppTheme(darkTheme = true) {
             ConstraintLayout(
             ConstraintLayout(
@@ -271,6 +286,32 @@ class CallActivity : AppCompatActivity() {
                         )
                         )
                     }
                     }
                 }
                 }
+
+                // Snack bar for errors
+                val scaffoldState = rememberScaffoldState()
+                val scope = rememberCoroutineScope()
+                if(error != null) {
+                    Scaffold(
+                        scaffoldState = scaffoldState,
+                        floatingActionButton = {
+                            ExtendedFloatingActionButton(
+                                text = { Text("Show snackbar") },
+                                onClick = {
+                                    // show snackbar as a suspend function
+                                    scope.launch {
+                                        scaffoldState.snackbarHostState.showSnackbar(error?.toString() ?: "")
+                                    }
+                                }
+                            )
+                        },
+                        content = { innerPadding ->
+                            Text(
+                                text = "Body content",
+                                modifier = Modifier.padding(innerPadding).fillMaxSize().wrapContentSize()
+                            )
+                        }
+                    )
+                }
             }
             }
         }
         }
     }
     }

@@ -6,6 +6,7 @@ import android.media.projection.MediaProjectionManager
 import android.os.Bundle
 import android.os.Bundle
 import android.os.Parcelable
 import android.os.Parcelable
 import android.view.View
 import android.view.View
+import android.widget.Toast
 import androidx.activity.result.contract.ActivityResultContracts
 import androidx.activity.result.contract.ActivityResultContracts
 import androidx.appcompat.app.AppCompatActivity
 import androidx.appcompat.app.AppCompatActivity
 import androidx.lifecycle.lifecycleScope
 import androidx.lifecycle.lifecycleScope
@@ -71,6 +72,15 @@ class CallActivity : AppCompatActivity() {
                 }
                 }
         }
         }
 
 
+        lifecycleScope.launchWhenStarted {
+            viewModel.error.collect {
+                if(it != null){
+                    Toast.makeText(this@CallActivity, "Error: $it", Toast.LENGTH_LONG).show()
+                    viewModel.dismissError()
+                }
+            }
+        }
+
         // speaker view setup
         // speaker view setup
         lifecycleScope.launchWhenCreated {
         lifecycleScope.launchWhenCreated {
             viewModel.room.filterNotNull().take(1)
             viewModel.room.filterNotNull().take(1)