Scheduler.asCoroutineDispatcher extension for Rx 2.x

Author: elizarov

reactive/kotlinx-coroutines-rx2/README.md

@@ -33,11 +33,13 @@ Conversion functions:
 | [Job.asCompletable][kotlinx.coroutines.experimental.Job.asCompletable] | Converts job to hot completable
 | [Deferred.asSingle][kotlinx.coroutines.experimental.Deferred.asSingle] | Converts deferred value to hot single
 | [ReceiveChannel.asObservable][kotlinx.coroutines.experimental.channels.ReceiveChannel.asObservable] | Converts streaming channel to hot observable
+| [Scheduler.asCoroutineDispatcher][io.reactivex.Scheduler.asCoroutineDispatcher] | Converts scheduler to [CoroutineDispatcher]
 
 <!--- SITE_ROOT https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core -->
 <!--- DOCS_ROOT kotlinx-coroutines-core/target/dokka/kotlinx-coroutines-core -->
 <!--- INDEX kotlinx.coroutines.experimental -->
 [CoroutineScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-scope/index.html
+[CoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-dispatcher/index.html
 <!--- INDEX kotlinx.coroutines.experimental.channels -->
 [ProducerScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-producer-scope/index.html
 [ReceiveChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-receive-channel/index.html
@@ -57,6 +59,7 @@ Conversion functions:
 [kotlinx.coroutines.experimental.Job.asCompletable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/kotlinx.coroutines.experimental.-job/as-completable.html
 [kotlinx.coroutines.experimental.Deferred.asSingle]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/kotlinx.coroutines.experimental.-deferred/as-single.html
 [kotlinx.coroutines.experimental.channels.ReceiveChannel.asObservable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/kotlinx.coroutines.experimental.channels.-receive-channel/as-observable.html
+[io.reactivex.Scheduler.asCoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/io.reactivex.-scheduler/as-coroutine-dispatcher.html
 <!--- END -->
 
 # Package kotlinx.coroutines.experimental.rx2

reactive/kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxAwait.kt

@@ -92,7 +92,7 @@ public suspend fun <T> ObservableSource<T>.awaitSingle(): T = awaitOne(Mode.SING
 
 // ------------------------ private ------------------------
 
-private fun CancellableContinuation<*>.disposeOnCompletion(d: Disposable) =
+internal fun CancellableContinuation<*>.disposeOnCompletion(d: Disposable) =
     invokeOnCompletion { d.dispose() }
 
 private enum class Mode(val s: String) {

reactive/kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxScheduler.kt

@@ -0,0 +1,59 @@
+/*
+ * 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.rx2
+
+import io.reactivex.Scheduler
+import kotlinx.coroutines.experimental.CancellableContinuation
+import kotlinx.coroutines.experimental.CoroutineDispatcher
+import kotlinx.coroutines.experimental.Delay
+import kotlinx.coroutines.experimental.DisposableHandle
+import java.util.concurrent.TimeUnit
+import kotlin.coroutines.experimental.CoroutineContext
+
+/**
+ * Converts an instance of [Scheduler] to an implementation of [CoroutineDispatcher]
+ * and provides native [delay][Delay.delay] support.
+ */
+public fun Scheduler.asCoroutineDispatcher() = SchedulerCoroutineDispatcher(this)
+
+/**
+ * Implements [CoroutineDispatcher] on top of an arbitrary [Scheduler].
+ * @param scheduler a scheduler.
+ */
+public class SchedulerCoroutineDispatcher(private val scheduler: Scheduler) : CoroutineDispatcher(), Delay {
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
+        scheduler.scheduleDirect(block)
+    }
+
+    override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+        val disposable = scheduler.scheduleDirect({
+            with(continuation) { resumeUndispatched(Unit) }
+        }, time, unit)
+        continuation.disposeOnCompletion(disposable)
+    }
+
+    override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
+        val disposable = scheduler.scheduleDirect(block, time, unit)
+        return object : DisposableHandle {
+            override fun dispose() {
+                disposable.dispose()
+            }
+        }
+    }
+
+    override fun toString(): String = scheduler.toString()
+}

reactive/kotlinx-coroutines-rx2/src/test/kotlin/kotlinx/coroutines/experimental/rx2/SchedulerTest.kt

@@ -0,0 +1,44 @@
+/*
+ * 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.rx2
+
+import io.reactivex.schedulers.Schedulers
+import kotlinx.coroutines.experimental.*
+import org.hamcrest.core.IsEqual
+import org.hamcrest.core.IsNot
+import org.junit.Assert.assertThat
+import org.junit.Test
+
+class SchedulerTest : TestBase() {
+    @Test
+    fun testIoScheduler(): Unit = runBlocking {
+        expect(1)
+        val mainThread = Thread.currentThread()
+        run(Schedulers.io().asCoroutineDispatcher()) {
+            val t1 = Thread.currentThread()
+            println(t1)
+            assertThat(t1, IsNot(IsEqual(mainThread)))
+            expect(2)
+            delay(100)
+            val t2 = Thread.currentThread()
+            println(t2)
+            assertThat(t2, IsNot(IsEqual(mainThread)))
+            expect(3)
+        }
+        finish(4)
+    }
+}