Update Reactor support to leverage Bismuth release train

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

@@ -5,9 +5,8 @@ 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 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 +15,32 @@ 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)
 
-    override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
-        val disposable = scheduler.schedule({
-            with(continuation) { resumeUndispatched(Unit) }
-        }, time, unit)
+		continuation.disposeOnCompletion(disposable.asDisposableHandle())
+	}
 
-        continuation.disposeOnCompletion(disposable.asDisposableHandle())
-    }
+	override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
+			scheduler.schedule(block, time, unit).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 =
+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

@@ -27,6 +27,7 @@ import org.junit.Assert.assertEquals
 import org.junit.Assert.fail
 import org.junit.Test
 import reactor.core.publisher.Flux
+import java.time.Duration.*
 
 /**
  * Tests emitting single item with [flux].
@@ -62,7 +63,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.*
 
 /**
  * Tests emitting single item with [mono].
@@ -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) {