BroadcastChannel factory function

Author: elizarov

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

@@ -41,7 +41,7 @@ class ArrayBroadcastChannel<E>(
     val capacity: Int
 ) : AbstractSendChannel<E>(), BroadcastChannel<E> {
     init {
-        check(capacity >= 1) { "ArrayBroadcastChannel capacity must be at least 1, but $capacity was specified" }
+        require(capacity >= 1) { "ArrayBroadcastChannel capacity must be at least 1, but $capacity was specified" }
     }
 
     private val bufferLock = ReentrantLock()

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

@@ -35,7 +35,7 @@ public open class ArrayChannel<E>(
     val capacity: Int
 ) : AbstractChannel<E>() {
     init {
-        check(capacity >= 1) { "ArrayChannel capacity must be at least 1, but $capacity was specified" }
+        require(capacity >= 1) { "ArrayChannel capacity must be at least 1, but $capacity was specified" }
     }
 
     private val lock = ReentrantLock()

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

@@ -16,14 +16,40 @@
 
 package kotlinx.coroutines.experimental.channels
 
+import kotlinx.coroutines.experimental.channels.Channel.Factory.CONFLATED
+import kotlinx.coroutines.experimental.channels.Channel.Factory.UNLIMITED
 import java.io.Closeable
 
 /**
  * Broadcast channel is a non-blocking primitive for communication between the sender and multiple receivers
  * that subscribe for the elements using [open] function and unsubscribe using [SubscriptionReceiveChannel.close]
  * function.
+ *
+ * See [BroadcastChannel()][BroadcastChannel.invoke] factory function for the description of available
+ * broadcast channel implementations.
  */
 public interface BroadcastChannel<E> : SendChannel<E> {
+    /**
+     * Factory for broadcast channels.
+     */
+    public companion object Factory {
+        /**
+         * Creates a broadcast channel with the specified buffer capacity.
+         *
+         * The resulting channel type depends on the specified [capacity] parameter:
+         * * when `capacity` positive, but less than [UNLIMITED] -- creates [ArrayBroadcastChannel];
+         * * when `capacity` is [CONFLATED] -- creates [ConflatedBroadcastChannel] that conflates back-to-back sends;
+         * * otherwise -- throws [IllegalArgumentException].
+         */
+        public operator fun <E> invoke(capacity: Int): BroadcastChannel<E> =
+            when (capacity) {
+                0 -> throw IllegalArgumentException("Unsupported 0 capacity for BroadcastChannel")
+                UNLIMITED -> throw IllegalArgumentException("Unsupported UNLIMITED capacity for BroadcastChannel")
+                CONFLATED -> ConflatedBroadcastChannel()
+                else -> ArrayBroadcastChannel(capacity)
+            }
+    }
+
     /**
      * Subscribes to this [BroadcastChannel] and returns a channel to receive elements from it.
      * The resulting channel shall be [closed][SubscriptionReceiveChannel.close] to unsubscribe from this

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

@@ -243,22 +243,22 @@ public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {
         public const val CONFLATED = -1
 
         /**
-         * Creates a channel with specified buffer capacity (or without a buffer by default).
+         * Creates a channel with the specified buffer capacity (or without a buffer by default).
          *
          * The resulting channel type depends on the specified [capacity] parameter:
-         * * when `capacity` is 0 -- creates [RendezvousChannel] without a buffer;
+         * * when `capacity` is 0 (default) -- creates [RendezvousChannel] without a buffer;
          * * when `capacity` is [UNLIMITED] -- creates [LinkedListChannel] with buffer of unlimited size;
          * * when `capacity` is [CONFLATED] -- creates [ConflatedChannel] that conflates back-to-back sends;
-         * * otherwise -- creates [ArrayChannel] with a buffer of the specified `capacity`.
+         * * when `capacity` is positive, but less than [UNLIMITED] -- creates [ArrayChannel] with a buffer of the specified `capacity`;
+         * * otherwise -- throws [IllegalArgumentException].
          */
-        public operator fun <E> invoke(capacity: Int = 0): Channel<E> {
-            return when (capacity) {
+        public operator fun <E> invoke(capacity: Int = 0): Channel<E> =
+            when (capacity) {
                 0 -> RendezvousChannel()
                 UNLIMITED -> LinkedListChannel()
                 CONFLATED -> ConflatedChannel()
                 else -> ArrayChannel(capacity)
             }
-        }
     }
 }
 

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+import org.hamcrest.MatcherAssert.assertThat
+import org.hamcrest.core.IsInstanceOf
+import org.junit.Test
+
+class BroadcastChannelFactoryTest {
+    @Test(expected = IllegalArgumentException::class)
+    fun testRendezvousChannelNotSupported() {
+        BroadcastChannel<Int>(0)
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun testLinkedListChannelNotSupported() {
+        BroadcastChannel<Int>(Channel.UNLIMITED)
+    }
+
+    @Test
+    fun testConflatedBroadcastChannel() {
+        assertThat(BroadcastChannel<Int>(Channel.CONFLATED), IsInstanceOf(ConflatedBroadcastChannel::class.java))
+    }
+
+    @Test
+    fun testArrayBroadcastChannel() {
+        assertThat(BroadcastChannel<Int>(1), IsInstanceOf(ArrayBroadcastChannel::class.java))
+        assertThat(BroadcastChannel<Int>(10), IsInstanceOf(ArrayBroadcastChannel::class.java))
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun testInvalidCapacityNotSupported() {
+        BroadcastChannel<Int>(-2)
+    }
+}

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+import org.hamcrest.MatcherAssert.assertThat
+import org.hamcrest.core.IsInstanceOf
+import org.junit.Test
+
+class ChannelFactoryTest {
+    @Test
+    fun testRendezvousChannel() {
+        assertThat(Channel<Int>(), IsInstanceOf(RendezvousChannel::class.java))
+        assertThat(Channel<Int>(0), IsInstanceOf(RendezvousChannel::class.java))
+    }
+
+    @Test
+    fun testLinkedListChannel() {
+        assertThat(Channel<Int>(Channel.UNLIMITED), IsInstanceOf(LinkedListChannel::class.java))
+    }
+
+    @Test
+    fun testConflatedChannel() {
+        assertThat(Channel<Int>(Channel.CONFLATED), IsInstanceOf(ConflatedChannel::class.java))
+    }
+
+    @Test
+    fun testArrayChannel() {
+        assertThat(Channel<Int>(1), IsInstanceOf(ArrayChannel::class.java))
+        assertThat(Channel<Int>(10), IsInstanceOf(ArrayChannel::class.java))
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun testInvalidCapacityNotSupported() {
+        Channel<Int>(-2)
+    }
+}