Update Reactor support to leverage Bismuth release train

PR #141

Author: sdeleuze

reactive/kotlinx-coroutines-reactor/pom.xml

@@ -64,7 +64,7 @@
             <dependency>
                 <groupId>io.projectreactor</groupId>
                 <artifactId>reactor-bom</artifactId>
-                <version>Aluminium-SR3</version>
+                <version>Bismuth-RELEASE</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
@@ -77,7 +77,7 @@
             <artifactId>reactor-core</artifactId>
         </dependency>
         <dependency>
-            <groupId>io.projectreactor.addons</groupId>
+            <groupId>io.projectreactor</groupId>
             <artifactId>reactor-test</artifactId>
             <scope>test</scope>
         </dependency>

reactive/kotlinx-coroutines-reactor/src/main/kotlin/kotlinx/coroutines/experimental/reactor/Scheduler.kt

@@ -1,13 +1,8 @@
 package kotlinx.coroutines.experimental.reactor
 
-import kotlinx.coroutines.experimental.CancellableContinuation
-import kotlinx.coroutines.experimental.CoroutineDispatcher
-import kotlinx.coroutines.experimental.Delay
-import kotlinx.coroutines.experimental.DisposableHandle
-import kotlinx.coroutines.experimental.disposeOnCompletion
-import reactor.core.Cancellation
+import kotlinx.coroutines.experimental.*
+import reactor.core.Disposable
 import reactor.core.scheduler.Scheduler
-import reactor.core.scheduler.TimedScheduler
 import java.util.concurrent.TimeUnit
 import kotlin.coroutines.experimental.CoroutineContext
 
@@ -16,45 +11,31 @@ import kotlin.coroutines.experimental.CoroutineContext
  */
 fun Scheduler.asCoroutineDispatcher() = SchedulerCoroutineDispatcher(this)
 
-/**
- * Converts an instance of [TimedScheduler] to an implementation of [CoroutineDispatcher]
- * and provides native [delay][Delay.delay] support.
- */
-fun TimedScheduler.asCoroutineDispatcher() = TimedSchedulerCoroutineDispatcher(this)
-
 /**
  * Implements [CoroutineDispatcher] on top of an arbitrary [Scheduler].
  * @param scheduler a scheduler.
  */
-open class SchedulerCoroutineDispatcher(private val scheduler: Scheduler) : CoroutineDispatcher() {
+open class SchedulerCoroutineDispatcher(private val scheduler: Scheduler) : CoroutineDispatcher(), Delay {
     override fun dispatch(context: CoroutineContext, block: Runnable) {
         scheduler.schedule(block)
     }
 
-    override fun toString(): String = scheduler.toString()
-    override fun equals(other: Any?): Boolean = other is SchedulerCoroutineDispatcher && other.scheduler === scheduler
-    override fun hashCode(): Int = System.identityHashCode(scheduler)
-}
-
-/**
- * Implements [CoroutineDispatcher] on top of an arbitrary [TimedScheduler].
- * @param scheduler a timed scheduler.
- */
-open class TimedSchedulerCoroutineDispatcher(private val scheduler: TimedScheduler) : SchedulerCoroutineDispatcher(scheduler), Delay {
-
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
         val disposable = scheduler.schedule({
             with(continuation) { resumeUndispatched(Unit) }
         }, time, unit)
-
         continuation.disposeOnCompletion(disposable.asDisposableHandle())
     }
 
     override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
         scheduler.schedule(block, time, unit).asDisposableHandle()
+
+    override fun toString(): String = scheduler.toString()
+    override fun equals(other: Any?): Boolean = other is SchedulerCoroutineDispatcher && other.scheduler === scheduler
+    override fun hashCode(): Int = System.identityHashCode(scheduler)
 }
 
-private fun Cancellation.asDisposableHandle(): DisposableHandle =
-        object : DisposableHandle {
-            override fun dispose() = this@asDisposableHandle.dispose()
-        }
+private fun Disposable.asDisposableHandle(): DisposableHandle =
+    object : DisposableHandle {
+        override fun dispose() = this@asDisposableHandle.dispose()
+    }

reactive/kotlinx-coroutines-reactor/src/test/kotlin/kotlinx/coroutines/experimental/reactor/FluxSingleTest.kt

@@ -17,16 +17,13 @@
 package kotlinx.coroutines.experimental.reactor
 
 import kotlinx.coroutines.experimental.CommonPool
-import kotlinx.coroutines.experimental.reactive.awaitFirst
-import kotlinx.coroutines.experimental.reactive.awaitFirstOrDefault
-import kotlinx.coroutines.experimental.reactive.awaitLast
-import kotlinx.coroutines.experimental.reactive.awaitSingle
-import kotlinx.coroutines.experimental.reactive.consumeEach
+import kotlinx.coroutines.experimental.reactive.*
 import kotlinx.coroutines.experimental.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Assert.fail
 import org.junit.Test
 import reactor.core.publisher.Flux
+import java.time.Duration.ofMillis
 
 /**
  * Tests emitting single item with [flux].
@@ -62,7 +59,7 @@ class FluxSingleTest {
     @Test
     fun testSingleWithDelay() {
         val flux = flux(CommonPool) {
-            send(Flux.just("O").delayMillis(50).awaitSingle() + "K")
+            send(Flux.just("O").delayElements(ofMillis(50)).awaitSingle() + "K")
         }
 
         checkSingleValue(flux) {

reactive/kotlinx-coroutines-reactor/src/test/kotlin/kotlinx/coroutines/experimental/reactor/MonoTest.kt

@@ -32,6 +32,7 @@ import org.junit.Before
 import org.junit.Test
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.time.Duration.ofMillis
 
 /**
  * Tests emitting single item with [mono].
@@ -87,7 +88,7 @@ class MonoTest : TestBase() {
             null
         }
         expect(2)
-        mono.subscribe ({}, { throw it }, {
+        mono.subscribe({}, { throw it }, {
             expect(5)
         })
         expect(3)
@@ -148,7 +149,7 @@ class MonoTest : TestBase() {
     @Test
     fun testMonoWithDelay() {
         val mono = mono(CommonPool) {
-            Flux.just("O").delayMillis(50).awaitSingle() + "K"
+            Flux.just("O").delayElements(ofMillis(50)).awaitSingle() + "K"
         }
 
         checkMonoValue(mono) {