Version 1.2.0

CHANGES.md

@@ -1,5 +1,21 @@
 # Change log for kotlinx.coroutines
 
+## Version 1.2.0
+
+ * Kotlin updated to 1.3.30.
+ * New API: `CancellableContinuation.resume` with `onCancelling` lambda (#1044) to consistently handle closeable resources.
+ * Play services task version updated to 16.0.1.
+ * `ReceiveChannel.isEmpty` is no longer deprecated
+
+A lot of `Flow` improvements:
+  * Purity property is renamed to context preservation and became more restrictive.
+  * `zip` and `combineLatest` operators.
+  * Integration with RxJava2
+  * `flatMap`, `merge` and `concatenate` are replaced with `flattenConcat`, `flattenMerge`, `flatMapConcat` and `flatMapMerge`.
+  * Various documentation improvements and minor bug fixes.
+
+Note that `Flow` **is not** leaving its [preview status](/docs/compatibility.md#flow-preview-api).
+
 ## Version 1.2.0-alpha-2
 
 This release contains major [feature preview](/docs/compatibility.md#flow-preview-api): cold streams aka `Flow` (#254). 

README.md

@@ -2,10 +2,10 @@
 
 [![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://api.bintray.com/packages/kotlin/kotlinx/kotlinx.coroutines/images/download.svg?version=1.2.0-alpha-2) ](https://bintray.com/kotlin/kotlinx/kotlinx.coroutines/1.2.0-alpha-2)
+[![Download](https://api.bintray.com/packages/kotlin/kotlinx/kotlinx.coroutines/images/download.svg?version=1.2.0) ](https://bintray.com/kotlin/kotlinx/kotlinx.coroutines/1.2.0)
 
 Library support for Kotlin coroutines with [multiplatform](#multiplatform) support.
-This is a companion version for Kotlin `1.3.21` release.
+This is a companion version for Kotlin `1.3.30` release.
 
 ```kotlin
 GlobalScope.launch {
@@ -75,15 +75,15 @@ Add dependencies (you can also add other modules that you need):
 <dependency>
     <groupId>org.jetbrains.kotlinx</groupId>
     <artifactId>kotlinx-coroutines-core</artifactId>
-    <version>1.2.0-alpha-2</version>
+    <version>1.2.0</version>
 </dependency>
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```xml
 <properties>
-    <kotlin.version>1.3.21</kotlin.version>
+    <kotlin.version>1.3.30</kotlin.version>
 </properties>
 ```
 
@@ -93,15 +93,15 @@ Add dependencies (you can also add other modules that you need):
 
 ```groovy
 dependencies {
-    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.0-alpha-2'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.0'
 }
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```groovy
 buildscript {
-    ext.kotlin_version = '1.3.21'
+    ext.kotlin_version = '1.3.30'
 }
 ```
 
@@ -119,15 +119,15 @@ Add dependencies (you can also add other modules that you need):
 
 ```groovy
 dependencies {
-    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.0-alpha-2")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.0")
 }
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```groovy
 plugins {
-    kotlin("jvm") version "1.3.21"
+    kotlin("jvm") version "1.3.30"
 }
 ```
 
@@ -147,7 +147,7 @@ Add [`kotlinx-coroutines-android`](ui/kotlinx-coroutines-android)
 module as dependency when using `kotlinx.coroutines` on Android:
 
 ```groovy
-implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.2.0-alpha-2'
+implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.2.0'
 ```
 This gives you access to Android [Dispatchers.Main](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-android/kotlinx.coroutines.android/kotlinx.coroutines.-dispatchers/index.html)
 coroutine dispatcher and also makes sure that in case of crashed coroutine with unhandled exception this

RELEASE.md

@@ -63,14 +63,18 @@ To release new `<version>` of `kotlinx-coroutines`:
 
 5. Announce new release in [Slack](https://kotlinlang.slack.com)   
 
-6. Switch into `develop` branch:<br>
+6. Create a ticket to update coroutines version on [try.kotlinlang.org](try.kotlinlang.org).
+   * Use [KT-30870](https://youtrack.jetbrains.com/issue/KT-30870) as a template
+   * This step should be skipped for eap versions that are not merged to `master`
+
+7. Switch into `develop` branch:<br>
    `git checkout develop`
 
-7. Fetch the latest `master`:<br>
+8. Fetch the latest `master`:<br>
    `git fetch` 
 
-8. Merge release from `master`:<br>
+9. Merge release from `master`:<br>
    `git merge origin/master`
 
-9. Push updates to `develop`:<br>
+10. Push updates to `develop`:<br>
    `git push`      

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

@@ -40,6 +40,7 @@ public abstract interface class kotlinx/coroutines/CancellableContinuation : kot
 	public abstract fun isActive ()Z
 	public abstract fun isCancelled ()Z
 	public abstract fun isCompleted ()Z
+	public abstract fun resume (Ljava/lang/Object;Lkotlin/jvm/functions/Function1;)V
 	public abstract fun resumeUndispatched (Lkotlinx/coroutines/CoroutineDispatcher;Ljava/lang/Object;)V
 	public abstract fun resumeUndispatchedWithException (Lkotlinx/coroutines/CoroutineDispatcher;Ljava/lang/Throwable;)V
 	public abstract fun tryResume (Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
@@ -58,20 +59,18 @@ public class kotlinx/coroutines/CancellableContinuationImpl : kotlin/coroutines/
 	public fun getCallerFrame ()Lkotlin/coroutines/jvm/internal/CoroutineStackFrame;
 	public fun getContext ()Lkotlin/coroutines/CoroutineContext;
 	public fun getContinuationCancellationCause (Lkotlinx/coroutines/Job;)Ljava/lang/Throwable;
-	public final fun getDelegate ()Lkotlin/coroutines/Continuation;
 	public final fun getResult ()Ljava/lang/Object;
 	public fun getStackTraceElement ()Ljava/lang/StackTraceElement;
-	public fun getSuccessfulResult (Ljava/lang/Object;)Ljava/lang/Object;
 	public synthetic fun initCancellability ()V
 	public fun invokeOnCancellation (Lkotlin/jvm/functions/Function1;)V
 	public fun isActive ()Z
 	public fun isCancelled ()Z
 	public fun isCompleted ()Z
 	protected fun nameString ()Ljava/lang/String;
+	public fun resume (Ljava/lang/Object;Lkotlin/jvm/functions/Function1;)V
 	public fun resumeUndispatched (Lkotlinx/coroutines/CoroutineDispatcher;Ljava/lang/Object;)V
 	public fun resumeUndispatchedWithException (Lkotlinx/coroutines/CoroutineDispatcher;Ljava/lang/Throwable;)V
 	public fun resumeWith (Ljava/lang/Object;)V
-	public fun takeState ()Ljava/lang/Object;
 	public fun toString ()Ljava/lang/String;
 	public fun tryResume (Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
 	public fun tryResumeWithException (Ljava/lang/Throwable;)Ljava/lang/Object;
@@ -794,8 +793,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun broadcastIn (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/CoroutineScope;ILkotlinx/coroutines/CoroutineStart;)Lkotlinx/coroutines/channels/BroadcastChannel;
 	public static synthetic fun broadcastIn$default (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/CoroutineScope;ILkotlinx/coroutines/CoroutineStart;ILjava/lang/Object;)Lkotlinx/coroutines/channels/BroadcastChannel;
 	public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public static final fun concatenate (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun concatenate (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun combineLatest (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun count (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun count (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun delayEach (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
@@ -808,8 +806,12 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun filter (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun filterNot (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun filterNotNull (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun flatMap (Lkotlinx/coroutines/flow/Flow;IILkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
-	public static synthetic fun flatMap$default (Lkotlinx/coroutines/flow/Flow;IILkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun flatMapConcat (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun flatMapMerge (Lkotlinx/coroutines/flow/Flow;IILkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static synthetic fun flatMapMerge$default (Lkotlinx/coroutines/flow/Flow;IILkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun flattenConcat (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun flattenMerge (Lkotlinx/coroutines/flow/Flow;II)Lkotlinx/coroutines/flow/Flow;
+	public static synthetic fun flattenMerge$default (Lkotlinx/coroutines/flow/Flow;IIILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun flow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun flowOf ([Ljava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun flowOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;I)Lkotlinx/coroutines/flow/Flow;
@@ -821,8 +823,6 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun fold (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun map (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun mapNotNull (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun merge (Ljava/lang/Iterable;II)Lkotlinx/coroutines/flow/Flow;
-	public static synthetic fun merge$default (Ljava/lang/Iterable;IIILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onEach (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onErrorCollect (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public static synthetic fun onErrorCollect$default (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
@@ -844,14 +844,17 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static synthetic fun toSet$default (Lkotlinx/coroutines/flow/Flow;Ljava/util/Set;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
 	public static final fun transform (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun unsafeFlow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun zip (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 }
 
 public final class kotlinx/coroutines/flow/MigrationKt {
 	public static final fun BehaviourSubject ()Ljava/lang/Object;
 	public static final fun PublishSubject ()Ljava/lang/Object;
 	public static final fun ReplaySubject ()Ljava/lang/Object;
-	public static final fun concat (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun concatMap (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun flatMap (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun flatten (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun merge (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun observeOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onErrorResume (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun publishOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
@@ -863,7 +866,7 @@ public final class kotlinx/coroutines/flow/MigrationKt {
 }
 
 public final class kotlinx/coroutines/flow/internal/SafeCollector : kotlinx/coroutines/flow/FlowCollector {
-	public fun <init> (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/ContinuationInterceptor;)V
+	public fun <init> (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/CoroutineContext;)V
 	public fun emit (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 

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

@@ -28,6 +28,8 @@ public final class kotlinx/coroutines/rx2/RxConvertKt {
 	public static final fun asMaybe (Lkotlinx/coroutines/Deferred;Lkotlin/coroutines/CoroutineContext;)Lio/reactivex/Maybe;
 	public static final fun asObservable (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/CoroutineContext;)Lio/reactivex/Observable;
 	public static final fun asSingle (Lkotlinx/coroutines/Deferred;Lkotlin/coroutines/CoroutineContext;)Lio/reactivex/Single;
+	public static final fun from (Lkotlinx/coroutines/flow/Flow;)Lio/reactivex/Flowable;
+	public static final fun from (Lkotlinx/coroutines/flow/Flow;)Lio/reactivex/Observable;
 }
 
 public final class kotlinx/coroutines/rx2/RxFlowableKt {

gradle.properties

@@ -1,17 +1,18 @@
 # Kotlin
-version=1.2.0-alpha-2-SNAPSHOT
+version=1.2.0-SNAPSHOT
 group=org.jetbrains.kotlinx
-kotlin_version=1.3.21
+kotlin_version=1.3.30
 
 # Dependencies
 junit_version=4.12
-atomicFU_version=0.12.2
+atomicFU_version=0.12.3
 html_version=0.6.8
 lincheck_version=2.0
 dokka_version=0.9.16-rdev-2-mpp-hacks
 bintray_version=1.8.4-jetbrains-5
 byte_buddy_version=1.9.3
 reactor_vesion=3.2.5.RELEASE
+reactive_streams_version=1.0.2
 artifactory_plugin_version=4.7.3
 
 # JS

integration/kotlinx-coroutines-play-services/build.gradle

@@ -7,7 +7,7 @@ import java.util.zip.ZipFile
  * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-ext.tasks_version = '15.0.1'
+ext.tasks_version = '16.0.1'
 
 def attr = Attribute.of("artifactType", String.class)
 configurations {

integration/kotlinx-coroutines-play-services/src/Tasks.kt

@@ -50,7 +50,8 @@ public fun <T> Task<T>.asDeferred(): Deferred<T> {
     if (isComplete) {
         val e = exception
         return if (e == null) {
-            CompletableDeferred<T>().apply { if (isCanceled) cancel() else complete(result) }
+            @Suppress("UNCHECKED_CAST")
+            CompletableDeferred<T>().apply { if (isCanceled) cancel() else complete(result as T) }
         } else {
             CompletableDeferred<T>().apply { completeExceptionally(e) }
         }
@@ -60,7 +61,8 @@ public fun <T> Task<T>.asDeferred(): Deferred<T> {
     addOnCompleteListener {
         val e = it.exception
         if (e == null) {
-            if (isCanceled) result.cancel() else result.complete(it.result)
+            @Suppress("UNCHECKED_CAST")
+            if (isCanceled) result.cancel() else result.complete(it.result as T)
         } else {
             result.completeExceptionally(e)
         }
@@ -83,7 +85,8 @@ public suspend fun <T> Task<T>.await(): T {
             if (isCanceled) {
                 throw CancellationException("Task $this was cancelled normally.")
             } else {
-                result
+                @Suppress("UNCHECKED_CAST")
+                result as T
             }
         } else {
             throw e
@@ -94,7 +97,8 @@ public suspend fun <T> Task<T>.await(): T {
         addOnCompleteListener {
             val e = exception
             if (e == null) {
-                if (isCanceled) cont.cancel() else cont.resume(result)
+                @Suppress("UNCHECKED_CAST")
+                if (isCanceled) cont.cancel() else cont.resume(result as T)
             } else {
                 cont.resumeWithException(e)
             }

integration/kotlinx-coroutines-play-services/test/TaskTest.kt

@@ -92,6 +92,11 @@ class TaskTest : TestBase() {
         assertEquals(42, deferred.await())
     }
 
+    @Test
+    fun testNullResultTaskAsDeferred() = runTest {
+        assertNull(Tasks.forResult(null).asDeferred().await())
+    }
+
     @Test
     fun testCancelledTaskAsDeferred() = runTest {
         val deferred = Tasks.forCanceled<Int>().asDeferred()

kotlinx-coroutines-core/README.md

@@ -70,7 +70,11 @@ Synchronization primitives (mutex).
 
 # Package kotlinx.coroutines.channels
 
-Channels -- non-blocking primitives for communicating a stream of elements between coroutines.
+Channels — non-blocking primitives for communicating a stream of elements between coroutines.
+
+# Package kotlinx.coroutines.flow
+
+Flow — asynchronous cold stream of elements.
 
 # Package kotlinx.coroutines.selects
 

kotlinx-coroutines-core/common/src/CancellableContinuation.kt

@@ -123,10 +123,16 @@ public interface CancellableContinuation<in T> : Continuation<T> {
      * with cancellation exception. Otherwise, the handler will be invoked once on cancellation if this
      * continuation is cancelled.
      *
-     * Installed [handler] should not throw any exceptions. If it does, they will get caught,
-     * wrapped into [CompletionHandlerException], and rethrown, potentially causing the crash of unrelated code.
+     * Installed [handler] should not throw any exceptions.
+     * If it does, they will get caught, wrapped into [CompletionHandlerException] and
+     * processed as uncaught exception in the context of the current coroutine
+     * (see [CoroutineExceptionHandler]).
      *
      * At most one [handler] can be installed on one continuation.
+     *
+     * **Note**: Implementation of `CompletionHandler` must be fast, non-blocking, and thread-safe.
+     * This handler can be invoked concurrently with the surrounding code.
+     * There is no guarantee on the execution context in which the [handler] is invoked.
      */
     public fun invokeOnCancellation(handler: CompletionHandler)
 
@@ -151,6 +157,34 @@ public interface CancellableContinuation<in T> : Continuation<T> {
      */
     @ExperimentalCoroutinesApi
     public fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable)
+
+    /**
+     * Resumes this continuation with a given [value] and calls the specified [onCancellation]
+     * handler when resumed too late (when continuation was already cancelled) or when resumed
+     * successfully (before cancellation), but coroutine's job was cancelled before it had a
+     * chance to run in its dispatcher, so that suspended function threw an exception
+     * instead of returning this value.
+     *
+     * Installed [onCancellation] handler should not throw any exceptions.
+     * If it does, they will get caught, wrapped into [CompletionHandlerException] and
+     * processed as uncaught exception in the context of the current coroutine
+     * (see [CoroutineExceptionHandler]).
+     *
+     * This function shall be used when resuming with a resource that must be closed by the
+     * code that had called the corresponding suspending function, e.g.:
+     *
+     * ```
+     * continuation.resume(resource) {
+     *     resource.close()
+     * }
+     * ```
+     *
+     * **Note**: Implementation of [onCancellation] handler must be fast, non-blocking, and thread-safe.
+     * This handler can be invoked concurrently with the surrounding code.
+     * There is no guarantee on the execution context in which the [onCancellation] handler is invoked.
+     */
+    @ExperimentalCoroutinesApi // since 1.2.0, tentatively graduates in 1.3.0
+    public fun resume(value: T, onCancellation: (cause: Throwable) -> Unit)
 }
 
 /**