runBlocking is improved to properly support specified dispatchers,

so that `runBlocking(UI) { ... }` can be used outside of UI thread.

Fixes #209

Author: elizarov

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

@@ -155,6 +155,11 @@ public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T
  * in this blocked thread until the completion of this coroutine.
  * See [CoroutineDispatcher] for the other implementations that are provided by `kotlinx.coroutines`.
  *
+ * When [CoroutineDispatcher] is explicitly specified in the [context], then the new coroutine runs in the context of
+ * the specified dispatcher while the current thread is blocked. If the specified dispatcher implements [EventLoop]
+ * interface and this `runBlocking` invocation is performed from inside of the this event loop's thread, then
+ * this event loop is processed using its [processNextEvent][EventLoop.processNextEvent] method until coroutine completes.
+ *
  * If this blocked thread is interrupted (see [Thread.interrupt]), then the coroutine job is cancelled and
  * this `runBlocking` invocation throws [InterruptedException].
  *
@@ -166,9 +171,13 @@ public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T
 @Throws(InterruptedException::class)
 public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
     val currentThread = Thread.currentThread()
-    val eventLoop = if (context[ContinuationInterceptor] == null) BlockingEventLoop(currentThread) else null
-    val newContext = newCoroutineContext(context + (eventLoop ?: EmptyCoroutineContext))
-    val coroutine = BlockingCoroutine<T>(newContext, currentThread, privateEventLoop = eventLoop != null)
+    val contextInterceptor = context[ContinuationInterceptor]
+    val privateEventLoop = contextInterceptor == null // create private event loop if no dispatcher is specified
+    val eventLoop = if (privateEventLoop) BlockingEventLoop(currentThread) else contextInterceptor as? EventLoop
+    val newContext = newCoroutineContext(
+        if (privateEventLoop) context + (eventLoop as ContinuationInterceptor) else context
+    )
+    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop, privateEventLoop)
     coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
     return coroutine.joinBlocking()
 }
@@ -210,10 +219,9 @@ private class RunCompletion<in T>(
 private class BlockingCoroutine<T>(
     parentContext: CoroutineContext,
     private val blockedThread: Thread,
+    private val eventLoop: EventLoop?,
     private val privateEventLoop: Boolean
 ) : AbstractCoroutine<T>(parentContext, true) {
-    private val eventLoop: EventLoop? = parentContext[ContinuationInterceptor] as? EventLoop
-
     init {
         if (privateEventLoop) require(eventLoop is BlockingEventLoop)
     }

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

@@ -0,0 +1,68 @@
+/*
+ * 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 kotlin.coroutines.experimental.*
+import kotlin.test.*
+
+class RunBlockingTest : TestBase() {
+    @Test
+    fun testPrivateEventLoop() {
+        expect(1)
+        runBlocking {
+            expect(2)
+            assertTrue(coroutineContext[ContinuationInterceptor] is EventLoop)
+            yield() // is supported!
+            expect(3)
+        }
+        finish(4)
+    }
+
+    @Test
+    fun testOuterEventLoop() {
+        expect(1)
+        runBlocking {
+            expect(2)
+            val outerEventLoop = coroutineContext[ContinuationInterceptor] as EventLoop
+            runBlocking(coroutineContext) {
+                expect(3)
+                // still same event loop
+                assertTrue(coroutineContext[ContinuationInterceptor] === outerEventLoop)
+                yield() // still works
+                expect(4)
+            }
+            expect(5)
+        }
+        finish(6)
+    }
+
+    @Test
+    fun testOtherDispatcher() {
+        expect(1)
+        val name = "RunBlockingTest.testOtherDispatcher"
+        val thread = newSingleThreadContext(name)
+        runBlocking(thread) {
+            expect(2)
+            assertTrue(coroutineContext[ContinuationInterceptor] === thread)
+            assertTrue(Thread.currentThread().name.contains(name))
+            yield() // should work
+            expect(3)
+        }
+        finish(4)
+        thread.close()
+    }
+}