Merge pull request #2708 from Kotlin/version-1.5.0

Version 1.5.0

Author: qwwdfsad

CHANGES.md

@@ -1,5 +1,48 @@
 # Change log for kotlinx.coroutines
 
+## Version 1.5.0
+
+Note that this is a full changelog relative to 1.4.3 version. Changelog relative to 1.5.0-RC can be found in the end.
+
+### Channels API
+
+* Major channels API rework (#330, #974). Existing `offer`, `poll`, and `sendBlocking` methods are deprecated, internal `receiveCatching` and `onReceiveCatching` removed, `receiveOrNull` and `onReceiveOrNull` are completely deprecated. Previously deprecated `SendChannel.isFull` declaration is removed. Channel operators deprecated with `ERROR` are now `HIDDEN`.
+* New methods `receiveCatching`, `onReceiveCatching` `trySend`, `tryReceive`, and `trySendBlocking` along with the new result type `ChannelResult` are introduced. They provide better type safety, are less error-prone, and have a consistent future-proof naming scheme.  The full rationale behind this change can be found [here](https://github.com/Kotlin/kotlinx.coroutines/issues/974#issuecomment-806569582).
+* `BroadcastChannel` and `ConflatedBroadcastChannel` are marked as `ObsoleteCoroutinesApi` in the favor or `SharedFlow` and `StateFlow`. The migration scheme can be found in their documentation. These classes will be deprecated in the next major release.
+* `callbackFlow` and `channelFlow` are promoted to stable API.
+
+### Reactive integrations
+
+* All existing API in modules `kotlinx-coroutines-rx2`, `kotlinx-coroutines-rx3`, `kotlinx-coroutines-reactive`, `kotlinx-coroutines-reactor`, and `kotlinx-coroutines-jdk9` were revisited and promoted to stable (#2545).
+* `publish` is no longer allowed to emit `null` values (#2646).
+* Misleading `awaitSingleOr*` functions on `Publisher` type are deprecated (#2591).
+* `MaybeSource.await` is deprecated in the favor of `awaitSingle`, additional lint functions for `Mono` are added in order to prevent ambiguous `Publisher` usages (#2628, #1587).
+* `ContextView` support in `kotlinx-coroutines-reactor` (#2575).
+* All reactive builders no longer ignore inner cancellation exceptions preventing their completion (#2262, #2646).
+* `MaybeSource.collect` and `Maybe.collect` properly finish when they are completed without a value (#2617).
+* All exceptions are now consistently handled according to reactive specification, whether they are considered 'fatal' or not by reactive frameworks (#2646).
+
+### Other improvements
+
+* `Flow.last` and `Flow.lastOrNull` operators (#2246).
+* `Flow.runningFold` operator (#2641).
+* `CoroutinesTimeout` rule for JUnit5 (#2197).
+* Internals of `Job` and `AbstractCoroutine` was reworked, resulting in smaller code size, less memory footprint, and better performance (#2513, #2512).
+* `CancellationException` from Kotlin standard library is used for cancellation on Koltin/JS and Kotlin/Native (#2638).
+* Introduced new `DelicateCoroutineApi` annotation that warns users about potential target API pitfalls and suggests studying API's documentation first. The only delicate API right now is `GlobalScope` (#2637).
+* Fixed bug introduced in `1.4.3` when `kotlinx-coroutines-core.jar` triggered IDEA debugger failure (#2619).
+* Fixed memory leak of `ChildHandlerNode` with reusable continuations (#2564).
+* Various documentation improvements (#2555, #2589, #2592, #2583, #2437, #2616, #2633, #2560).
+
+### Changelog relative to version 1.5.0-RC
+
+* Fail-fast during `emitAll` called from cancelled `onCompletion` operator (#2700).
+* Flows returned by `stateIn`/`shareIn` keep strong reference to sharing job (#2557).
+* Rename internal `TimeSource` to `AbstractTimeSource` due to import issues (#2691).
+* Reverted the change that triggered IDEA coroutines debugger crash (#2695, reverted #2291).
+* `watchosX64` target support for Kotlin/Native (#2524).
+* Various documentation fixes and improvements.
+
 ## Version 1.5.0-RC
 
 ### Channels API

README.md

@@ -2,7 +2,7 @@
 
 [![official JetBrains project](https://jb.gg/badges/official.svg)](https://confluence.jetbrains.com/display/ALL/JetBrains+on+GitHub)
 [![GitHub license](https://img.shields.io/badge/license-Apache%20License%202.0-blue.svg?style=flat)](https://www.apache.org/licenses/LICENSE-2.0)
-[![Download](https://img.shields.io/maven-central/v/org.jetbrains.kotlinx/kotlinx-coroutines-core/1.5.0-RC)](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core/1.5.0-RC/pom)
+[![Download](https://img.shields.io/maven-central/v/org.jetbrains.kotlinx/kotlinx-coroutines-core/1.5.0)](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core/1.5.0/pom)
 [![Kotlin](https://img.shields.io/badge/kotlin-1.5.0-blue.svg?logo=kotlin)](http://kotlinlang.org)
 [![Slack channel](https://img.shields.io/badge/chat-slack-green.svg?logo=slack)](https://kotlinlang.slack.com/messages/coroutines/)
 
@@ -83,7 +83,7 @@ Add dependencies (you can also add other modules that you need):
 <dependency>
     <groupId>org.jetbrains.kotlinx</groupId>
     <artifactId>kotlinx-coroutines-core</artifactId>
-    <version>1.5.0-RC</version>
+    <version>1.5.0</version>
 </dependency>
 ```
 
@@ -101,7 +101,7 @@ Add dependencies (you can also add other modules that you need):
 
 ```groovy
 dependencies {
-    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.0-RC'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.0'
 }
 ```
 
@@ -127,7 +127,7 @@ Add dependencies (you can also add other modules that you need):
 
 ```groovy
 dependencies {
-    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.0-RC")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.0")
 }
 ```
 
@@ -147,7 +147,7 @@ Add [`kotlinx-coroutines-android`](ui/kotlinx-coroutines-android)
 module as a dependency when using `kotlinx.coroutines` on Android:
 
 ```groovy
-implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.5.0-RC'
+implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.5.0'
 ```
 
 This gives you access to the Android [Dispatchers.Main]
@@ -180,7 +180,7 @@ In common code that should get compiled for different platforms, you can add a d
 ```groovy
 commonMain {
     dependencies {
-        implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.0-RC")
+        implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.0")
     }
 }
 ```
@@ -192,7 +192,7 @@ Platform-specific dependencies are recommended to be used only for non-multiplat
 #### JS
 
 Kotlin/JS version of `kotlinx.coroutines` is published as 
-[`kotlinx-coroutines-core-js`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-js/1.5.0-RC/jar)
+[`kotlinx-coroutines-core-js`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-js/1.5.0/jar)
 (follow the link to get the dependency declaration snippet) and as [`kotlinx-coroutines-core`](https://www.npmjs.com/package/kotlinx-coroutines-core) NPM package. 
 
 #### Native

docs/images/after.png

@@ -3,7 +3,7 @@
 #
 
 # Kotlin
-version=1.5.0-RC-SNAPSHOT
+version=1.5.0-SNAPSHOT
 group=org.jetbrains.kotlinx
 kotlin_version=1.5.0
 

docs/images/before.png

@@ -24,6 +24,7 @@ kotlin {
         addTarget(presets.watchosArm32)
         addTarget(presets.watchosArm64)
         addTarget(presets.watchosX86)
+        addTarget(presets.watchosX64)
     }
 
     sourceSets {

gradle.properties

@@ -302,9 +302,18 @@ public interface ReceiveChannel<out E> {
      * * Its name was not aligned with the rest of the API and tried to mimic Java's queue instead.
      *
      * See https://github.com/Kotlin/kotlinx.coroutines/issues/974 for more context.
+     *
+     * ### Replacement note
+     *
+     * The replacement `tryReceive().getOrNull()` is a default that ignores all close exceptions and
+     * proceeds with `null`, while `poll` throws an exception if the channel was closed with an exception.
+     * Replacement with the very same 'poll' semantics is `tryReceive().onClosed { if (it != null) throw it }.getOrNull()`
      */
-    @Deprecated(level = DeprecationLevel.WARNING,
-        message = "Deprecated in the favour of 'tryReceive'",
+    @Deprecated(
+        level = DeprecationLevel.WARNING,
+        message = "Deprecated in the favour of 'tryReceive'. " +
+            "Please note that the provided replacement does not rethrow channel's close cause as 'poll' did, " +
+            "for the precise replacement please refer to the 'poll' documentation",
         replaceWith = ReplaceWith("tryReceive().getOrNull()")
     ) // Warning since 1.5.0
     public fun poll(): E? {
@@ -322,12 +331,18 @@ public interface ReceiveChannel<out E> {
      * - Was throwing if the channel has failed even though its signature may suggest it returns 'null'
      * - It didn't really belong to core channel API and can be exposed as an extension instead.
      *
-     * @suppress doc
+     * ### Replacement note
+     *
+     * The replacement `receiveCatching().getOrNull()` is a safe default that ignores all close exceptions and
+     * proceeds with `null`, while `receiveOrNull` throws an exception if the channel was closed with an exception.
+     * Replacement with the very same `receiveOrNull` semantics is `receiveCatching().onClosed { if (it != null) throw it }.getOrNull()`.
      */
     @Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")
     @LowPriorityInOverloadResolution
     @Deprecated(
-        message = "Deprecated in favor of receiveCatching",
+        message = "Deprecated in favor of 'receiveCatching'. " +
+            "Please note that the provided replacement does not rethrow channel's close cause as 'receiveOrNull' did, " +
+            "for the detailed replacement please refer to the 'receiveOrNull' documentation",
         level = DeprecationLevel.ERROR,
         replaceWith = ReplaceWith("receiveCatching().getOrNull()")
     ) // Warning since 1.3.0, error in 1.5.0, will be hidden in 1.6.0

gradle/compile-native-multiplatform.gradle

@@ -30,6 +30,7 @@ public suspend fun <T> FlowCollector<T>.emitAll(channel: ReceiveChannel<T>): Uni
     emitAllImpl(channel, consume = true)
 
 private suspend fun <T> FlowCollector<T>.emitAllImpl(channel: ReceiveChannel<T>, consume: Boolean) {
+    ensureActive()
     // Manually inlined "consumeEach" implementation that does not use iterator but works via "receiveCatching".
     // It has smaller and more efficient spilled state which also allows to implement a manual kludge to
     // fix retention of the last emitted value.

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

@@ -194,7 +194,15 @@ public fun <T> Flow<T>.onEmpty(
     }
 }
 
-private class ThrowingCollector(private val e: Throwable) : FlowCollector<Any?> {
+/*
+ * 'emitAll' methods call this to fail-fast before starting to collect
+ * their sources (that may not have any elements for a long time).
+ */
+internal fun FlowCollector<*>.ensureActive() {
+    if (this is ThrowingCollector) throw e
+}
+
+internal class ThrowingCollector(@JvmField val e: Throwable) : FlowCollector<Any?> {
     override suspend fun emit(value: Any?) {
         throw e
     }

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

@@ -144,8 +144,8 @@ public fun <T> Flow<T>.shareIn(
         onBufferOverflow = config.onBufferOverflow
     )
     @Suppress("UNCHECKED_CAST")
-    scope.launchSharing(config.context, config.upstream, shared, started, NO_VALUE as T)
-    return shared.asSharedFlow()
+    val job = scope.launchSharing(config.context, config.upstream, shared, started, NO_VALUE as T)
+    return ReadonlySharedFlow(shared, job)
 }
 
 private class SharingConfig<T>(
@@ -197,7 +197,7 @@ private fun <T> CoroutineScope.launchSharing(
     shared: MutableSharedFlow<T>,
     started: SharingStarted,
     initialValue: T
-) {
+): Job =
     launch(context) { // the single coroutine to rule the sharing
         // Optimize common built-in started strategies
         when {
@@ -230,7 +230,6 @@ private fun <T> CoroutineScope.launchSharing(
             }
         }
     }
-}
 
 // -------------------------------- stateIn --------------------------------
 
@@ -303,8 +302,8 @@ public fun <T> Flow<T>.stateIn(
 ): StateFlow<T> {
     val config = configureSharing(1)
     val state = MutableStateFlow(initialValue)
-    scope.launchSharing(config.context, config.upstream, state, started, initialValue)
-    return state.asStateFlow()
+    val job = scope.launchSharing(config.context, config.upstream, state, started, initialValue)
+    return ReadonlyStateFlow(state, job)
 }
 
 /**
@@ -332,7 +331,7 @@ private fun <T> CoroutineScope.launchSharingDeferred(
             upstream.collect { value ->
                 state?.let { it.value = value } ?: run {
                     state = MutableStateFlow(value).also {
-                        result.complete(it.asStateFlow())
+                        result.complete(ReadonlyStateFlow(it, coroutineContext.job))
                     }
                 }
             }
@@ -351,23 +350,27 @@ private fun <T> CoroutineScope.launchSharingDeferred(
  * Represents this mutable shared flow as a read-only shared flow.
  */
 public fun <T> MutableSharedFlow<T>.asSharedFlow(): SharedFlow<T> =
-    ReadonlySharedFlow(this)
+    ReadonlySharedFlow(this, null)
 
 /**
  * Represents this mutable state flow as a read-only state flow.
  */
 public fun <T> MutableStateFlow<T>.asStateFlow(): StateFlow<T> =
-    ReadonlyStateFlow(this)
+    ReadonlyStateFlow(this, null)
 
 private class ReadonlySharedFlow<T>(
-    flow: SharedFlow<T>
+    flow: SharedFlow<T>,
+    @Suppress("unused")
+    private val job: Job? // keeps a strong reference to the job (if present)
 ) : SharedFlow<T> by flow, CancellableFlow<T>, FusibleFlow<T> {
     override fun fuse(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow) =
         fuseSharedFlow(context, capacity, onBufferOverflow)
 }
 
 private class ReadonlyStateFlow<T>(
-    flow: StateFlow<T>
+    flow: StateFlow<T>,
+    @Suppress("unused")
+    private val job: Job? // keeps a strong reference to the job (if present)
 ) : StateFlow<T> by flow, CancellableFlow<T>, FusibleFlow<T> {
     override fun fuse(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow) =
         fuseStateFlow(context, capacity, onBufferOverflow)

kotlinx-coroutines-core/common/src/flow/operators/Emitters.kt

@@ -127,5 +127,7 @@ public suspend fun <T> Flow<T>.collectLatest(action: suspend (value: T) -> Unit)
  * Collects all the values from the given [flow] and emits them to the collector.
  * It is a shorthand for `flow.collect { value -> emit(value) }`.
  */
-@BuilderInference
-public suspend inline fun <T> FlowCollector<T>.emitAll(flow: Flow<T>): Unit = flow.collect(this)
+public suspend fun <T> FlowCollector<T>.emitAll(flow: Flow<T>) {
+    ensureActive()
+    flow.collect(this)
+}

kotlinx-coroutines-core/common/src/flow/operators/Share.kt

@@ -1,10 +1,11 @@
 /*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.internal.*
 import kotlin.test.*
 
@@ -290,4 +291,24 @@ class OnCompletionTest : TestBase() {
         val expected = (1..5).toList() + (-1)
         assertEquals(expected, result)
     }
+
+    @Test
+    fun testCancelledEmitAllFlow() = runTest {
+        // emitAll does not call 'collect' on onCompletion collector
+        // if the target flow is empty
+        flowOf(1, 2, 3)
+            .onCompletion { emitAll(MutableSharedFlow()) }
+            .take(1)
+            .collect()
+    }
+
+    @Test
+    fun testCancelledEmitAllChannel() = runTest {
+        // emitAll does not call 'collect' on onCompletion collector
+        // if the target channel is empty
+        flowOf(1, 2, 3)
+            .onCompletion { emitAll(Channel()) }
+            .take(1)
+            .collect()
+    }
 }

kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt

@@ -10,21 +10,21 @@ package kotlinx.coroutines
 import java.util.concurrent.locks.*
 import kotlin.internal.InlineOnly
 
-internal interface TimeSource {
-    fun currentTimeMillis(): Long
-    fun nanoTime(): Long
-    fun wrapTask(block: Runnable): Runnable
-    fun trackTask()
-    fun unTrackTask()
-    fun registerTimeLoopThread()
-    fun unregisterTimeLoopThread()
-    fun parkNanos(blocker: Any, nanos: Long) // should return immediately when nanos <= 0
-    fun unpark(thread: Thread)
+internal abstract class AbstractTimeSource {
+    abstract fun currentTimeMillis(): Long
+    abstract fun nanoTime(): Long
+    abstract fun wrapTask(block: Runnable): Runnable
+    abstract fun trackTask()
+    abstract fun unTrackTask()
+    abstract fun registerTimeLoopThread()
+    abstract fun unregisterTimeLoopThread()
+    abstract fun parkNanos(blocker: Any, nanos: Long) // should return immediately when nanos <= 0
+    abstract fun unpark(thread: Thread)
 }
 
 // For tests only
 // @JvmField: Don't use JvmField here to enable R8 optimizations via "assumenosideeffects"
-internal var timeSource: TimeSource? = null
+internal var timeSource: AbstractTimeSource? = null
 
 @InlineOnly
 internal inline fun currentTimeMillis(): Long =