Merge pull request #1 from PandoraMedia/rxJava-interop

Add samples demonstrating interoperability with RxJava

build.gradle

@@ -14,6 +14,8 @@ repositories {
 dependencies {
     implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
     implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.0'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-rx2:1.3.0'
+    implementation "io.reactivex.rxjava2:rxjava:2.2.7"
     implementation "io.reactivex.rxjava3:rxjava:3.0.0-RC2"
     testCompile group: 'junit', name: 'junit', version: '4.12'
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,6 @@
+#Wed Sep 18 11:29:24 PDT 2019
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.3-all.zip
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.3-bin.zip
-zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
+zipStoreBase=GRADLE_USER_HOME

src/main/kotlin/interop/completable/CompletableInterop.kt

@@ -0,0 +1,17 @@
+package interop.completable
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.rx2.rxCompletable
+import kotlin.coroutines.CoroutineContext
+
+class CompletableInterop(
+    override val coroutineContext: CoroutineContext = Dispatchers.Default
+): CoroutineScope {
+    fun doWork() = CoroutineScope(coroutineContext).rxCompletable {
+        delay(1000)
+        println("Hello from the coroutine!")
+    }
+}
+

src/main/kotlin/interop/completable/TestCompletableInterop.kt

@@ -0,0 +1,9 @@
+package interop.completable
+
+fun main() {
+    println("Creating an instance of CompletableInterop")
+    val completableInterop = CompletableInterop()
+    completableInterop.doWork().subscribe()
+
+    Thread.sleep(2000) // Sleep the main thread to await the completion of the background thread
+}

src/main/kotlin/interop/flowable/FlowableInterop.kt

@@ -0,0 +1,17 @@
+package interop.flowable
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.rx2.rxFlowable
+import kotlin.coroutines.CoroutineContext
+
+class FlowableInterop(
+    override val coroutineContext: CoroutineContext = Dispatchers.Default
+): CoroutineScope {
+    fun getFlowable() = CoroutineScope(coroutineContext).rxFlowable {
+        (1..10).map {
+            send(it)
+        }
+    }
+}

src/main/kotlin/interop/flowable/TestFlowableInterop.kt

@@ -0,0 +1,13 @@
+package interop.flowable
+
+fun main() {
+    println("Creating FlowableInterop")
+    val flowableInterop = FlowableInterop()
+    flowableInterop.getFlowable()
+        .subscribe({
+            println(it)
+            Thread.sleep(1000) // notice that the producing coroutine's backpressure is handled
+        },::println)
+
+    Thread.sleep(11000) // allow background work to occur before main thread exits
+}

src/main/kotlin/interop/maybe/MaybeInterop.kt

@@ -0,0 +1,15 @@
+package interop.maybe
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.rx2.rxMaybe
+import kotlin.coroutines.CoroutineContext
+import kotlin.random.Random
+
+class MaybeInterop(
+    override val coroutineContext: CoroutineContext = Dispatchers.Default
+): CoroutineScope {
+    fun doWorkMaybe() = CoroutineScope(coroutineContext).rxMaybe {
+        return@rxMaybe if (Random.Default.nextBoolean()) "A value from the function" else null
+    }
+}

src/main/kotlin/interop/maybe/TestMaybeInterop.kt

@@ -0,0 +1,9 @@
+package interop.maybe
+
+fun main() {
+    println("Creating an instance of MaybeInterop")
+    val maybeInterop = MaybeInterop()
+    maybeInterop.doWorkMaybe().subscribe(::println, ::println)
+
+    Thread.sleep(2000) // Sleep the main thread to await the completion of the background thread
+}

src/main/kotlin/interop/observable/ObservableInterop.kt

@@ -0,0 +1,18 @@
+package interop.observable
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.rx2.rxObservable
+import kotlin.coroutines.CoroutineContext
+
+class ObservableInterop(
+    override val coroutineContext: CoroutineContext = Dispatchers.Default
+): CoroutineScope {
+    fun getStream() = CoroutineScope(coroutineContext).rxObservable {
+        (1..10).map {
+            delay(100)
+            send(it)
+        }
+    }
+}

src/main/kotlin/interop/observable/TestObservableInterop.kt

@@ -0,0 +1,9 @@
+package interop.observable
+
+fun main() {
+    println("Creating ObservableInterop")
+    val observableInterop = ObservableInterop()
+    observableInterop.getStream().subscribe(::println, ::println)
+
+    Thread.sleep(2000) // Sleep main thread to allow coroutine to complete
+}

src/main/kotlin/interop/single/SingleInterop.kt

@@ -0,0 +1,15 @@
+package interop.single
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.rx2.rxSingle
+import kotlin.coroutines.CoroutineContext
+
+class SingleInterop(
+    override val coroutineContext: CoroutineContext = Dispatchers.Default
+): CoroutineScope {
+    fun doWork() = CoroutineScope(coroutineContext).rxSingle {
+        delay(1000)
+        return@rxSingle "Hello!"
+    }
+}
+

src/main/kotlin/interop/single/TestSingleInterop.kt

@@ -0,0 +1,9 @@
+package interop.single
+
+fun main() {
+    println("Creating an instance of SingleInterop")
+    val singleInterop = SingleInterop()
+    singleInterop.doWork().subscribe(::println, ::println)
+
+    Thread.sleep(2000) // Sleep the main thread to await the completion of the background thread
+}