CoroutineDispatcher.asExecutor() to respect isDispatchNeeded (#3683)

Calling `Dispatchers.Main.immediate.asExecutor().execute { ... }`
should respect the fact that it's immediate and avoid dispatch (and
posting to the Looper) when in the right context.

kotlinx-coroutines-core/jvm/src/Executors.kt

@@ -108,7 +108,14 @@ public fun CoroutineDispatcher.asExecutor(): Executor =
     (this as? ExecutorCoroutineDispatcher)?.executor ?: DispatcherExecutor(this)
 
 private class DispatcherExecutor(@JvmField val dispatcher: CoroutineDispatcher) : Executor {
-    override fun execute(block: Runnable) = dispatcher.dispatch(EmptyCoroutineContext, block)
+    override fun execute(block: Runnable) {
+        if (dispatcher.isDispatchNeeded(EmptyCoroutineContext)) {
+            dispatcher.dispatch(EmptyCoroutineContext, block)
+        } else {
+            block.run()
+        }
+    }
+
     override fun toString(): String = dispatcher.toString()
 }
 

kotlinx-coroutines-core/jvm/test/ExecutorsTest.kt

@@ -81,6 +81,22 @@ class ExecutorsTest : TestBase() {
         finish(4)
     }
 
+    @Test
+    fun testCustomDispatcherToExecutorDispatchNotNeeded() {
+        expect(1)
+        val dispatcher = object : CoroutineDispatcher() {
+            override fun isDispatchNeeded(context: CoroutineContext) = false
+
+            override fun dispatch(context: CoroutineContext, block: Runnable) {
+                fail("should not dispatch")
+            }
+        }
+        dispatcher.asExecutor().execute {
+            expect(2)
+        }
+        finish(3)
+    }
+
     @Test
     fun testTwoThreads() {
         val ctx1 = newSingleThreadContext("Ctx1")
@@ -106,4 +122,4 @@ class ExecutorsTest : TestBase() {
         dispatcher.close()
         check(executorService.isShutdown)
     }
-}
+}