Promote reactive bridges for Flow to stable API (#1549)

qwwdfsad · web-flow · commit 835ed4da5a71 · 2019-09-19T17:29:04.000+03:00
* Promote reactive adapters for Flow to stable API
* Promote Publisher&lt;T&gt;.collect to stable API
* Promote rx2 extensions to stable, increase deprecation level for obsolete reactive primitives
diff --git a/reactive/kotlinx-coroutines-reactive/src/Channel.kt b/reactive/kotlinx-coroutines-reactive/src/Channel.kt
@@ -20,23 +20,21 @@ import org.reactivestreams.*
  * @param request how many items to request from publisher in advance (optional, one by default).
  */
 @ObsoleteCoroutinesApi
-@Suppress("CONFLICTING_OVERLOADS")
 public fun <T> Publisher<T>.openSubscription(request: Int = 1): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>(request)
     subscribe(channel)
     return channel
 }
 
 // Will be promoted to error in 1.3.0, removed in 1.4.0
-@Deprecated(message = "Use collect instead", level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.collect(action)"))
+@Deprecated(message = "Use collect instead", level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("this.collect(action)"))
 public suspend inline fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) =
     openSubscription().consumeEach(action)
 
 /**
  * Subscribes to this [Publisher] and performs the specified action for each received element.
  * Cancels subscription if any exception happens during collect.
  */
-@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
 public suspend inline fun <T> Publisher<T>.collect(action: (T) -> Unit) =
     openSubscription().consumeEach(action)
 
diff --git a/reactive/kotlinx-coroutines-reactive/src/Publish.kt b/reactive/kotlinx-coroutines-reactive/src/Publish.kt
@@ -48,7 +48,7 @@ public fun <T> publish(
 
 @Deprecated(
     message = "CoroutineScope.publish is deprecated in favour of top-level publish",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("publish(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0. Binary compatibility with Spring
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-reactive/src/ReactiveFlow.kt b/reactive/kotlinx-coroutines-reactive/src/ReactiveFlow.kt
@@ -23,14 +23,12 @@ import kotlin.coroutines.*
  * If any of the resulting flow transformations fails, subscription is immediately cancelled and all in-flights elements
  * are discarded.
  */
-@ExperimentalCoroutinesApi
 public fun <T : Any> Publisher<T>.asFlow(): Flow<T> =
     PublisherAsFlow(this, 1)
 
 /**
  * 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>(
diff --git a/reactive/kotlinx-coroutines-reactor/src/Flux.kt b/reactive/kotlinx-coroutines-reactor/src/Flux.kt
@@ -50,7 +50,7 @@ public fun <T> flux(
 
 @Deprecated(
     message = "CoroutineScope.flux is deprecated in favour of top-level flux",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("flux(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0. Binary compatibility with Spring
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-reactor/src/Mono.kt b/reactive/kotlinx-coroutines-reactor/src/Mono.kt
@@ -39,7 +39,7 @@ public fun <T> mono(
 
 @Deprecated(
     message = "CoroutineScope.mono is deprecated in favour of top-level mono",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("mono(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-reactor/src/ReactorFlow.kt b/reactive/kotlinx-coroutines-reactor/src/ReactorFlow.kt
@@ -4,7 +4,6 @@
 
 package kotlinx.coroutines.reactor
 
-import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.reactive.FlowSubscription
@@ -15,7 +14,6 @@ import reactor.core.publisher.Flux
  * Converts the given flow to a cold flux.
  * The original flow is cancelled when the flux subscriber is disposed.
  */
-@ExperimentalCoroutinesApi
 public fun <T: Any> Flow<T>.asFlux(): Flux<T> = FlowAsFlux(this)
 
 private class FlowAsFlux<T : Any>(private val flow: Flow<T>) : Flux<T>() {
diff --git a/reactive/kotlinx-coroutines-reactor/test/FluxMultiTest.kt b/reactive/kotlinx-coroutines-reactor/test/FluxMultiTest.kt
@@ -73,7 +73,7 @@ class FluxMultiTest : TestBase() {
         val mono = mono {
             var result = ""
             try {
-                flux.consumeEach { result += it }
+                flux.collect { result += it }
             } catch(e: IOException) {
                 result += e.message
             }
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxChannel.kt b/reactive/kotlinx-coroutines-rx2/src/RxChannel.kt
@@ -19,7 +19,6 @@ import kotlinx.coroutines.internal.*
  *           See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
  */
 @ObsoleteCoroutinesApi
-@Suppress("CONFLICTING_OVERLOADS")
 public fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
     subscribe(channel)
@@ -34,36 +33,33 @@ public fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
  *           See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
  */
 @ObsoleteCoroutinesApi
-@Suppress("CONFLICTING_OVERLOADS")
 public fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
     subscribe(channel)
     return channel
 }
 
 // Will be promoted to error in 1.3.0, removed in 1.4.0
-@Deprecated(message = "Use collect instead", level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.collect(action)"))
+@Deprecated(message = "Use collect instead", level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("this.collect(action)"))
 public suspend inline fun <T> MaybeSource<T>.consumeEach(action: (T) -> Unit) =
     openSubscription().consumeEach(action)
 
 // Will be promoted to error in 1.3.0, removed in 1.4.0
-@Deprecated(message = "Use collect instead", level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.collect(action)"))
+@Deprecated(message = "Use collect instead", level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("this.collect(action)"))
 public suspend inline fun <T> ObservableSource<T>.consumeEach(action: (T) -> Unit) =
     openSubscription().consumeEach(action)
 
 /**
  * Subscribes to this [MaybeSource] and performs the specified action for each received element.
  * Cancels subscription if any exception happens during collect.
  */
-@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
 public suspend inline fun <T> MaybeSource<T>.collect(action: (T) -> Unit) =
     openSubscription().consumeEach(action)
 
 /**
  * Subscribes to this [ObservableSource] and performs the specified action for each received element.
  * Cancels subscription if any exception happens during collect.
  */
-@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
 public suspend inline fun <T> ObservableSource<T>.collect(action: (T) -> Unit) =
     openSubscription().consumeEach(action)
 
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxCompletable.kt b/reactive/kotlinx-coroutines-rx2/src/RxCompletable.kt
@@ -36,7 +36,7 @@ public fun rxCompletable(
 
 @Deprecated(
     message = "CoroutineScope.rxCompletable is deprecated in favour of top-level rxCompletable",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("rxCompletable(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxFlowable.kt b/reactive/kotlinx-coroutines-rx2/src/RxFlowable.kt
@@ -45,7 +45,7 @@ public fun <T: Any> rxFlowable(
 
 @Deprecated(
     message = "CoroutineScope.rxFlowable is deprecated in favour of top-level rxFlowable",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("rxFlowable(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxMaybe.kt b/reactive/kotlinx-coroutines-rx2/src/RxMaybe.kt
@@ -37,7 +37,7 @@ public fun <T> rxMaybe(
 
 @Deprecated(
     message = "CoroutineScope.rxMaybe is deprecated in favour of top-level rxMaybe",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("rxMaybe(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxObservable.kt b/reactive/kotlinx-coroutines-rx2/src/RxObservable.kt
@@ -46,7 +46,7 @@ public fun <T : Any> rxObservable(
 
 @Deprecated(
     message = "CoroutineScope.rxObservable is deprecated in favour of top-level rxObservable",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("rxObservable(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt b/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt
@@ -36,7 +36,7 @@ public fun <T : Any> rxSingle(
 
 @Deprecated(
     message = "CoroutineScope.rxSingle is deprecated in favour of top-level rxSingle",
-    level = DeprecationLevel.WARNING,
+    level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("rxSingle(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
 @LowPriorityInOverloadResolution
diff --git a/reactive/kotlinx-coroutines-rx2/test/CompletableTest.kt b/reactive/kotlinx-coroutines-rx2/test/CompletableTest.kt
@@ -150,7 +150,7 @@ class CompletableTest : TestBase() {
 
     @Test
     fun testFatalExceptionInSubscribe() = runTest {
-        GlobalScope.rxCompletable(Dispatchers.Unconfined + CoroutineExceptionHandler{ _, e -> assertTrue(e is LinkageError); expect(2)}) {
+        rxCompletable(Dispatchers.Unconfined + CoroutineExceptionHandler{ _, e -> assertTrue(e is LinkageError); expect(2)}) {
             expect(1)
             42
         }.subscribe({ throw LinkageError() })
@@ -159,7 +159,7 @@ class CompletableTest : TestBase() {
 
     @Test
     fun testFatalExceptionInSingle() = runTest {
-        GlobalScope.rxCompletable(Dispatchers.Unconfined) {
+        rxCompletable(Dispatchers.Unconfined) {
             throw LinkageError()
         }.subscribe({ expectUnreached()  }, { expect(1); assertTrue(it is LinkageError) })
         finish(2)
diff --git a/reactive/kotlinx-coroutines-rx2/test/MaybeTest.kt b/reactive/kotlinx-coroutines-rx2/test/MaybeTest.kt
@@ -280,7 +280,7 @@ class MaybeTest : TestBase() {
 
     @Test
     fun testFatalExceptionInSubscribe() = runTest {
-        GlobalScope.rxMaybe(Dispatchers.Unconfined + CoroutineExceptionHandler{ _, e -> assertTrue(e is LinkageError); expect(2)}) {
+        rxMaybe(Dispatchers.Unconfined + CoroutineExceptionHandler{ _, e -> assertTrue(e is LinkageError); expect(2)}) {
             expect(1)
             42
         }.subscribe({ throw LinkageError() })
@@ -289,7 +289,7 @@ class MaybeTest : TestBase() {
 
     @Test
     fun testFatalExceptionInSingle() = runTest {
-        GlobalScope.rxMaybe(Dispatchers.Unconfined) {
+        rxMaybe(Dispatchers.Unconfined) {
             throw LinkageError()
         }.subscribe({ expectUnreached()  }, { expect(1); assertTrue(it is LinkageError) })
         finish(2)
diff --git a/reactive/kotlinx-coroutines-rx2/test/ObservableMultiTest.kt b/reactive/kotlinx-coroutines-rx2/test/ObservableMultiTest.kt
@@ -18,7 +18,7 @@ class ObservableMultiTest : TestBase() {
     @Test
     fun testNumbers() {
         val n = 100 * stressTestMultiplier
-        val observable = GlobalScope.rxObservable {
+        val observable = rxObservable {
             repeat(n) { send(it) }
         }
         checkSingleValue(observable.toList()) { list ->
@@ -30,7 +30,7 @@ class ObservableMultiTest : TestBase() {
     @Test
     fun testConcurrentStress() {
         val n = 10_000 * stressTestMultiplier
-        val observable = GlobalScope.rxObservable {
+        val observable = rxObservable {
             newCoroutineContext(coroutineContext)
             // concurrent emitters (many coroutines)
             val jobs = List(n) {
@@ -51,7 +51,7 @@ class ObservableMultiTest : TestBase() {
     @Test
     fun testIteratorResendUnconfined() {
         val n = 10_000 * stressTestMultiplier
-        val observable = GlobalScope.rxObservable(Dispatchers.Unconfined) {
+        val observable = rxObservable(Dispatchers.Unconfined) {
             Observable.range(0, n).collect { send(it) }
         }
         checkSingleValue(observable.toList()) { list ->
@@ -62,7 +62,7 @@ class ObservableMultiTest : TestBase() {
     @Test
     fun testIteratorResendPool() {
         val n = 10_000 * stressTestMultiplier
-        val observable = GlobalScope.rxObservable {
+        val observable = rxObservable {
             Observable.range(0, n).collect { send(it) }
         }
         checkSingleValue(observable.toList()) { list ->
@@ -72,14 +72,14 @@ class ObservableMultiTest : TestBase() {
 
     @Test
     fun testSendAndCrash() {
-        val observable = GlobalScope.rxObservable {
+        val observable = rxObservable {
             send("O")
             throw IOException("K")
         }
         val single = rxSingle {
             var result = ""
             try {
-                observable.consumeEach { result += it }
+                observable.collect { result += it }
             } catch(e: IOException) {
                 result += e.message
             }
diff --git a/reactive/kotlinx-coroutines-rx2/test/ObservableSingleTest.kt b/reactive/kotlinx-coroutines-rx2/test/ObservableSingleTest.kt
diff --git a/reactive/kotlinx-coroutines-rx2/test/SingleTest.kt b/reactive/kotlinx-coroutines-rx2/test/SingleTest.kt

Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,7 @@ class FluxMultiTest : TestBase() {`
`73`	`73`	`val mono = mono {`
`74`	`74`	`var result = ""`
`75`	`75`	`try {`
`76`		`- flux.consumeEach { result += it }`
	`76`	`+ flux.collect { result += it }`
`77`	`77`	`} catch(e: IOException) {`
`78`	`78`	`result += e.message`
`79`	`79`	`}`