update kotlin versions and tests

@@ -2,8 +2,9 @@
 
 buildscript {
     ext {
-        compose_version = '1.0.5'
-        kotlin_version = '1.5.31'
+        compose_version = '1.1.0-rc01'
+        compose_compiler_version = '1.1.0-rc02'
+        kotlin_version = '1.6.10'
         java_version = JavaVersion.VERSION_1_8
         dokka_version = '1.5.0'
     }
@@ -13,7 +14,7 @@ buildscript {
         jcenter()
     }
     dependencies {
-        classpath 'com.android.tools.build:gradle:7.0.3'
+        classpath 'com.android.tools.build:gradle:7.0.4'
         classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
         classpath "org.jetbrains.kotlin:kotlin-serialization:$kotlin_version"
         classpath "org.jetbrains.dokka:dokka-gradle-plugin:$dokka_version"
@@ -69,7 +70,7 @@ ext {
                     'service'           : "com.google.auto.service:auto-service:${versions.autoService}",
                     'serviceAnnotations': "com.google.auto.service:auto-service-annotations:${versions.autoService}",
             ],
-            kotlinx_coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.5.2",
+            kotlinx_coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.0",
             timber            : "com.github.ajalt:timberkt:1.5.1",
             // lint
             lint              : "com.android.tools.lint:lint:${versions.lint}",

 package io.livekit.android.events
 
 import kotlinx.coroutines.flow.SharedFlow
-import kotlinx.coroutines.flow.collect
 
 interface EventListenable<out T> {
     val events: SharedFlow<T>
 }
 
 suspend inline fun <T> EventListenable<T>.collect(crossinline action: suspend (value: T) -> Unit) {
-    return events.collect(action)
+    events.collect { value -> action(value) }
 }
\ No newline at end of file

+package io.livekit.android
+
+import io.livekit.android.coroutines.TestCoroutineRule
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runTest
+import org.junit.Rule
+import org.mockito.junit.MockitoJUnit
+
+@ExperimentalCoroutinesApi
+abstract class BaseTest {
+
+    @get:Rule
+    var mockitoRule = MockitoJUnit.rule()
+
+    @get:Rule
+    var coroutineRule = TestCoroutineRule()
+
+    @ExperimentalCoroutinesApi
+    fun runTest(testBody: suspend TestScope.() -> Unit) = coroutineRule.scope.runTest(testBody = testBody)
+}

@@ -2,7 +2,6 @@ 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
@@ -12,22 +11,13 @@ 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()
+abstract class MockE2ETest : BaseTest() {
 
     internal lateinit var component: TestLiveKitComponent
     lateinit var context: Context
@@ -46,7 +36,7 @@ abstract class MockE2ETest {
         wsFactory = component.websocketFactory()
     }
 
-    fun connect() {
+    suspend fun connect() {
         val job = coroutineRule.scope.launch {
             room.connect(
                 url = SignalClientTest.EXAMPLE_URL,
@@ -56,12 +46,8 @@ abstract class MockE2ETest {
         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()

 package io.livekit.android.coroutines
 
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.test.TestCoroutineDispatcher
-import kotlinx.coroutines.test.TestCoroutineScope
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.resetMain
+import kotlinx.coroutines.test.setMain
+import org.junit.Assert
 import org.junit.rules.TestRule
 import org.junit.runner.Description
 import org.junit.runners.model.Statement
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class TestCoroutineRule : TestRule {
-    val dispatcher = TestCoroutineDispatcher()
-    val scope = TestCoroutineScope(dispatcher)
+    val dispatcher = UnconfinedTestDispatcher()
+    val scope = TestScope(dispatcher)
 
     override fun apply(base: Statement, description: Description?) = object : Statement() {
         @Throws(Throwable::class)
         override fun evaluate() {
+            Dispatchers.setMain(dispatcher)
             base.evaluate()
-            scope.cleanupTestCoroutines()
+            val timeAfterTest = dispatcher.scheduler.currentTime
+            dispatcher.scheduler.advanceUntilIdle() // run the remaining tasks
+            Assert.assertEquals(
+                timeAfterTest,
+                dispatcher.scheduler.currentTime
+            ) // will fail if there were tasks scheduled at a later moment
+            Dispatchers.resetMain()
         }
     }
 }
\ No newline at end of file

@@ -6,7 +6,6 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.test.runBlockingTest
 
 open class FlowCollector<T>(
     private val flow: Flow<T>,
@@ -21,13 +20,9 @@ open class FlowCollector<T>(
      * Stop collecting events. returns the events collected.
      */
     @OptIn(ExperimentalCoroutinesApi::class)
-    fun stopCollecting(): List<T> {
+    suspend fun stopCollecting(): List<T> {
         signal.compareAndSet(null, Unit)
-        var events: List<T> = emptyList()
-        runBlockingTest {
-            events = collectEventsDeferred.await()
-        }
-        return events
+        return collectEventsDeferred.await()
     }
 
 }
\ No newline at end of file

@@ -6,7 +6,7 @@ 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 kotlinx.coroutines.test.runTest
 import livekit.LivekitRtc
 import org.junit.Assert
 import org.junit.Before
@@ -34,7 +34,7 @@ class RTCEngineMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun iceSubscriberConnect() = runBlockingTest {
+    fun iceSubscriberConnect() = runTest {
         connect()
 
         val remoteOffer = SessionDescription(SessionDescription.Type.OFFER, "remote_offer")
@@ -59,7 +59,7 @@ class RTCEngineMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun reconnectOnFailure() = runBlockingTest {
+    fun reconnectOnFailure() = runTest {
         connect()
         val oldWs = wsFactory.ws
         wsFactory.listener.onFailure(oldWs, Exception(), null)

@@ -13,7 +13,6 @@ import io.livekit.android.room.track.Track
 import io.livekit.android.util.toOkioByteString
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.test.runBlockingTest
 import org.junit.Assert
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -25,12 +24,12 @@ import org.robolectric.RobolectricTestRunner
 class RoomMockE2ETest : MockE2ETest() {
 
     @Test
-    fun connectTest() {
+    fun connectTest() = runTest {
         connect()
     }
 
     @Test
-    fun connectFailureProperlyContinues() {
+    fun connectFailureProperlyContinues() = runTest {
 
         var didThrowException = false
         val job = coroutineRule.scope.launch {
@@ -46,15 +45,13 @@ class RoomMockE2ETest : MockE2ETest() {
 
         wsFactory.listener.onFailure(wsFactory.ws, Exception(), null)
 
-        runBlockingTest {
         job.join()
-        }
 
         Assert.assertTrue(didThrowException)
     }
 
     @Test
-    fun roomUpdateTest() {
+    fun roomUpdateTest() = runTest {
         connect()
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
         wsFactory.listener.onMessage(wsFactory.ws, SignalClientTest.ROOM_UPDATE.toOkioByteString())
@@ -69,7 +66,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun connectionQualityUpdateTest() {
+    fun connectionQualityUpdateTest() = runTest {
         connect()
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
         wsFactory.listener.onMessage(
@@ -84,7 +81,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun participantConnected() {
+    fun participantConnected() = runTest {
         connect()
 
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
@@ -99,7 +96,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun participantDisconnected() {
+    fun participantDisconnected() = runTest {
         connect()
         wsFactory.listener.onMessage(
             wsFactory.ws,
@@ -118,7 +115,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun onActiveSpeakersChanged() {
+    fun onActiveSpeakersChanged() = runTest {
         connect()
 
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
@@ -133,7 +130,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun participantMetadataChanged() {
+    fun participantMetadataChanged() = runTest {
         connect()
 
         wsFactory.listener.onMessage(
@@ -153,7 +150,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun trackStreamStateChanged() {
+    fun trackStreamStateChanged() = runTest {
         connect()
 
         wsFactory.listener.onMessage(
@@ -189,7 +186,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun trackSubscriptionPermissionChanged() {
+    fun trackSubscriptionPermissionChanged() = runTest {
         connect()
 
         wsFactory.listener.onMessage(
@@ -224,7 +221,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun onConnectionAvailableWillReconnect() {
+    fun onConnectionAvailableWillReconnect() = runTest {
         connect()
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
         val network = Mockito.mock(Network::class.java)
@@ -237,7 +234,7 @@ class RoomMockE2ETest : MockE2ETest() {
     }
 
     @Test
-    fun leave() {
+    fun leave() = runTest {
         connect()
         val eventCollector = EventCollector(room.events, coroutineRule.scope)
         wsFactory.listener.onMessage(

@@ -8,8 +8,8 @@ import io.livekit.android.events.EventCollector
 import io.livekit.android.events.RoomEvent
 import io.livekit.android.mock.MockEglBase
 import io.livekit.android.room.participant.LocalParticipant
-import kotlinx.coroutines.*
-import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runTest
 import livekit.LivekitModels
 import org.junit.Assert
 import org.junit.Before
@@ -32,6 +32,7 @@ class RoomTest {
 
     @get:Rule
     var mockitoRule = MockitoJUnit.rule()
+
     @get:Rule
     var coroutineRule = TestCoroutineRule()
 
@@ -64,7 +65,7 @@ class RoomTest {
         )
     }
 
-    fun connect() {
+    suspend fun connect() {
         rtcEngine.stub {
             onBlocking { rtcEngine.join(any(), any(), anyOrNull()) }
                 .doReturn(SignalClientTest.JOIN.join)
@@ -73,24 +74,20 @@ class RoomTest {
             onBlocking { rtcEngine.client }
                 .doReturn(Mockito.mock(SignalClient::class.java))
         }
-        val job = coroutineRule.scope.launch {
+
         room.connect(
             url = SignalClientTest.EXAMPLE_URL,
             token = "",
         )
     }
-        runBlockingTest {
-            job.join()
-        }
-    }
 
     @Test
-    fun connectTest() {
+    fun connectTest() = runTest {
         connect()
     }
 
     @Test
-    fun onConnectionAvailableWillReconnect() {
+    fun onConnectionAvailableWillReconnect() = runTest {
         connect()
 
         val network = Mockito.mock(Network::class.java)
@@ -100,7 +97,7 @@ class RoomTest {
     }
 
     @Test
-    fun onDisconnect() {
+    fun onDisconnect() = runTest {
         connect()
 
         val eventCollector = EventCollector(room.events, coroutineRule.scope)

 package io.livekit.android.room
 
 import com.google.protobuf.util.JsonFormat
+import io.livekit.android.BaseTest
 import io.livekit.android.mock.MockWebSocketFactory
 import io.livekit.android.mock.TestData
 import io.livekit.android.util.toOkioByteString
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.async
-import kotlinx.coroutines.test.TestCoroutineDispatcher
-import kotlinx.coroutines.test.TestCoroutineScope
-import kotlinx.coroutines.test.runBlockingTest
 import kotlinx.serialization.json.Json
 import livekit.LivekitModels
 import livekit.LivekitRtc
 import okhttp3.*
-import org.junit.After
 import org.junit.Assert
 import org.junit.Before
 import org.junit.Test
+import org.mockito.Mock
 import org.mockito.Mockito
 import org.mockito.kotlin.any
 import org.mockito.kotlin.argThat
 import org.webrtc.SessionDescription
 
 @ExperimentalCoroutinesApi
-class SignalClientTest {
+class SignalClientTest : BaseTest() {
 
     lateinit var wsFactory: MockWebSocketFactory
     lateinit var client: SignalClient
+
+    @Mock
     lateinit var listener: SignalClient.Listener
-    lateinit var okHttpClient: OkHttpClient
 
-    lateinit var coroutineDispatcher: TestCoroutineDispatcher
-    lateinit var coroutineScope: TestCoroutineScope
+    @Mock
+    lateinit var okHttpClient: OkHttpClient
 
     @Before
     fun setup() {
-        coroutineDispatcher = TestCoroutineDispatcher()
-        coroutineScope = TestCoroutineScope(coroutineDispatcher)
         wsFactory = MockWebSocketFactory()
-        okHttpClient = Mockito.mock(OkHttpClient::class.java)
         client = SignalClient(
             wsFactory,
             JsonFormat.parser(),
@@ -46,17 +42,11 @@ class SignalClientTest {
             Json,
             useJson = false,
             okHttpClient = okHttpClient,
-            ioDispatcher = coroutineDispatcher
+            ioDispatcher = coroutineRule.dispatcher
         )
-        listener = Mockito.mock(SignalClient.Listener::class.java)
         client.listener = listener
     }
 
-    @After
-    fun tearDown() {
-        coroutineScope.cleanupTestCoroutines()
-    }
-
     private fun createOpenResponse(request: Request): Response {
         return Response.Builder()
             .request(request)
@@ -75,36 +65,35 @@ class SignalClientTest {
     }
 
     @Test
-    fun joinAndResponse() {
-        val job = coroutineScope.async {
+    fun joinAndResponse() = runTest {
+        println("dispatcher = ${this.coroutineContext}")
+        val job = async {
             client.join(EXAMPLE_URL, "")
         }
 
         connectWebsocketAndJoin()
 
-        runBlockingTest {
         val response = job.await()
+        Assert.assertEquals(true, client.isConnected)
         Assert.assertEquals(response, JOIN.join)
     }
-    }
 
     @Test
-    fun reconnect() {
-        val job = coroutineScope.async {
+    fun reconnect() = runTest {
+        val job = async {
             client.reconnect(EXAMPLE_URL, "")
         }
 
         client.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request))
 
-        runBlockingTest {
         job.await()
-        }
+        Assert.assertEquals(true, client.isConnected)
     }
 
     @Test
-    fun joinFailure() {
+    fun joinFailure() = runTest {
         var failed = false
-        val job = coroutineScope.async {
+        val job = async {
             try {
                 client.join(EXAMPLE_URL, "")
             } catch (e: Exception) {
@@ -113,34 +102,34 @@ class SignalClientTest {
         }
 
         client.onFailure(wsFactory.ws, Exception(), null)
-        runBlockingTest { job.await() }
+        job.await()
 
         Assert.assertTrue(failed)
     }
 
     @Test
-    fun listenerNotCalledUntilOnReady() {
-        val job = coroutineScope.async {
+    fun listenerNotCalledUntilOnReady() = runTest {
+        val job = async {
             client.join(EXAMPLE_URL, "")
         }
 
         connectWebsocketAndJoin()
         client.onMessage(wsFactory.ws, OFFER.toOkioByteString())
 
-        runBlockingTest { job.await() }
+        job.await()
 
         Mockito.verifyNoInteractions(listener)
     }
 
     @Test
-    fun listenerCalledAfterOnReady() {
-        val job = coroutineScope.async {
+    fun listenerCalledAfterOnReady() = runTest {
+        val job = async {
             client.join(EXAMPLE_URL, "")
         }
         connectWebsocketAndJoin()
         client.onMessage(wsFactory.ws, OFFER.toOkioByteString())
 
-        runBlockingTest { job.await() }
+        job.await()
         client.onReady()
         Mockito.verify(listener)
             .onOffer(argThat { type == SessionDescription.Type.OFFER && description == OFFER.offer.sdp })
@@ -151,12 +140,12 @@ class SignalClientTest {
      * [WebSocketListener.onClosed]. Ensure that listener is called properly.
      */
     @Test
-    fun listenerNotifiedAfterFailure() {
-        val job = coroutineScope.async {
+    fun listenerNotifiedAfterFailure() = runTest {
+        val job = async {
             client.join(EXAMPLE_URL, "")
         }
         connectWebsocketAndJoin()
-        runBlockingTest { job.await() }
+        job.await()
 
         client.onFailure(wsFactory.ws, Exception(), null)
 

@@ -4,6 +4,8 @@ import io.livekit.android.coroutines.TestCoroutineRule
 import io.livekit.android.events.EventCollector
 import io.livekit.android.events.ParticipantEvent
 import io.livekit.android.room.track.TrackPublication
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runTest
 import livekit.LivekitModels
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
@@ -11,6 +13,7 @@ import org.junit.Before
 import org.junit.Rule
 import org.junit.Test
 
+@ExperimentalCoroutinesApi
 class ParticipantTest {
 
     @get:Rule
@@ -24,7 +27,7 @@ class ParticipantTest {
     }
 
     @Test
-    fun updateFromInfo() {
+    fun updateFromInfo() = runTest {
         participant.updateFromInfo(INFO)
 
         assertTrue(participant.hasInfo)
@@ -36,7 +39,7 @@ class ParticipantTest {
     }
 
     @Test
-    fun setMetadataCallsListeners() {
+    fun setMetadataCallsListeners() = runTest {
         class MetadataListener : ParticipantListener {
             var wasCalled = false
             lateinit var participantValue: Participant
@@ -69,7 +72,7 @@ class ParticipantTest {
     }
 
     @Test
-    fun setMetadataChangedEvent() {
+    fun setMetadataChangedEvent() = runTest {
         val eventCollector = EventCollector(participant.events, coroutineRule.scope)
         val prevMetadata = participant.metadata
         val metadata = "metadata"
@@ -87,7 +90,7 @@ class ParticipantTest {
     }
 
     @Test
-    fun setIsSpeakingChangedEvent() {
+    fun setIsSpeakingChangedEvent() = runTest {
         val eventCollector = EventCollector(participant.events, coroutineRule.scope)
         val newIsSpeaking = !participant.isSpeaking
         participant.isSpeaking = newIsSpeaking
@@ -104,7 +107,7 @@ class ParticipantTest {
     }
 
     @Test
-    fun addTrackPublication() {
+    fun addTrackPublication() = runTest {
         val audioPublication = TrackPublication(TRACK_INFO, null, participant)
         participant.addTrackPublication(audioPublication)
 

@@ -38,8 +38,7 @@ android {
         compose true
     }
     composeOptions {
-        kotlinCompilerExtensionVersion compose_version
-        kotlinCompilerVersion kotlin_version
+        kotlinCompilerExtensionVersion compose_compiler_version
     }
     packagingOptions {
         resources {