Merge pull request #922 from hangga/rxjava-to-coroutines

KTLN-524 - Convert RxJava Single to Kotlin Coroutine Deferred

Author: theangrydev

core-kotlin-modules/core-kotlin-concurrency-3/pom.xml

@@ -11,4 +11,31 @@
         <version>1.0.0-SNAPSHOT</version>
     </parent>
 
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.reactivex.rxjava3</groupId>
+            <artifactId>rxkotlin</artifactId>
+            <version>${rxkotlin.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-rx3</artifactId>
+            <version>${rxcoroutines.version}</version>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <rxkotlin.version>3.0.1</rxkotlin.version>
+        <rxcoroutines.version>1.8.1</rxcoroutines.version>
+        <junit.version>4.13.2</junit.version>
+    </properties>
+    
 </project>

core-kotlin-modules/core-kotlin-concurrency-3/src/test/kotlin/com/baeldung/singlerxjavatocoroutinedeferred/SingleRxJavaToCoroutineDeferredUnitTest.kt

@@ -0,0 +1,80 @@
+package com.baeldung.singlerxjavatocoroutinedeferred
+
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.schedulers.Schedulers
+import kotlinx.coroutines.*
+import kotlinx.coroutines.rx3.await
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlin.coroutines.suspendCoroutine
+
+class SingleRxJavaToCoroutineDeferredUnitTest {
+
+    data class Product(val id: Int, val name: String, val price: Double)
+
+    private val allProducts = listOf(
+        Product(1, "Samsung", 1200.0),
+        Product(2, "Oppo", 800.0),
+        Product(3, "Nokia", 450.0),
+        Product(4, "Lenovo", 550.0),
+        Product(5, "ASUS", 400.0)
+    )
+
+    private fun getFilteredProducts(): Single<List<Product>> {
+        return Single.just(
+            allProducts
+        ).map { products ->
+            products.sortedBy { it.price }.filter { it.price > 500 }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    private suspend fun Deferred<*>.assertOver500AndSorted() {
+        assertThat(this.await() as List<*>).containsExactly(
+            Product(4, "Lenovo", 550.0),
+            Product(2, "Oppo", 800.0),
+            Product(1, "Samsung", 1200.0)
+        )
+    }
+
+    @Test
+    fun `using async and blockingGet`() = runBlocking {
+        val deferred = async { getFilteredProducts().blockingGet() }
+        deferred.assertOver500AndSorted()
+    }
+
+    @Test
+    fun `using subscribe and CompletableDeferred`() = runBlocking {
+        val deferred = CompletableDeferred<List<Product>>()
+        getFilteredProducts().subscribe(deferred::complete, deferred::completeExceptionally)
+        deferred.assertOver500AndSorted()
+    }
+
+    @Test
+    fun `using suspendCoroutines`(): Unit = runBlocking {
+        val deferred = async {
+            suspendCoroutine { continuation ->
+                getFilteredProducts().subscribe(continuation::resume, continuation::resumeWithException)
+            }
+        }
+        deferred.assertOver500AndSorted()
+    }
+
+    @Test
+    fun `using suspendCancellableCoroutine`(): Unit = runBlocking {
+        val deferred = async {
+            suspendCancellableCoroutine { continuation ->
+                getFilteredProducts().subscribe(continuation::resume, continuation::resumeWithException)
+            }
+        }
+        deferred.assertOver500AndSorted()
+    }
+
+    @Test
+    fun `using Kotlin Coroutines Rx3`() = runBlocking {
+        val deferred = async { getFilteredProducts().await() }
+        deferred.assertOver500AndSorted()
+    }
+
+}