~fixup replacement for ReceiveChannel.asObservable

Author: qwwdfsad

reactive/kotlinx-coroutines-rx2/src/RxConvert.kt

@@ -137,8 +137,9 @@ public fun <T: Any> Flow<T>.asFlowable(context: CoroutineContext = EmptyCoroutin
 
 @Deprecated(
     message = "Deprecated in the favour of Flow",
-    level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("this.consumeAsFlow().asObservable()") // Deprecated since 1.4.0
-)
+    level = DeprecationLevel.ERROR,
+    replaceWith = ReplaceWith("this.consumeAsFlow().asObservable(context)", "kotlinx.coroutines.flow.consumeAsFlow")
+) // Deprecated since 1.4.0
 public fun <T : Any> ReceiveChannel<T>.asObservable(context: CoroutineContext): Observable<T> = rxObservable(context) {
     for (t in this@asObservable)
         send(t)

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

@@ -6,6 +6,7 @@ package kotlinx.coroutines.rx2
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.flow.*
 import org.junit.Assert
 import org.junit.Test
 import kotlin.test.*
@@ -126,7 +127,7 @@ class ConvertTest : TestBase() {
             delay(50)
             send("K")
         }
-        val observable = c.asObservable(Dispatchers.Unconfined)
+        val observable = c.consumeAsFlow().asObservable(Dispatchers.Unconfined)
         checkSingleValue(observable.reduce { t1, t2 -> t1 + t2 }.toSingle()) {
             assertEquals("OK", it)
         }
@@ -140,7 +141,7 @@ class ConvertTest : TestBase() {
             delay(50)
             throw TestException("K")
         }
-        val observable = c.asObservable(Dispatchers.Unconfined)
+        val observable = c.consumeAsFlow().asObservable(Dispatchers.Unconfined)
         val single = rxSingle(Dispatchers.Unconfined) {
             var result = ""
             try {
@@ -155,4 +156,4 @@ class ConvertTest : TestBase() {
             assertEquals("OK", it)
         }
     }
-}
+}

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

@@ -6,6 +6,7 @@ package kotlinx.coroutines.rx2
 
 import io.reactivex.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
 import org.junit.Test
 import org.junit.runner.*
 import org.junit.runners.*
@@ -92,7 +93,7 @@ class IntegrationTest(
         assertFailsWith<IllegalArgumentException> { observable.awaitSingle() }
         checkNumbers(n, observable)
         val channel = observable.openSubscription()
-        checkNumbers(n, channel.asObservable(ctx(coroutineContext)))
+        checkNumbers(n, channel.consumeAsFlow().asObservable(ctx(coroutineContext)))
         channel.cancel()
     }
 
@@ -131,4 +132,4 @@ class IntegrationTest(
         assertEquals(n, last)
     }
 
-}
+}