BlockingPool implementation.

Author: fvasco

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/BlockingPool.kt

@@ -0,0 +1,47 @@
+/*
+ * 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
+
+import java.util.concurrent.Executor
+import java.util.concurrent.Executors
+import kotlin.coroutines.experimental.CoroutineContext
+
+/**
+ * This is a common pool of shared threads for blocking tasks.
+ *
+ * It uses a *unbounded* cached thread pool.
+ */
+object BlockingPool : CoroutineDispatcher(), Executor {
+
+    private val threadGroup = ThreadGroup("kotlinx.coroutines.BlockingPool")
+
+    private val executor = Executors.newCachedThreadPool({
+        Thread(threadGroup, it, "kotlinx.coroutines.BlockingPool-${System.identityHashCode(it).toString(16)}").apply {
+            isDaemon = true
+        }
+    })
+
+    override fun execute(command: Runnable) = executor.execute(command)
+
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
+        execute(block)
+    }
+
+    fun isDispatchNeeded(): Boolean = threadGroup !== Thread.currentThread().threadGroup
+
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = isDispatchNeeded()
+}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt

@@ -24,6 +24,27 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
 
 // --------------- basic coroutine builders ---------------
 
+/**
+ * Execute the given [block] in an unbounded blocking executor.
+ * If the current thread is a blocking one then the [block] is executed immediately,
+ * otherwise the coroutine is suspended and the [block] is scheduled for execution on a different thread.
+ */
+public suspend fun <T> blocking(block: () -> T): T = when {
+    BlockingPool.isDispatchNeeded() ->
+        suspendCoroutine { continuation ->
+            BlockingPool.execute {
+                val res = try {
+                    block()
+                } catch (exception: Throwable) {
+                    continuation.resumeWithException(exception)
+                    return@execute
+                }
+                continuation.resume(res)
+            }
+        }
+    else -> block()
+}
+
 /**
  * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
  * The coroutine is cancelled when the resulting job is [cancelled][Job.cancel].