Fix ReactorFlow integration bug exposed by TestBase

Author: qwwdfsad

reactive/kotlinx-coroutines-reactor/src/ReactorFlow.kt

@@ -21,7 +21,7 @@ public fun <T: Any> Flow<T>.asFlux(): Flux<T> = FlowAsFlux(this)
 private class FlowAsFlux<T : Any>(private val flow: Flow<T>) : Flux<T>() {
     override fun subscribe(subscriber: CoreSubscriber<in T>?) {
         if (subscriber == null) throw NullPointerException()
-        val hasContext = subscriber.currentContext().isEmpty
+        val hasContext = !subscriber.currentContext().isEmpty
         val source = if (hasContext) flow.flowOn(subscriber.currentContext().asCoroutineContext()) else flow
         subscriber.onSubscribe(FlowSubscription(source, subscriber))
     }

reactive/kotlinx-coroutines-reactor/test/FlowAsFluxTest.kt

@@ -10,18 +10,18 @@ import kotlin.test.assertEquals
 
 class FlowAsFluxTest : TestBase() {
     @Test
-    fun testFlowToFluxContextPropagation() = runBlocking<Unit> {
+    fun testFlowToFluxContextPropagation() {
         val flux = flow<String> {
-            (1..4).forEach { i -> emit(m(i).awaitFirst()) }
+            (1..4).forEach { i -> emit(createMono(i).awaitFirst()) }
         }   .asFlux()
             .subscriberContext(Context.of(1, "1"))
             .subscriberContext(Context.of(2, "2", 3, "3", 4, "4"))
-        var i = 0
-        flux.subscribe { str -> i++; println(str); assertEquals(str, i.toString()) }
+        val list = flux.collectList().block()!!
+        assertEquals(listOf("1", "2", "3", "4"), list)
     }
 
-    private fun m(i: Int): Mono<String> = mono {
-        val ctx = coroutineContext[ReactorContext]?.context
-        ctx?.getOrDefault(i, "noValue")
+    private fun createMono(i: Int): Mono<String> = mono {
+        val ctx = coroutineContext[ReactorContext]!!.context
+        ctx.getOrDefault(i, "noValue")
     }
 }

reactive/kotlinx-coroutines-reactor/test/FluxSingleTest.kt

@@ -12,6 +12,12 @@ import reactor.core.publisher.*
 import java.time.Duration.*
 
 class FluxSingleTest : TestBase() {
+
+    @Before
+    fun setup() {
+        ignoreLostThreads("parallel-")
+    }
+
     @Test
     fun testSingleNoWait() {
         val flux = flux {
@@ -167,7 +173,7 @@ class FluxSingleTest : TestBase() {
     @Test
     fun testExceptionFromCoroutine() {
         val flux = flux<String> {
-            error(Flux.just("O").awaitSingle() + "K")
+            throw IllegalStateException(Flux.just("O").awaitSingle() + "K")
         }
 
         checkErroneous(flux) {

reactive/kotlinx-coroutines-reactor/test/ReactorContextTest.kt

@@ -7,38 +7,40 @@ import kotlinx.coroutines.reactive.*
 import org.junit.Test
 import reactor.core.publisher.*
 import reactor.util.context.*
+import kotlin.coroutines.*
 import kotlin.test.*
 
 class ReactorContextTest : TestBase() {
     @Test
     fun testMonoHookedContext() = runBlocking {
         val mono = mono(Context.of(1, "1", 7, "7").asCoroutineContext()) {
-            val ctx = coroutineContext[ReactorContext]?.context
+            val ctx = reactorContext()
             buildString {
-                (1..7).forEach { append(ctx?.getOrDefault(it, "noValue")) }
+                (1..7).forEach { append(ctx.getOrDefault(it, "noValue")) }
             }
         }  .subscriberContext(Context.of(2, "2", 3, "3", 4, "4", 5, "5"))
            .subscriberContext { ctx -> ctx.put(6, "6") }
         assertEquals(mono.awaitFirst(), "1234567")
     }
 
     @Test
-    fun testFluxContext() = runBlocking<Unit> {
+    fun testFluxContext() {
         val flux = flux(Context.of(1, "1", 7, "7").asCoroutineContext()) {
-            val ctx = coroutineContext[ReactorContext]!!.context
+            val ctx = reactorContext()
             (1..7).forEach { send(ctx.getOrDefault(it, "noValue")) }
-        }   .subscriberContext(Context.of(2, "2", 3, "3", 4, "4", 5, "5"))
+        }
+            .subscriberContext(Context.of(2, "2", 3, "3", 4, "4", 5, "5"))
             .subscriberContext { ctx -> ctx.put(6, "6") }
-        var i = 0
-        flux.subscribe { str -> i++; assertEquals(str, i.toString()) }
+        val list = flux.collectList().block()!!
+        assertEquals((1..7).map { it.toString() }, list)
     }
 
     @Test
     fun testAwait() = runBlocking(Context.of(3, "3").asCoroutineContext()) {
         val result = mono(Context.of(1, "1").asCoroutineContext()) {
-            val ctx = coroutineContext[ReactorContext]?.context
+            val ctx = reactorContext()
             buildString {
-                (1..3).forEach { append(ctx?.getOrDefault(it, "noValue")) }
+                (1..3).forEach { append(ctx.getOrDefault(it, "noValue")) }
             }
         }  .subscriberContext(Context.of(2, "2"))
             .awaitFirst()
@@ -47,36 +49,34 @@ class ReactorContextTest : TestBase() {
 
     @Test
     fun testMonoAwaitContextPropagation() = runBlocking(Context.of(7, "7").asCoroutineContext()) {
-        assertEquals(m().awaitFirst(), "7")
-        assertEquals(m().awaitFirstOrDefault("noValue"), "7")
-        assertEquals(m().awaitFirstOrNull(), "7")
-        assertEquals(m().awaitFirstOrElse { "noValue" }, "7")
-        assertEquals(m().awaitLast(), "7")
-        assertEquals(m().awaitSingle(), "7")
+        assertEquals(createMono().awaitFirst(), "7")
+        assertEquals(createMono().awaitFirstOrDefault("noValue"), "7")
+        assertEquals(createMono().awaitFirstOrNull(), "7")
+        assertEquals(createMono().awaitFirstOrElse { "noValue" }, "7")
+        assertEquals(createMono().awaitLast(), "7")
+        assertEquals(createMono().awaitSingle(), "7")
     }
 
     @Test
     fun testFluxAwaitContextPropagation() = runBlocking<Unit>(
         Context.of(1, "1", 2, "2", 3, "3").asCoroutineContext()
     ) {
-        assertEquals(f().awaitFirst(), "1")
-        assertEquals(f().awaitFirstOrDefault("noValue"), "1")
-        assertEquals(f().awaitFirstOrNull(), "1")
-        assertEquals(f().awaitFirstOrElse { "noValue" }, "1")
-        assertEquals(f().awaitLast(), "3")
-        var i = 0
-        f().subscribe { str -> i++; assertEquals(str, i.toString()) }
+        assertEquals(createFlux().awaitFirst(), "1")
+        assertEquals(createFlux().awaitFirstOrDefault("noValue"), "1")
+        assertEquals(createFlux().awaitFirstOrNull(), "1")
+        assertEquals(createFlux().awaitFirstOrElse { "noValue" }, "1")
+        assertEquals(createFlux().awaitLast(), "3")
     }
 
-    private fun m(): Mono<String> = mono {
-        val ctx = coroutineContext[ReactorContext]?.context
-        ctx?.getOrDefault(7, "noValue")
+    private fun createMono(): Mono<String> = mono {
+        val ctx = reactorContext()
+        ctx.getOrDefault(7, "noValue")
     }
 
 
-    private fun f(): Flux<String?> = flux {
-        val ctx = coroutineContext[ReactorContext]?.context
-        (1..3).forEach { send(ctx?.getOrDefault(it, "noValue")) }
+    private fun createFlux(): Flux<String?> = flux {
+        val ctx = reactorContext()
+        (1..3).forEach { send(ctx.getOrDefault(it, "noValue")) }
     }
 
     @Test
@@ -95,17 +95,17 @@ class ReactorContextTest : TestBase() {
     fun testFlowToFluxDirectContextPropagation() = runBlocking(
         Context.of(1, "1", 2, "2", 3, "3").asCoroutineContext()
     ) {
-        var i = 0
         // convert resulting flow to channel using "produceIn"
         val channel = bar().produceIn(this)
-        channel.consumeEach { str ->
-            i++; assertEquals(str, i.toString())
-        }
-        assertEquals(i, 3)
+        val list = channel.toList()
+        assertEquals(listOf("1", "2", "3"), list)
     }
 
     private fun bar(): Flow<String> = flux {
-        val ctx = coroutineContext[ReactorContext]!!.context
+        val ctx = reactorContext()
         (1..3).forEach { send(ctx.getOrDefault(it, "noValue")) }
     }.asFlow()
+
+    private suspend fun reactorContext() =
+        coroutineContext[ReactorContext]!!.context
 }

reactive/kotlinx-coroutines-rx2/test/ObservableSingleTest.kt

@@ -11,6 +11,11 @@ import org.junit.Assert.*
 import java.util.concurrent.*
 
 class ObservableSingleTest : TestBase() {
+    @Before
+    fun setup() {
+        ignoreLostThreads("RxComputationThreadPool-", "RxCachedWorkerPoolEvictor-", "RxSchedulerPurge-")
+    }
+
     @Test
     fun testSingleNoWait() {
         val observable = rxObservable {
@@ -166,7 +171,7 @@ class ObservableSingleTest : TestBase() {
     @Test
     fun testExceptionFromCoroutine() {
         val observable = rxObservable<String> {
-            error(Observable.just("O").awaitSingle() + "K")
+            throw IllegalStateException(Observable.just("O").awaitSingle() + "K")
         }
 
         checkErroneous(observable) {