Improve documentation, rename await to awaitClose, improve awaitClose implementation

Author: qwwdfsad

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

@@ -719,8 +719,8 @@ public final class kotlinx/coroutines/channels/ConflatedBroadcastChannel : kotli
 }
 
 public final class kotlinx/coroutines/channels/ProduceKt {
-	public static final fun await (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public static synthetic fun await$default (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
+	public static final fun awaitClose (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static synthetic fun awaitClose$default (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
 	public static final fun produce (Lkotlinx/coroutines/CoroutineScope;Lkotlin/coroutines/CoroutineContext;ILkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/channels/ReceiveChannel;
 	public static final fun produce (Lkotlinx/coroutines/CoroutineScope;Lkotlin/coroutines/CoroutineContext;ILkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/channels/ReceiveChannel;
 	public static synthetic fun produce$default (Lkotlinx/coroutines/CoroutineScope;Lkotlin/coroutines/CoroutineContext;ILkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/channels/ReceiveChannel;

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

@@ -29,25 +29,29 @@ public interface ProducerScope<in E> : CoroutineScope, SendChannel<E> {
  * Suspends the current coroutine until the channel is either [closed][SendChannel.close] or [cancelled][ReceiveChannel.cancel]
  * and invokes the given [block] before resuming the coroutine.
  *
+ * Note that when producer channel is cancelled this function resumes with cancellation exception,
+ * so putting the code after calling this function would not lead to its execution in case of cancellation.
+ * That is why this code takes a lambda parameter.
+ *
  * Example of usage:
  * ```
  * val callbackEventsStream = produce {
  *     val disposable = registerChannelInCallback(channel)
- *     await { disposable.dispose() }
+ *     awaitClose { disposable.dispose() }
  * }
  * ```
  */
 @ExperimentalCoroutinesApi
-public suspend fun <T> ProducerScope<T>.await(block: () -> Unit = {}) {
-    check(kotlin.coroutines.coroutineContext[Job] === this) { "await() can be invoke only from the producer context" }
-    suspendCancellableCoroutine<Unit> { cont ->
-        invokeOnClose {
-            try {
-                block()
-            } finally {
+public suspend fun <T> ProducerScope<T>.awaitClose(block: () -> Unit = {}) {
+    check(kotlin.coroutines.coroutineContext[Job] === this) { "awaitClose() can be invoke only from the producer context" }
+    try {
+        suspendCancellableCoroutine<Unit> { cont ->
+            invokeOnClose {
                 cont.resume(Unit)
             }
         }
+    } finally {
+        block()
     }
 }
 

kotlinx-coroutines-core/common/src/flow/Builders.kt

@@ -222,7 +222,7 @@ public fun <T> flowViaChannel(
  * on the resulting flow.
  *
  * This builder ensures thread-safety and context preservation, thus the provided [ProducerScope] can be used concurrently from different contexts.
- * The resulting flow will complete as soon as [ProducerScope], to artificially prolong it [await] can be used.
+ * The resulting flow will complete as soon as [ProducerScope], to artificially prolong it [awaitClose] can be used.
  * For more detailed example please refer to [callbackFlow] documentation.
  *
  * To control backpressure, [bufferSize] is used and matches directly the `capacity` parameter of [Channel] factory.
@@ -271,7 +271,7 @@ public fun <T> channelFlow(bufferSize: Int = 16, @BuilderInference block: suspen
  * This builder ensures thread-safety and context preservation, thus the provided [ProducerScope] can be used from any context,
  * e.g. from the callback-based API. The flow completes as soon as its scope completes, thus if you are using channel from the
  * callback-based API, to artificially prolong scope lifetime and avoid memory-leaks related to unregistered resources,
- * [await] extension should be used. [await] argument will be invoked when either flow consumer cancels flow collection
+ * [awaitClose] extension should be used. [awaitClose] argument will be invoked when either flow consumer cancels flow collection
  * or when callback-based API invokes [SendChannel.close] manually.
  *
  * To control backpressure, [bufferSize] is used and matches directly the `capacity` parameter of [Channel] factory.
@@ -283,10 +283,12 @@ public fun <T> channelFlow(bufferSize: Int = 16, @BuilderInference block: suspen
  * fun flowFrom(api: CallbackBasedApi): Flow<T> = callbackFlow {
  *     val callback = object : Callback { // implementation of some callback interface
  *         override fun onNextValue(value: T) {
- *             offer(value) // Note: offer drops value when buffer is full
+ *             // Note: offer drops value when buffer is full
+ *             // Channel.UNLIMITED can be used to avoid overfill
+ *             offer(value)
  *         }
  *         override fun onApiError(cause: Throwable) {
- *             cancel("API Error", CancellationException(cause))
+ *             cancel(CancellationException("API Error", cause))
  *         }
  *         override fun onCompleted() = channel.close()
  *     }

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

@@ -99,7 +99,7 @@ class ProduceTest : TestBase() {
         val parent = Job()
         val channel = produce<Int>(parent) {
             expect(2)
-            await { expect(4) }
+            awaitClose { expect(4) }
         }
         expect(1)
         yield()
@@ -119,7 +119,7 @@ class ProduceTest : TestBase() {
                 expect(3)
                 this@produce.cancel()
             }
-            await { expect(4) }
+            awaitClose { expect(4) }
         }
         expect(1)
         parent.complete()
@@ -132,7 +132,7 @@ class ProduceTest : TestBase() {
         val parent = Job()
         produce<Int>(parent) {
             expect(2)
-            await { expect(4) }
+            awaitClose { expect(4) }
         }
         expect(1)
         yield()
@@ -145,7 +145,7 @@ class ProduceTest : TestBase() {
     fun testAwaitIllegalState() = runTest {
         val channel = produce<Int> {  }
         @Suppress("RemoveExplicitTypeArguments") // KT-31525
-        assertFailsWith<IllegalStateException> { (channel as ProducerScope<*>).await<Nothing>() }
+        assertFailsWith<IllegalStateException> { (channel as ProducerScope<*>).awaitClose<Nothing>() }
     }
 
     private suspend fun cancelOnCompletion(coroutineContext: CoroutineContext) = CoroutineScope(coroutineContext).apply {

kotlinx-coroutines-core/common/test/flow/channels/FlowCallbackTest.kt

@@ -38,7 +38,7 @@ class FlowCallbackTest : TestBase() {
                 close()
             }
             expect(1)
-            await()
+            awaitClose()
         }
 
         assertEquals(listOf(1), flow.toList())