Remove deprecated methods from the Java Flow integration module

The `kotlinx-coroutines-jdk9` module, being a copy of
`kotlinx-coroutines-reactive`, has its share of legacy APIs
defined. As this is a new module, there is no reason to
preserve them.

The changes include:
* `Migration.kt`, being a file completely dedicated to warnings
  about old API usage, has been removed.
* `IntegrationTest.kt` changed slightly so that it no longer uses
  the subscription channel API, which is deprecated.
* `Channel.kt` is now just an implementation detail. It does not
  expose any public methods.
* In particular, `Publisher<T>.collect` has been moved to
  `ReactiveFlow.kt` and is no longer inline, as it would expose
  `openSubscription`, which is deprecated.
* `PublisherSubscriptionSelectTest`, which tests use of `select`
  with the subscription channel API, is not included anymore.
* `Convert.kt` also has been removed altogether, having no
  non-deprecated methods.

Author: dkhalanskyjb

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

@@ -7,28 +7,10 @@ public final class kotlinx/coroutines/jdk9/AwaitKt {
 	public static final fun awaitSingle (Ljava/util/concurrent/Flow$Publisher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
-public final class kotlinx/coroutines/jdk9/ChannelKt {
-	public static final fun collect (Ljava/util/concurrent/Flow$Publisher;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public static final fun consumeEach (Ljava/util/concurrent/Flow$Publisher;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public static final fun openSubscription (Ljava/util/concurrent/Flow$Publisher;I)Lkotlinx/coroutines/channels/ReceiveChannel;
-	public static synthetic fun openSubscription$default (Ljava/util/concurrent/Flow$Publisher;IILjava/lang/Object;)Lkotlinx/coroutines/channels/ReceiveChannel;
-}
-
 public abstract interface class kotlinx/coroutines/jdk9/ContextInjector {
 	public abstract fun injectCoroutineContext (Ljava/util/concurrent/Flow$Publisher;Lkotlin/coroutines/CoroutineContext;)Ljava/util/concurrent/Flow$Publisher;
 }
 
-public final class kotlinx/coroutines/jdk9/ConvertKt {
-	public static final fun asPublisher (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/CoroutineContext;)Ljava/util/concurrent/Flow$Publisher;
-	public static synthetic fun asPublisher$default (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/CoroutineContext;ILjava/lang/Object;)Ljava/util/concurrent/Flow$Publisher;
-}
-
-public final class kotlinx/coroutines/jdk9/FlowKt {
-	public static final fun asFlow (Ljava/util/concurrent/Flow$Publisher;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun asFlow (Ljava/util/concurrent/Flow$Publisher;I)Lkotlinx/coroutines/flow/Flow;
-	public static final fun asPublisher (Lkotlinx/coroutines/flow/Flow;)Ljava/util/concurrent/Flow$Publisher;
-}
-
 public final class kotlinx/coroutines/jdk9/FlowSubscription : kotlinx/coroutines/AbstractCoroutine, java/util/concurrent/Flow$Subscription {
 	public final field flow Lkotlinx/coroutines/flow/Flow;
 	public final field subscriber Ljava/util/concurrent/Flow$Subscriber;
@@ -65,5 +47,6 @@ public final class kotlinx/coroutines/jdk9/PublisherCoroutine : kotlinx/coroutin
 public final class kotlinx/coroutines/jdk9/ReactiveFlowKt {
 	public static final fun asFlow (Ljava/util/concurrent/Flow$Publisher;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun asPublisher (Lkotlinx/coroutines/flow/Flow;)Ljava/util/concurrent/Flow$Publisher;
+	public static final fun collect (Ljava/util/concurrent/Flow$Publisher;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 

reactive/kotlinx-coroutines-jdk9/src/Channel.kt

@@ -24,27 +24,12 @@ import java.util.concurrent.Flow.*
  * }
  * ```
  */
-@Deprecated(
-    message = "Transforming publisher to channel is deprecated, use asFlow() instead",
-    level = DeprecationLevel.WARNING) // Will be error in 1.4
-public fun <T> Publisher<T>.openSubscription(request: Int = 1): ReceiveChannel<T> {
+internal 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.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.
- */
-public suspend inline fun <T> Publisher<T>.collect(action: (T) -> Unit) =
-    openSubscription().consumeEach(action)
-
 @Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER", "SubscriberImplementation")
 private class SubscriptionChannel<T>(
     private val request: Int

reactive/kotlinx-coroutines-jdk9/src/Convert.kt

@@ -37,6 +37,13 @@ public fun <T : Any> Publisher<T>.asFlow(): Flow<T> =
  */
 public fun <T : Any> Flow<T>.asPublisher(): Publisher<T> = FlowAsPublisher(this)
 
+/**
+ * Subscribes to this [Publisher] and performs the specified action for each received element.
+ * Cancels subscription if any exception happens during collect.
+ */
+public suspend fun <T> Publisher<T>.collect(action: (T) -> Unit) =
+    openSubscription().consumeEach(action)
+
 private class PublisherAsFlow<T : Any>(
         private val publisher: Publisher<T>,
         context: CoroutineContext = EmptyCoroutineContext,

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

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.jdk9
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.flowOn
 import org.hamcrest.MatcherAssert.*
 import org.hamcrest.core.*
 import org.junit.*
@@ -90,9 +91,8 @@ class IntegrationTest(
         assertThat(pub.awaitFirstOrElse { 0 }, IsEqual(1))
         assertIAE { pub.awaitSingle() }
         checkNumbers(n, pub)
-        val channel = pub.openSubscription()
-        checkNumbers(n, channel.asPublisher(ctx(coroutineContext)))
-        channel.cancel()
+        val flow = pub.asFlow()
+        checkNumbers(n, flow.flowOn(ctx(coroutineContext)).asPublisher())
     }
 
     @Test