Support for blocking calls inside coroutines (#79)

Author: Francesco Vasco

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

@@ -16,6 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
+import java.util.concurrent.Executor
 import java.util.concurrent.locks.LockSupport
 import kotlin.coroutines.experimental.*
 import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
@@ -24,6 +25,24 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
 
 // --------------- basic coroutine builders ---------------
 
+/**
+ * Suspend current coroutine and execute the [block] blocking code in [executor].
+ * The current coroutine is resumed after [block] execution.
+ *
+ * @param executor the executor for blocking code
+ * @param block the blocking code
+ */
+suspend fun <T> blocking(executor: Executor = DefaultBlockingExecutor, block: () -> T) =
+        suspendCoroutine<T> { cont ->
+            executor.execute {
+                try {
+                    cont.resume(block())
+                } catch (t: Throwable) {
+                    cont.resumeWithException(t)
+                }
+            }
+        }
+
 /**
  * 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].

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

@@ -0,0 +1,43 @@
+/*
+ * 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.*
+
+private const val DEFAULT_CORE_POOL_SIZE = 1
+private const val DEFAULT_KEEP_ALIVE = 60_000L // in milliseconds
+private const val DEFAULT_MAXIMUM_POOL_SIZE = 256
+
+private val KEEP_ALIVE = try {
+    java.lang.Long.getLong("kotlinx.coroutines.DefaultBlockingExecutor.keepAlive", DEFAULT_KEEP_ALIVE)
+} catch (e: SecurityException) {
+    DEFAULT_KEEP_ALIVE
+}
+
+private val MAXIMUM_POOL_SIZE = try {
+    java.lang.Integer.getInteger("kotlinx.coroutines.DefaultBlockingExecutor.maximumPoolSize", DEFAULT_MAXIMUM_POOL_SIZE)
+} catch (e: SecurityException) {
+    DEFAULT_MAXIMUM_POOL_SIZE
+}
+
+internal object DefaultBlockingExecutor : Executor by ThreadPoolExecutor(
+        DEFAULT_CORE_POOL_SIZE,
+        MAXIMUM_POOL_SIZE,
+        KEEP_ALIVE, TimeUnit.MILLISECONDS,
+        LinkedBlockingQueue<Runnable>(),
+        ThreadFactory { Thread(it, "kotlinx.coroutines.DefaultBlockingExecutor").apply { isDaemon = true } }
+)

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

@@ -19,6 +19,7 @@ package kotlinx.coroutines.experimental
 import kotlinx.coroutines.experimental.selects.SelectBuilder
 import kotlinx.coroutines.experimental.selects.SelectInstance
 import kotlinx.coroutines.experimental.selects.select
+import java.util.concurrent.Executor
 import kotlin.coroutines.experimental.CoroutineContext
 
 /**
@@ -160,6 +161,21 @@ public fun <T> async(
 public fun <T> async(context: CoroutineContext, start: Boolean, block: suspend CoroutineScope.() -> T): Deferred<T> =
     async(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block)
 
+/**
+ * Execute the blocking code [block] in the [executor] without blocking current coroutine.
+ * [start] policy is same as [async] method, but [CoroutineStart.UNDISPATCHED] is an invalid option
+ *
+ * @param executor the executor for blocking code
+ * @param start start option, [CoroutineStart.UNDISPATCHED] is invalid
+ * @param block the blocking code
+ */
+public fun <T> blockingAsync(executor: Executor = DefaultBlockingExecutor,
+                             start: CoroutineStart = CoroutineStart.DEFAULT,
+                             block: () -> T): Deferred<T> {
+    require(start != CoroutineStart.UNDISPATCHED) { "Start blocking code undispatched is not supported" }
+    return async(executor.asCoroutineDispatcher(), start) { block() }
+}
+
 /**
  * @suppress **Deprecated**: `defer` was renamed to `async`.
  */

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/AsyncTest.kt

@@ -144,6 +144,24 @@ class AsyncTest : TestBase() {
         override fun toString(): String = error("toString")
     }
 
+
+    @Test
+    fun blockingAsync() = runBlocking {
+        val d = blockingAsync { 42 }
+        check(d.await() == 42)
+    }
+
+    @Test
+    fun blockingAsyncLazy() = runBlocking {
+        val d = blockingAsync(start = CoroutineStart.LAZY) { 42 }
+        check(d.await() == 42)
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun blockingAsyncUndispatched() = runBlocking {
+        blockingAsync(start = CoroutineStart.UNDISPATCHED) { 42 }
+    }
+
     @Test
     fun testDeferBadClass() = runBlocking {
         val bad = BadClass()
@@ -154,4 +172,4 @@ class AsyncTest : TestBase() {
         assertTrue(d.await() === bad)
         finish(2)
     }
-}
+}