Merge 'develop' into project-structure

# Conflicts:
#	reactive/kotlinx-coroutines-rx2/src/test/kotlin/guide/test/GuideReactiveTest.kt

Author: elizarov

CHANGES.md

@@ -1,5 +1,9 @@
 # Change log for kotlinx.coroutines
 
+## Version 0.23.4
+
+* Recompiled with Kotlin 1.2.51 to solve broken metadata problem (see [KT-24944](https://youtrack.jetbrains.com/issue/KT-24944)).
+
 ## Version 0.23.3
 
 * Kotlin 1.2.50.

README.md

@@ -2,12 +2,12 @@
 
 [![official JetBrains project](http://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)](http://www.apache.org/licenses/LICENSE-2.0)
-[![Download](https://api.bintray.com/packages/kotlin/kotlinx/kotlinx.coroutines/images/download.svg?version=0.23.3) ](https://bintray.com/kotlin/kotlinx/kotlinx.coroutines/0.23.3)
+[![Download](https://api.bintray.com/packages/kotlin/kotlinx/kotlinx.coroutines/images/download.svg?version=0.23.4) ](https://bintray.com/kotlin/kotlinx/kotlinx.coroutines/0.23.4)
 
 Library support for Kotlin coroutines in
 [Kotlin/JVM](core/README.md) and
 [Kotlin/JS](js/README.md).
-This is a companion version for Kotlin 1.2.50 release.
+This is a companion version for Kotlin 1.2.51 release.
 
 ```kotlin
 launch {
@@ -64,15 +64,15 @@ Add dependencies (you can also add other modules that you need):
 <dependency>
     <groupId>org.jetbrains.kotlinx</groupId>
     <artifactId>kotlinx-coroutines-core</artifactId>
-    <version>0.23.3</version>
+    <version>0.23.4</version>
 </dependency>
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```xml
 <properties>
-    <kotlin.version>1.2.50</kotlin.version>
+    <kotlin.version>1.2.51</kotlin.version>
 </properties>
 ```
 
@@ -81,14 +81,14 @@ And make sure that you use the latest Kotlin version:
 Add dependencies (you can also add other modules that you need):
 
 ```groovy
-compile 'org.jetbrains.kotlinx:kotlinx-coroutines-core:0.23.3'
+compile 'org.jetbrains.kotlinx:kotlinx-coroutines-core:0.23.4'
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```groovy
 buildscript {
-    ext.kotlin_version = '1.2.50'
+    ext.kotlin_version = '1.2.51'
 }
 ```
 
@@ -105,6 +105,20 @@ repository {
 Use `org.jetbrains.kotlinx:kotlinx-coroutines-core-js:<version>` artifact in your Gradle/Maven dependencies 
 or install [`kotlinx-coroutines-core`](https://www.npmjs.com/package/kotlinx-coroutines-core) package via NPM. 
 
+### Android
+
+Add [`kotlinx-coroutines-android`](ui/kotlinx-coroutines-android)
+module as dependency when using `kotlinx.coroutines` on Android:
+
+```groovy
+compile 'org.jetbrains.kotlinx:kotlinx-coroutines-android:0.23.4'
+```
+
+This gives you access to Android [UI](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-android/kotlinx.coroutines.experimental.android/-u-i.html)
+coroutine dispatcher and also makes sure that in case of crashed coroutine with unhandled exception this
+exception is logged before crashing Android application, similarly to the way uncaught exceptions in 
+threads are handled by Android runtime. 
+
 ### ProGuard
 
 In obfuscated code, fields with different types can have the same names,

build.gradle

@@ -41,6 +41,12 @@ allprojects {
     if (useKotlinSnapshot) {
         kotlin_version = '1.2-SNAPSHOT'
     }
+
+    repositories {
+        jcenter()
+        maven { url "https://kotlin.bintray.com/kotlin-dev" }
+        maven { url "https://kotlin.bintray.com/kotlinx" }
+    }
 }
 
 

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

@@ -44,8 +44,14 @@ class ArrayBroadcastChannel<E>(
         require(capacity >= 1) { "ArrayBroadcastChannel capacity must be at least 1, but $capacity was specified" }
     }
 
+    /*
+     *  Writes to buffer are guarded by bufferLock, but reads from buffer are concurrent with writes
+     *    - Write element to buffer then write "tail" (volatile)
+     *    - Read "tail" (volatile), then read element from buffer
+     *  So read/writes to buffer need not be volatile
+     */
     private val bufferLock = ReentrantLock()
-    private val buffer = arrayOfNulls<Any?>(capacity) // guarded by bufferLock
+    private val buffer = arrayOfNulls<Any?>(capacity)
 
     // head & tail are Long (64 bits) and we assume that they never wrap around
     // head, tail, and size are guarded by bufferLock
@@ -56,14 +62,7 @@ class ArrayBroadcastChannel<E>(
     @Volatile
     private var size: Int = 0
 
-    /*
-        Writes to buffer are guarded by bufferLock, but reads from buffer are concurrent with writes
-          - Write element to buffer then write "tail" (volatile)
-          - Read "tail" (volatile), then read element from buffer
-        So read/writes to buffer need not be volatile
-     */
-
-    private val subs = subscriberList<Subscriber<E>>()
+    private val subscribers = subscriberList<Subscriber<E>>()
 
     override val isBufferAlwaysFull: Boolean get() = false
     override val isBufferFull: Boolean get() = size >= capacity
@@ -81,7 +80,7 @@ class ArrayBroadcastChannel<E>(
 
     public override fun cancel(cause: Throwable?): Boolean =
         close(cause).also {
-            for (sub in subs) sub.cancel(cause)
+            for (sub in subscribers) sub.cancel(cause)
         }
 
     // result is `OFFER_SUCCESS | OFFER_FAILED | Closed`
@@ -96,7 +95,7 @@ class ArrayBroadcastChannel<E>(
             this.size = size + 1
             this.tail = tail + 1
         }
-        // if offered successfully, then check subs outside of lock
+        // if offered successfully, then check subscribers outside of lock
         checkSubOffers()
         return OFFER_SUCCESS
     }
@@ -117,7 +116,7 @@ class ArrayBroadcastChannel<E>(
             this.size = size + 1
             this.tail = tail + 1
         }
-        // if offered successfully, then check subs outside of lock
+        // if offered successfully, then check subscribers outside of lock
         checkSubOffers()
         return OFFER_SUCCESS
     }
@@ -126,7 +125,7 @@ class ArrayBroadcastChannel<E>(
         var updated = false
         var hasSubs = false
         @Suppress("LoopToCallChain") // must invoke `checkOffer` on every sub
-        for (sub in subs) {
+        for (sub in subscribers) {
             hasSubs = true
             if (sub.checkOffer()) updated = true
         }
@@ -142,12 +141,12 @@ class ArrayBroadcastChannel<E>(
         bufferLock.withLock {
             if (addSub != null) {
                 addSub.subHead = tail // start from last element
-                val wasEmpty = subs.isEmpty()
-                subs.add(addSub)
+                val wasEmpty = subscribers.isEmpty()
+                subscribers.add(addSub)
                 if (!wasEmpty) return // no need to update when adding second and etc sub
             }
             if (removeSub != null) {
-                subs.remove(removeSub)
+                subscribers.remove(removeSub)
                 if (head != removeSub.subHead) return // no need to update
             }
             val minHead = computeMinHead()
@@ -190,7 +189,7 @@ class ArrayBroadcastChannel<E>(
 
     private fun computeMinHead(): Long {
         var minHead = Long.MAX_VALUE
-        for (sub in subs)
+        for (sub in subscribers)
             minHead = minHead.coerceAtMost(sub.subHead) // volatile (atomic) reads of subHead
         return minHead
     }

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

@@ -21,6 +21,30 @@ import kotlin.coroutines.experimental.*
 import kotlin.test.*
 
 class ArrayBroadcastChannelTest : TestBase() {
+
+    @Test
+    fun testConcurrentModification() = runTest {
+        val channel = ArrayBroadcastChannel<Int>(1)
+        val s1 = channel.openSubscription()
+        val s2 = channel.openSubscription()
+
+        val job1 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(1)
+            s1.receive()
+            s1.cancel()
+        }
+
+        val job2 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            s2.receive()
+        }
+
+        expect(3)
+        channel.send(1)
+        joinAll(job1, job2)
+        finish(4)
+    }
+
     @Test
     fun testBasic() = runTest {
         expect(1)

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

@@ -22,6 +22,29 @@ import kotlin.test.*
 
 class ConflatedBroadcastChannelTest : TestBase() {
 
+    @Test
+    fun testConcurrentModification() = runTest {
+        val channel = ConflatedBroadcastChannel<Int>()
+        val s1 = channel.openSubscription()
+        val s2 = channel.openSubscription()
+
+        val job1 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(1)
+            s1.receive()
+            s1.cancel()
+        }
+
+        val job2 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            s2.receive()
+        }
+
+        expect(3)
+        channel.send(1)
+        joinAll(job1, job2)
+        finish(4)
+    }
+
     @Test
     fun testBasicScenario() = runTest {
         expect(1)

core/kotlinx-coroutines-core/test/guide/example-channel-06.kt

@@ -29,8 +29,8 @@ fun produceNumbers() = produce<Int> {
 }
 
 fun launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
-    channel.consumeEach {
-        println("Processor #$id received $it")
+    for (msg in channel) {
+        println("Processor #$id received $msg")
     }    
 }
 

coroutines-guide.md

@@ -1531,8 +1531,8 @@ received number:
 
 ```kotlin
 fun launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
-    channel.consumeEach {
-        println("Processor #$id received $it")
+    for (msg in channel) {
+        println("Processor #$id received $msg")
     }    
 }
 ```
@@ -1571,6 +1571,11 @@ Processor #3 received 10
 Note, that cancelling a producer coroutine closes its channel, thus eventually terminating iteration
 over the channel that processor coroutines are doing.
 
+Also, pay attention to how we explicitly iterate over channel with `for` loop to perform fan-out in `launchProcessor` code. 
+Unlike `consumeEach`, this `for` loop pattern is perfectly safe to use from multiple coroutines. If one of the processor 
+coroutines fails, then others would still be processing the channel, while a processor that is written via `consumeEach` 
+always consumes (cancels) the underlying channel on its normal or abnormal termination.     
+
 ### Fan-in
 
 Multiple coroutines may send to the same channel.

gradle.properties

@@ -1,7 +1,7 @@
-version = 0.23.3-SNAPSHOT
+version = 0.23.4-SNAPSHOT
 group = org.jetbrains.kotlinx
 
-kotlin_version = 1.2.50
+kotlin_version = 1.2.51
 kotlin_native_version = 0.7-dev-1436
 junit_version = 4.12
 atomicFU_version = 0.10.3

gradle/compile-common.gradle

@@ -10,8 +10,3 @@ dependencies {
     testCompile "org.jetbrains.kotlin:kotlin-test-common:$kotlin_version"
     testCompile "org.jetbrains.kotlin:kotlin-test-annotations-common:$kotlin_version"
 }
-
-repositories {
-    jcenter()
-    maven { url "https://kotlin.bintray.com/kotlinx" }
-}