Make sure consume elements from the TestChannel

packages/test-base/src/commonMain/kotlin/io/realm/kotlin/test/util/Utils.kt

@@ -22,9 +22,9 @@ import io.realm.kotlin.test.platform.PlatformUtils
 import io.realm.kotlin.types.RealmInstant
 import io.realm.kotlin.types.RealmObject
 import kotlinx.coroutines.TimeoutCancellationException
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.selects.onTimeout
-import kotlinx.coroutines.selects.select
+import kotlinx.coroutines.withTimeout
 import kotlinx.datetime.Instant
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.minutes
@@ -93,19 +93,34 @@ fun Instant.toRealmInstant(): RealmInstant {
     }
 }
 
+/**
+ * Channel implementation specifically suited for tests. Its size is unlimited, but will fail
+ * the test if canceled containing unconsumed elements.
+ */
+inline fun <T> TestChannel(): Channel<T> {
+    return Channel<T>(capacity = Channel.UNLIMITED, onBufferOverflow = BufferOverflow.SUSPEND) {
+        throw AssertionError("Failed to deliver: $it")
+    }
+}
+
 // Variant of `Channel.receiveOrFail()` that will will throw if a timeout is hit.
 suspend fun <T : Any?> Channel<T>.receiveOrFail(timeout: Duration = 1.minutes, message: String? = null): T {
-//    return withTimeout(timeout) {
-//        [email protected]()
-//    }
-    return select {
-        this@receiveOrFail.onReceive {
-            println("Got: $it")
-            it
-        }
-        onTimeout(timeout) {
-            @Suppress("invisible_member")
-            throw TimeoutCancellationException("Timeout after $timeout: ${if (message.isNullOrBlank()) "<no message>" else message}")
+    try {
+        return withTimeout(timeout) {
+            this@receiveOrFail.receive()
         }
+    } catch (ex: TimeoutCancellationException) {
+        throw AssertionError("Timeout after $timeout: ${if (message.isNullOrBlank()) "<no message>" else message}")
     }
+//    return select {
+//        [email protected] {
+//            println("Got: $it")
+// //            receive() // Make sure to empty the channel
+//             it
+//        }
+//        onTimeout(timeout) {
+//            @Suppress("invisible_member")
+//            throw TimeoutCancellationException()
+//        }
+//    }
 }

packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/VersionTrackingTests.kt

@@ -31,8 +31,8 @@ import io.realm.kotlin.log.LogLevel
 import io.realm.kotlin.log.RealmLog
 import io.realm.kotlin.notifications.RealmChange
 import io.realm.kotlin.notifications.ResultsChange
-import io.realm.kotlin.test.common.notifications.TestChannel
 import io.realm.kotlin.test.platform.PlatformUtils
+import io.realm.kotlin.test.util.TestChannel
 import io.realm.kotlin.test.util.receiveOrFail
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.async

packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmNotificationsTests.kt

@@ -28,11 +28,11 @@ import io.realm.kotlin.notifications.RealmChange
 import io.realm.kotlin.notifications.UpdatedRealm
 import io.realm.kotlin.test.common.utils.FlowableTests
 import io.realm.kotlin.test.platform.PlatformUtils
+import io.realm.kotlin.test.util.TestChannel
 import io.realm.kotlin.test.util.receiveOrFail
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.sync.Mutex
@@ -48,12 +48,6 @@ import kotlin.test.fail
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
-inline fun <T> TestChannel(): Channel<T> {
-    return Channel<T>(capacity = Channel.UNLIMITED, onBufferOverflow = BufferOverflow.SUSPEND) {
-        fail("Failed to deliver: $it")
-    }
-}
-
 class RealmNotificationsTests : FlowableTests {
 
     lateinit var tmpDir: String