Replace kotlinx.coroutines.reactive.FlowKt and kotlinx.coroutines.reactor.FlowKt with properly named files

Otherwise, it leads to a weird classname clash and forces to write FQ imports

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactive.txt

@@ -62,3 +62,8 @@ public final class kotlinx/coroutines/reactive/PublisherCoroutine : kotlinx/coro
 	public fun send (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
+public final class kotlinx/coroutines/reactive/ReactiveFlowKt {
+	public static final fun asFlow (Lorg/reactivestreams/Publisher;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun asPublisher (Lkotlinx/coroutines/flow/Flow;)Lorg/reactivestreams/Publisher;
+}
+

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactor.txt

@@ -36,6 +36,10 @@ public final class kotlinx/coroutines/reactor/ReactorContextKt {
 	public static final fun asCoroutineContext (Lreactor/util/context/Context;)Lkotlinx/coroutines/reactor/ReactorContext;
 }
 
+public final class kotlinx/coroutines/reactor/ReactorFlowKt {
+	public static final fun asFlux (Lkotlinx/coroutines/flow/Flow;)Lreactor/core/publisher/Flux;
+}
+
 public final class kotlinx/coroutines/reactor/SchedulerCoroutineDispatcher : kotlinx/coroutines/CoroutineDispatcher, kotlinx/coroutines/Delay {
 	public fun <init> (Lreactor/core/scheduler/Scheduler;)V
 	public fun delay (JLkotlin/coroutines/Continuation;)Ljava/lang/Object;

reactive/kotlinx-coroutines-reactive/src/FlowAsPublisher.kt

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+@file:JvmMultifileClass
+@file:JvmName("FlowKt")
+
+package kotlinx.coroutines.reactive
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import org.reactivestreams.*
+
+// Binary compatibility with Spring 5.2 RC
+@Deprecated(
+    message = "Replaced in favor of ReactiveFlow extension, please import kotlinx.coroutines.reactive.* instead of kotlinx.coroutines.reactive.FlowKt",
+    level = DeprecationLevel.ERROR
+)
+@JvmName("asFlow")
+public fun <T : Any> Publisher<T>.asFlowDeprecated(): Flow<T> = asFlow()
+
+// Binary compatibility with Spring 5.2 RC
+@Deprecated(
+    message = "Replaced in favor of ReactiveFlow extension, please import kotlinx.coroutines.reactive.* instead of kotlinx.coroutines.reactive.FlowKt",
+    level = DeprecationLevel.ERROR
+)
+@JvmName("asPublisher")
+public fun <T : Any> Flow<T>.asPublisherDeprecated(): Publisher<T> = asPublisher()
+
+@FlowPreview
+@Deprecated(
+    message = "batchSize parameter is deprecated, use .buffer() instead to control the backpressure",
+    level = DeprecationLevel.ERROR,
+    replaceWith = ReplaceWith("asFlow().buffer(batchSize)", imports = ["kotlinx.coroutines.flow.*"])
+)
+public fun <T : Any> Publisher<T>.asFlow(batchSize: Int): Flow<T> = asFlow().buffer(batchSize)

reactive/kotlinx-coroutines-reactive/src/Migration.kt

@@ -2,15 +2,14 @@
  * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-@file:JvmMultifileClass
-@file:JvmName("FlowKt")
-
 package kotlinx.coroutines.reactive
 
+import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.flow.internal.*
+import kotlinx.coroutines.intrinsics.*
 import org.reactivestreams.*
 import java.util.*
 import kotlin.coroutines.*
@@ -28,13 +27,11 @@ import kotlin.coroutines.*
 public fun <T : Any> Publisher<T>.asFlow(): Flow<T> =
     PublisherAsFlow(this, 1)
 
-@FlowPreview
-@Deprecated(
-    message = "batchSize parameter is deprecated, use .buffer() instead to control the backpressure",
-    level = DeprecationLevel.ERROR,
-    replaceWith = ReplaceWith("asFlow().buffer(batchSize)", imports = ["kotlinx.coroutines.flow.*"])
-)
-public fun <T : Any> Publisher<T>.asFlow(batchSize: Int): Flow<T> = asFlow().buffer(batchSize)
+/**
+ * Transforms the given flow to a spec-compliant [Publisher].
+ */
+@ExperimentalCoroutinesApi
+public fun <T : Any> Flow<T>.asPublisher(): Publisher<T> = FlowAsPublisher(this)
 
 private class PublisherAsFlow<T : Any>(
     private val publisher: Publisher<T>,
@@ -137,4 +134,94 @@ private val contextInjectors: List<ContextInjector> =
     ServiceLoader.load(ContextInjector::class.java, ContextInjector::class.java.classLoader).toList()
 
 private fun <T> Publisher<T>.injectCoroutineContext(coroutineContext: CoroutineContext) =
-    contextInjectors.fold(this) { pub, contextInjector -> contextInjector.injectCoroutineContext(pub, coroutineContext) }
+    contextInjectors.fold(this) { pub, contextInjector -> contextInjector.injectCoroutineContext(pub, coroutineContext) }
+
+
+/**
+ * Adapter that transforms [Flow] into TCK-complaint [Publisher].
+ * [cancel] invocation cancels the original flow.
+ */
+@Suppress("PublisherImplementation")
+private class FlowAsPublisher<T : Any>(private val flow: Flow<T>) : Publisher<T> {
+    override fun subscribe(subscriber: Subscriber<in T>?) {
+        if (subscriber == null) throw NullPointerException()
+        subscriber.onSubscribe(FlowSubscription(flow, subscriber))
+    }
+}
+
+/** @suppress */
+@InternalCoroutinesApi
+public class FlowSubscription<T>(
+    @JvmField val flow: Flow<T>,
+    @JvmField val subscriber: Subscriber<in T>
+) : Subscription, AbstractCoroutine<Unit>(Dispatchers.Unconfined, false) {
+    private val requested = atomic(0L)
+    private val producer = atomic<CancellableContinuation<Unit>?>(null)
+
+    override fun onStart() {
+        ::flowProcessing.startCoroutineCancellable(this)
+    }
+
+    private suspend fun flowProcessing() {
+        try {
+            consumeFlow()
+            subscriber.onComplete()
+        } catch (e: Throwable) {
+            try {
+                if (e is CancellationException) {
+                    subscriber.onComplete()
+                } else {
+                    subscriber.onError(e)
+                }
+            } catch (e: Throwable) {
+                // Last ditch report
+                handleCoroutineException(coroutineContext, e)
+            }
+        }
+    }
+
+    /*
+     * This method has at most one caller at any time (triggered from the `request` method)
+     */
+    private suspend fun consumeFlow() {
+        flow.collect { value ->
+            /*
+             * Flow is scopeless, thus if it's not active, its subscription was cancelled.
+             * No intermediate "child failed, but flow coroutine is not" states are allowed.
+             */
+            coroutineContext.ensureActive()
+            if (requested.value <= 0L) {
+                suspendCancellableCoroutine<Unit> {
+                    producer.value = it
+                    if (requested.value != 0L) it.resumeSafely()
+                }
+            }
+            requested.decrementAndGet()
+            subscriber.onNext(value)
+        }
+    }
+
+    override fun cancel() {
+        cancel(null)
+    }
+
+    override fun request(n: Long) {
+        if (n <= 0) {
+            return
+        }
+        start()
+        requested.update { value ->
+            val newValue = value + n
+            if (newValue <= 0L) Long.MAX_VALUE else newValue
+        }
+        val producer = producer.getAndSet(null) ?: return
+        producer.resumeSafely()
+    }
+
+    private fun CancellableContinuation<Unit>.resumeSafely() {
+        val token = tryResume(Unit)
+        if (token != null) {
+            completeResume(token)
+        }
+    }
+}

reactive/kotlinx-coroutines-reactive/src/PublisherAsFlow.kt renamed to reactive/kotlinx-coroutines-reactive/src/ReactiveFlow.kt

@@ -0,0 +1,13 @@
+@file:JvmName("FlowKt")
+
+package kotlinx.coroutines.reactor
+
+import kotlinx.coroutines.flow.*
+import reactor.core.publisher.*
+
+@Deprecated(
+    message = "Replaced in favor of ReactiveFlow extension, please import kotlinx.coroutines.reactor.* instead of kotlinx.coroutines.reactor.FlowKt",
+    level = DeprecationLevel.ERROR
+)
+@JvmName("asFlux")
+public fun <T : Any> Flow<T>.asFluxDeprecated(): Flux<T> = asFlux()

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

@@ -1,4 +1,6 @@
-@file:JvmName("FlowKt")
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
 
 package kotlinx.coroutines.reactor