Use kotlinx-coroutines-test, coroutines to 1.6.0, kotlin to 1.6.10

api/FlowExt.api

@@ -111,13 +111,13 @@ public final class com/hoc081098/flowext/Materialize_dematerializeKt {
 	public static final fun materialize (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 }
 
-public final class com/hoc081098/flowext/NeverFlowKt {
-	public static final fun neverFlow ()Lkotlinx/coroutines/flow/Flow;
+public final class com/hoc081098/flowext/NULL_VALUE {
+	public static final field INSTANCE Lcom/hoc081098/flowext/NULL_VALUE;
+	public final fun unbox (Ljava/lang/Object;)Ljava/lang/Object;
 }
 
-public final class com/hoc081098/flowext/NullValue {
-	public static final field INSTANCE Lcom/hoc081098/flowext/NullValue;
-	public final fun unbox (Ljava/lang/Object;)Ljava/lang/Object;
+public final class com/hoc081098/flowext/NeverFlowKt {
+	public static final fun neverFlow ()Lkotlinx/coroutines/flow/Flow;
 }
 
 public final class com/hoc081098/flowext/RangeKt {

build.gradle.kts

@@ -6,7 +6,7 @@ import org.jetbrains.kotlin.gradle.plugin.mpp.NativeBuildType
 import java.net.URL
 
 plugins {
-  kotlin("multiplatform") version "1.6.0"
+  kotlin("multiplatform") version "1.6.10"
   id("com.diffplug.spotless") version "6.0.4"
   id("maven-publish")
   id("com.vanniktech.maven.publish") version "0.18.0"
@@ -23,7 +23,7 @@ repositories {
   gradlePluginPortal()
 }
 
-val kotlinCoroutinesVersion = "1.6.0-RC2"
+val kotlinCoroutinesVersion = "1.6.0"
 val ktlintVersion = "0.43.2"
 
 kotlin {
@@ -84,7 +84,7 @@ kotlin {
       dependencies {
         implementation(kotlin("test-common"))
         implementation(kotlin("test-annotations-common"))
-        implementation("org.jetbrains.kotlinx:atomicfu:0.17.0")
+        implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$kotlinCoroutinesVersion")
       }
     }
     val jvmMain by getting {

src/commonMain/kotlin/com/hoc081098/flowext/bufferCount.kt

@@ -1,6 +1,5 @@
 package com.hoc081098.flowext
 
-import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 
@@ -13,7 +12,6 @@ import kotlinx.coroutines.flow.flow
  * For example if [startBufferEvery] is 2, then a new buffer will be started on every other value from the source.
  * A new buffer is started at the beginning of the source by default.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T> Flow<T>.bufferCount(
   bufferSize: Int,
   startBufferEvery: Int? = null,
@@ -27,7 +25,6 @@ public fun <T> Flow<T>.bufferCount(
   }
 }
 
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 private fun <T> Flow<T>.bufferSkip(bufferSize: Int, skip: Int): Flow<List<T>> {
   return flow {
     val buffers = ArrayDeque<MutableList<T>>()
@@ -72,7 +69,6 @@ private fun <T> Flow<T>.bufferSkip(bufferSize: Int, skip: Int): Flow<List<T>> {
   }
 }
 
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 private fun <T> Flow<T>.bufferExact(bufferSize: Int): Flow<List<T>> {
   return flow {
     var buffer: MutableList<T> = mutableListOf()

src/commonMain/kotlin/com/hoc081098/flowext/flatMapFirst.kt

@@ -2,7 +2,6 @@ package com.hoc081098.flowext
 
 import com.hoc081098.flowext.internal.AtomicBoolean
 import kotlinx.coroutines.CoroutineStart
-import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.channelFlow
@@ -24,14 +23,12 @@ import kotlinx.coroutines.launch
  *
  * @param transform A transform function to apply to value that was observed while no Flow is executing in parallel.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T, R> Flow<T>.flatMapFirst(transform: suspend (value: T) -> Flow<R>): Flow<R> =
   map(transform).flattenFirst()
 
 /**
  * Converts a higher-order [Flow] into a first-order [Flow] by dropping inner [Flow] while the previous inner [Flow] has not yet completed.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T> Flow<Flow<T>>.flattenFirst(): Flow<T> = channelFlow {
   val busy = AtomicBoolean(false)
 
@@ -52,13 +49,11 @@ public fun <T> Flow<Flow<T>>.flattenFirst(): Flow<T> = channelFlow {
 /**
  * This function is an alias to [flatMapFirst] operator.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T, R> Flow<T>.exhaustMap(transform: suspend (value: T) -> Flow<R>): Flow<R> =
   flatMapFirst(transform)
 
 /**
  * This function is an alias to [flattenFirst] operator.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 @Suppress("NOTHING_TO_INLINE")
 public inline fun <T> Flow<Flow<T>>.exhaustAll(): Flow<T> = flattenFirst()

src/commonMain/kotlin/com/hoc081098/flowext/mapIndexed.kt

@@ -1,14 +1,12 @@
 package com.hoc081098.flowext
 
-import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 
 /**
  * Returns a flow containing the results of applying the given [transform] function
  * to each value and its index in the original flow.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T, R> Flow<T>.mapIndexed(transform: suspend (index: Int, value: T) -> R): Flow<R> =
   flow {
     var index = 0

src/commonMain/kotlin/com/hoc081098/flowext/materialize_dematerialize.kt

@@ -2,7 +2,6 @@ package com.hoc081098.flowext
 
 import com.hoc081098.flowext.internal.ClosedException
 import com.hoc081098.flowext.internal.checkOwnership
-import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.flow
@@ -19,7 +18,6 @@ public fun <T> Flow<T>.materialize(): Flow<Event<T>> = map<T, Event<T>> { Event.
 /**
  * Converts a [Flow] of [Event] objects into the emissions that they represent.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T> Flow<Event<T>>.dematerialize(): Flow<T> = flow {
   try {
     collect {

src/commonMain/kotlin/com/hoc081098/flowext/takeUntil.kt

@@ -3,7 +3,6 @@ package com.hoc081098.flowext
 import com.hoc081098.flowext.internal.ClosedException
 import com.hoc081098.flowext.internal.checkOwnership
 import kotlinx.coroutines.CoroutineStart
-import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.collect
@@ -17,7 +16,6 @@ import kotlinx.coroutines.launch
  * @param notifier The [Flow] whose first emitted value or complete event
  * will cause the output [Flow] of [takeUntil] to stop emitting values from the source [Flow].
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <T, R> Flow<T>.takeUntil(notifier: Flow<R>): Flow<T> = flow {
   try {
     coroutineScope {

src/commonMain/kotlin/com/hoc081098/flowext/withLatestFrom.kt

@@ -1,7 +1,6 @@
 package com.hoc081098.flowext
 
 import kotlinx.coroutines.CoroutineStart
-import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.onSuccess
 import kotlinx.coroutines.coroutineScope
@@ -16,7 +15,6 @@ import kotlinx.coroutines.launch
  * @param other Second [Flow]
  * @param transform A transform function to apply to each value from self combined with the latest value from the second [Flow], if any.
  */
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 public fun <A, B, R> Flow<A>.withLatestFrom(
   other: Flow<B>,
   transform: suspend (A, B) -> R
@@ -47,7 +45,6 @@ public fun <A, B, R> Flow<A>.withLatestFrom(
   }
 }
 
-@InternalCoroutinesApi // TODO: Remove InternalCoroutinesApi (https://github.com/Kotlin/kotlinx.coroutines/issues/3078)
 @Suppress("NOTHING_TO_INLINE")
 public inline fun <A, B> Flow<A>.withLatestFrom(other: Flow<B>): Flow<Pair<A, B>> =
   withLatestFrom(other) { a, b -> a to b }

src/commonTest/kotlin/com/hoc081098/flowext/BufferCountTest.kt

@@ -7,6 +7,9 @@ import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertContentEquals
 import kotlin.test.assertIs
@@ -15,7 +18,7 @@ import kotlin.test.assertIs
 @InternalCoroutinesApi
 class BufferCountTest {
   @Test
-  fun testBufferCount_shouldEmitBuffersAtBufferSize() = suspendTest {
+  fun testBufferCount_shouldEmitBuffersAtBufferSize() = runTest {
     range(0, 10)
       .bufferCount(3)
       .test(
@@ -30,7 +33,7 @@ class BufferCountTest {
   }
 
   @Test
-  fun testBufferCount_shouldEmitBuffersAtBufferSizeWithStartBufferEvery() = suspendTest {
+  fun testBufferCount_shouldEmitBuffersAtBufferSizeWithStartBufferEvery() = runTest {
     range(0, 8)
       .bufferCount(3, 1)
       .test(
@@ -60,7 +63,7 @@ class BufferCountTest {
   }
 
   @Test
-  fun testBufferCount_shouldThrowExceptionWithFailureUpStream() = suspendTest {
+  fun testBufferCount_shouldThrowExceptionWithFailureUpStream() = runTest {
     flow<Int> { throw RuntimeException("Broken!") }
       .bufferCount(2)
       .test(null) {
@@ -75,7 +78,7 @@ class BufferCountTest {
   }
 
   @Test
-  fun testBufferCount_testCancellation() = suspendTest {
+  fun testBufferCount_testCancellation() = runTest {
     range(0, 10)
       .bufferCount(4)
       .take(2)
@@ -89,7 +92,7 @@ class BufferCountTest {
   }
 
   @Test
-  fun testBufferCount_testCancellationWithStartBufferEvery() = suspendTest {
+  fun testBufferCount_testCancellationWithStartBufferEvery() = runTest {
     range(0, 10)
       .bufferCount(4, 2)
       .take(2)
@@ -103,22 +106,27 @@ class BufferCountTest {
   }
 
   @Test
-  fun testBufferCount_shouldBufferProperly() = suspendTest {
+  fun testBufferCount_shouldBufferProperly() = runTest {
     val flow = MutableSharedFlow<Int>(extraBufferCapacity = 64)
 
     val results = mutableListOf<List<Int>>()
-    val job = flow.bufferCount(3, 1).onEach {
+    val job1 = flow.bufferCount(3, 1).onEach {
       results += it
       if (it == listOf(1, 2, 3)) {
         flow.tryEmit(4)
       }
     }.launchIn(this)
 
-    flow.tryEmit(1)
-    flow.tryEmit(2)
-    flow.tryEmit(3)
+    val job2 = launch {
+      flow.tryEmit(1)
+      flow.tryEmit(2)
+      flow.tryEmit(3)
+    }
+
+    advanceUntilIdle()
+    job1.cancel()
+    job2.cancel()
 
-    job.cancel()
     assertContentEquals(
       results,
       listOf(

src/commonTest/kotlin/com/hoc081098/flowext/ConcatTest.kt

@@ -4,6 +4,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.InternalCoroutinesApi
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertIs
@@ -12,7 +13,7 @@ import kotlin.test.assertIs
 @InternalCoroutinesApi
 class ConcatTest {
   @Test
-  fun testConcat_shouldEmitValuesFromMultipleFlows() = suspendTest {
+  fun testConcat_shouldEmitValuesFromMultipleFlows() = runTest {
     concat(
       flow1 = flowOf(1, 2, 3),
       flow2 = flowOf(4, 5, 6),
@@ -72,7 +73,7 @@ class ConcatTest {
   }
 
   @Test
-  fun testConcat_shouldConcatTheSameColdFlowMultipleTimes() = suspendTest {
+  fun testConcat_shouldConcatTheSameColdFlowMultipleTimes() = runTest {
     val flow = flowOf(1, 2, 3)
     val events = (1..3).map { Event.Value(it) }
 
@@ -135,7 +136,7 @@ class ConcatTest {
   }
 
   @Test
-  fun testConcat_firstFailureUpstream() = suspendTest {
+  fun testConcat_firstFailureUpstream() = runTest {
     val flow = flowOf(1, 2, 3)
     val failureFlow = flow<Nothing> { throw RuntimeException("Crash!") }
     val expectation: suspend (List<Event<Int>>) -> Unit = { events ->

src/commonTest/kotlin/com/hoc081098/flowext/DematerializeTest.kt

@@ -7,14 +7,15 @@ import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertFailsWith
 
 @ExperimentalCoroutinesApi
 @InternalCoroutinesApi
 class DematerializeTest {
   @Test
-  fun testDematerialize_shouldDematerializeAHappyFlow() = suspendTest {
+  fun testDematerialize_shouldDematerializeAHappyFlow() = runTest {
     flowOf(1, 2, 3)
       .materialize()
       .dematerialize()
@@ -67,7 +68,7 @@ class DematerializeTest {
   }
 
   @Test
-  fun testDematerialize_shouldDematerializeASadFlow() = suspendTest {
+  fun testDematerialize_shouldDematerializeASadFlow() = runTest {
     val ex = RuntimeException()
 
     flowOf(1, 2, 3)
@@ -115,7 +116,7 @@ class DematerializeTest {
   }
 
   @Test
-  fun testDematerialize_testCancellation() = suspendTest {
+  fun testDematerialize_testCancellation() = runTest {
     flowOf(1, 2, 3)
       .materialize()
       .dematerialize()