Fixed NPE when ArrayBroadcastChannel is closed concurrently with receive

Fixes #97

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ArrayBroadcastChannel.kt

@@ -338,13 +338,20 @@ class ArrayBroadcastChannel<E>(
             val subHead = this.subHead // guarded read (can be non-volatile read)
             // note: from the broadcastChannel we must read closed token first, then read its tail
             // because it is Ok if tail moves in between the reads (we make decision based on tail first)
-            val closed = broadcastChannel.closedForReceive // unguarded volatile read
+            val closedBroadcast = broadcastChannel.closedForReceive // unguarded volatile read
             val tail = broadcastChannel.tail // unguarded volatile read
             if (subHead >= tail) {
                 // no elements to poll from the queue -- check if closed
-                return closed ?: POLL_FAILED // must retest `needsToCheckOfferWithoutLock` outside of the lock
+                return closedBroadcast ?: POLL_FAILED // must retest `needsToCheckOfferWithoutLock` outside of the lock
             }
-            return broadcastChannel.elementAt(subHead)
+            // Get tentative result. This result may be wrong (completely invalid value, including null),
+            // because this subscription might get closed, moving channel's head past this subscription's head.
+            val result = broadcastChannel.elementAt(subHead)
+            // now check if this subscription was closed
+            val closedSub = this.closedForReceive
+            if (closedSub != null) return closedSub
+            // we know the subscription was not closed, so this tentative result is Ok to return
+            return result
         }
     }
 }

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ArrayBroadcastChannelTest.kt

@@ -132,4 +132,38 @@ class ArrayBroadcastChannelTest : TestBase() {
         yield()
         finish(6)
     }
+
+    @Test
+    fun testReceiveFullAfterClose() = runBlocking<Unit> {
+        val channel = BroadcastChannel<Int>(10)
+        val sub = channel.openSubscription()
+        // generate into buffer & close
+        for (x in 1..5) channel.send(x)
+        channel.close()
+        // make sure all of them are consumed
+        check(!sub.isClosedForReceive)
+        for (x in 1..5) check(sub.receive() == x)
+        check(sub.receiveOrNull() == null)
+        check(sub.isClosedForReceive)
+    }
+
+    @Test
+    fun testCloseSubDuringIteration() = runBlocking<Unit> {
+        val channel = BroadcastChannel<Int>(1)
+        // launch generator (for later) in this context
+        launch(coroutineContext) {
+            for (x in 1..5) channel.send(x)
+            channel.close()
+        }
+        // start consuming
+        val sub = channel.openSubscription()
+        var expected = 0
+        sub.consumeEach {
+            check(it == ++expected)
+            if (it == 2) {
+                sub.close()
+            }
+        }
+        check(expected == 2)
+    }
 }