Fixed hanged receive from a closed subscription of BroadcastChannel

Fixes #226

Author: elizarov

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

@@ -339,8 +339,9 @@ class ArrayBroadcastChannel<E>(
             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 closedBroadcast ?: POLL_FAILED // must retest `needsToCheckOfferWithoutLock` outside of the lock
+                // no elements to poll from the queue -- check if closed broads & closed this sub
+                // must retest `needsToCheckOfferWithoutLock` outside of the lock
+                return closedBroadcast ?: this.closedForReceive ?: POLL_FAILED
             }
             // 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.

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

@@ -19,7 +19,7 @@ package kotlinx.coroutines.experimental.channels
 import kotlinx.coroutines.experimental.*
 import org.hamcrest.core.IsEqual
 import org.hamcrest.core.IsNull
-import org.junit.Assert.assertThat
+import org.junit.Assert.*
 import org.junit.Test
 
 class ArrayBroadcastChannelTest : TestBase() {
@@ -166,4 +166,16 @@ class ArrayBroadcastChannelTest : TestBase() {
         }
         check(expected == 2)
     }
+
+    @Test
+    fun testReceiveFromClosedSub() = runTest(
+        expected = { it is ClosedReceiveChannelException }
+    ) {
+        val channel = BroadcastChannel<Int>(1)
+        val sub = channel.openSubscription()
+        assertFalse(sub.isClosedForReceive)
+        sub.close()
+        assertTrue(sub.isClosedForReceive)
+        sub.receive()
+    }
 }