Make ChannelIterator.next non-suspending

* Mark ReceiveChannel.map with @ObsoleteCoroutinesApi

Fixes #1162

Author: qwwdfsad

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

@@ -561,7 +561,12 @@ public final class kotlinx/coroutines/channels/Channel$Factory {
 
 public abstract interface class kotlinx/coroutines/channels/ChannelIterator {
 	public abstract fun hasNext (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public abstract fun next (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun next ()Ljava/lang/Object;
+	public abstract synthetic fun next (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
+public final class kotlinx/coroutines/channels/ChannelIterator$DefaultImpls {
+	public static synthetic fun next (Lkotlinx/coroutines/channels/ChannelIterator;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/channels/ChannelKt {

kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt

@@ -901,15 +901,15 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         }
 
         @Suppress("UNCHECKED_CAST")
-        override suspend fun next(): E {
+        override fun next(): E {
             val result = this.result
             if (result is Closed<*>) throw recoverStackTrace(result.receiveException)
             if (result !== POLL_FAILED) {
                 this.result = POLL_FAILED
                 return result as E
             }
-            // rare case when hasNext was not invoked yet -- just delegate to receive (leave state as is)
-            return channel.receive()
+
+            throw IllegalStateException("'hasNext' should be called prior to 'next' invocation")
         }
     }
 

kotlinx-coroutines-core/common/src/channels/Channel.kt

@@ -295,25 +295,34 @@ public interface ChannelIterator<out E> {
      */
     public suspend operator fun hasNext(): Boolean
 
+    @Deprecated(message = "Since 1.3.0, binary compatibility with versions <= 1.2.x", level = DeprecationLevel.HIDDEN)
+    @Suppress("INAPPLICABLE_JVM_NAME")
+    @JvmName("next")
+    public suspend fun next0(): E {
+        /*
+         * Before 1.3.0 the "next()" could have been used without invoking "hasNext" first and there were code samples
+         * demonstrating this behavior, so we preserve this logic for full binary backwards compatibility with previously
+         * compiled code.
+         */
+        if (!hasNext()) throw ClosedReceiveChannelException(DEFAULT_CLOSE_MESSAGE)
+        return next()
+    }
+
     /**
-     * Retrieves and removes the element from this channel suspending the caller while this channel
-     * is empty or throws [ClosedReceiveChannelException] if the channel
-     * [isClosedForReceive][ReceiveChannel.isClosedForReceive] without cause.
-     * It throws the original [close][SendChannel.close] cause exception if the channel has _failed_.
-     *
-     * This suspending function is cancellable. If the [Job] of the current coroutine is cancelled or completed while this
-     * function is suspended, this function immediately resumes with [CancellationException].
-     *
-     * *Cancellation of suspended receive is atomic* -- when this function
-     * throws [CancellationException] it means that the element was not retrieved from this channel.
-     * As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may
-     * continue to execute even after it was cancelled from the same thread in the case when this receive operation
-     * was already resumed and the continuation was posted for execution to the thread's queue.
+     * Retrieves the element from the current iterator previously removed from the channel by preceding call to [hasNext] or
+     * throws [IllegalStateException] if [hasNext] was not invoked.
+     * [next] should only be used in pair with [hasNext]:
+     * ```
+     * while (iterator.hasNext()) {
+     *     val element = iterator.next()
+     *     // ... handle element ...
+     * }
+     * ```
      *
-     * Note that this function does not check for cancellation when it is not suspended.
-     * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
+     * This method throws [ClosedReceiveChannelException] if the channel [isClosedForReceive][ReceiveChannel.isClosedForReceive] without cause.
+     * It throws the original [close][SendChannel.close] cause exception if the channel has _failed_.
      */
-    public suspend operator fun next(): E
+    public operator fun next(): E
 }
 
 /**

kotlinx-coroutines-core/common/src/channels/Channels.common.kt

@@ -1196,7 +1196,7 @@ public suspend inline fun <E, K, V, M : MutableMap<in K, MutableList<V>>> Receiv
  * The operation is _intermediate_ and _stateless_.
  * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
  */
-// todo: mark transform with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
+@ObsoleteCoroutinesApi
 public fun <E, R> ReceiveChannel<E>.map(context: CoroutineContext = Dispatchers.Unconfined, transform: suspend (E) -> R): ReceiveChannel<R> =
     GlobalScope.produce(context, onCompletion = consumes()) {
         consumeEach {

kotlinx-coroutines-core/common/test/channels/BasicOperationsTest.kt

@@ -73,6 +73,18 @@ class BasicOperationsTest : TestBase() {
         finish(4)
     }
 
+    @Test
+    fun testIterator() = runTest {
+        TestChannelKind.values().forEach { kind ->
+            val channel = kind.create()
+            val iterator = channel.iterator()
+            assertFailsWith<IllegalStateException> { iterator.next() }
+            channel.close()
+            assertFailsWith<IllegalStateException> { iterator.next() }
+            assertFalse(iterator.hasNext())
+        }
+    }
+
     private suspend fun testReceiveOrNull(kind: TestChannelKind) = coroutineScope {
         val channel = kind.create()
         val d = async(NonCancellable) {