Merge pull request #85 from babylonhealth/transform-fix

Moved basic `transform` to the background

Author: Mikolaj Leszczynski

orbit-2-core/src/main/java/com/babylon/orbit2/OrbitBasePlugin.kt

@@ -19,6 +19,7 @@ package com.babylon.orbit2
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.withContext
 
 internal class Transform<S : Any, E : Any, E2 : Any>(val block: Context<S, E>.() -> E2) :
     Operator<S, E2>
@@ -78,7 +79,9 @@ object OrbitBasePlugin : OrbitPlugin {
         return when (operator) {
             is Transform<*, *, *> -> flow.map {
                 with(operator as Transform<S, E, Any>) {
-                    createContext(it).block()
+                    withContext(containerContext.backgroundDispatcher) {
+                        createContext(it).block()
+                    }
                 }
             }
             is SideEffect<*, *, *> -> flow.onEach {

orbit-2-core/src/test/java/com/babylon/orbit2/BaseDslThreadingTest.kt

@@ -17,14 +17,16 @@
 package com.babylon.orbit2
 
 import com.appmattus.kotlinfixture.kotlinFixture
+import kotlinx.coroutines.newSingleThreadContext
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
 import java.util.concurrent.CountDownLatch
 
 internal class BaseDslThreadingTest {
 
     companion object {
-        const val EXPECTED_THREAD_PREFIX = "orbit"
+        const val ORBIT_THREAD_PREFIX = "orbit"
+        const val BACKGROUND_THREAD_PREFIX = "IO"
     }
 
     private val fixture = kotlinFixture()
@@ -38,19 +40,19 @@ internal class BaseDslThreadingTest {
         middleware.reducer(action)
 
         testStreamObserver.awaitCount(2)
-        assertThat(middleware.threadName).startsWith(EXPECTED_THREAD_PREFIX)
+        assertThat(middleware.threadName).startsWith(ORBIT_THREAD_PREFIX)
     }
 
     @Test
-    fun `transformer executes on orbit dispatcher`() {
+    fun `transformer executes on background dispatcher`() {
         val action = fixture<Int>()
         val middleware = BaseDslMiddleware()
         val testStreamObserver = middleware.container.stateStream.test()
 
         middleware.transformer(action)
 
         testStreamObserver.awaitCount(2)
-        assertThat(middleware.threadName).startsWith(EXPECTED_THREAD_PREFIX)
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     @Test
@@ -62,7 +64,7 @@ internal class BaseDslThreadingTest {
         middleware.postingSideEffect(action)
 
         testStreamObserver.awaitCount(1)
-        assertThat(middleware.threadName).startsWith(EXPECTED_THREAD_PREFIX)
+        assertThat(middleware.threadName).startsWith(ORBIT_THREAD_PREFIX)
     }
 
     @Test
@@ -74,14 +76,18 @@ internal class BaseDslThreadingTest {
 
         middleware.latch.await()
 
-        assertThat(middleware.threadName).startsWith(EXPECTED_THREAD_PREFIX)
+        assertThat(middleware.threadName).startsWith(ORBIT_THREAD_PREFIX)
     }
 
     private data class TestState(val id: Int)
 
     private class BaseDslMiddleware : Host<TestState, String> {
-        override val container = Container.create<TestState, String>(
-            TestState(42)
+
+        @Suppress("EXPERIMENTAL_API_USAGE")
+        override val container = RealContainer<TestState, String>(
+            initialState = TestState(42),
+            settings = Container.Settings(),
+            backgroundDispatcher = newSingleThreadContext(BACKGROUND_THREAD_PREFIX)
         )
         lateinit var threadName: String
         val latch = CountDownLatch(1)

orbit-2-coroutines/src/test/java/com/babylon/orbit2/OrbitCoroutinePluginDslThreadingTest.kt

@@ -17,17 +17,21 @@
 package com.babylon.orbit2
 
 import com.appmattus.kotlinfixture.kotlinFixture
-import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.newSingleThreadContext
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
-import java.util.concurrent.Executors
 
 internal class OrbitCoroutinePluginDslThreadingTest {
+
+    companion object {
+        const val BACKGROUND_THREAD_PREFIX = "IO"
+    }
+
     private val fixture = kotlinFixture()
 
     @BeforeEach
@@ -50,7 +54,7 @@ internal class OrbitCoroutinePluginDslThreadingTest {
         middleware.suspend(action)
 
         testStreamObserver.awaitCount(2)
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     @Test
@@ -63,7 +67,7 @@ internal class OrbitCoroutinePluginDslThreadingTest {
         middleware.flow(action)
 
         testStreamObserver.awaitCount(5)
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     private data class TestState(val id: Int)
@@ -73,8 +77,7 @@ internal class OrbitCoroutinePluginDslThreadingTest {
         override val container: Container<TestState, String> = RealContainer(
             initialState = TestState(42),
             settings = Container.Settings(),
-            backgroundDispatcher = Executors.newSingleThreadExecutor { Thread(it, "IO") }
-                .asCoroutineDispatcher()
+            backgroundDispatcher = newSingleThreadContext(BACKGROUND_THREAD_PREFIX)
         )
         lateinit var threadName: String
 

orbit-2-rxjava2/src/test/java/com/babylon/orbit2/OrbitRxJava2PluginDslThreadingTest.kt

@@ -21,15 +21,19 @@ import io.reactivex.Completable
 import io.reactivex.Maybe
 import io.reactivex.Observable
 import io.reactivex.Single
-import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.newSingleThreadContext
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import java.util.concurrent.CountDownLatch
-import java.util.concurrent.Executors
 
 internal class OrbitRxJava2PluginDslThreadingTest {
+
+    companion object {
+        const val BACKGROUND_THREAD_PREFIX = "IO"
+    }
+
     private val fixture = kotlinFixture()
 
     @BeforeEach
@@ -52,7 +56,7 @@ internal class OrbitRxJava2PluginDslThreadingTest {
         middleware.single(action)
 
         testStreamObserver.awaitCount(2)
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     @Test
@@ -65,7 +69,7 @@ internal class OrbitRxJava2PluginDslThreadingTest {
         middleware.maybe(action)
 
         testStreamObserver.awaitCount(2)
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     @Test
@@ -77,7 +81,7 @@ internal class OrbitRxJava2PluginDslThreadingTest {
         middleware.maybeNot(action)
 
         middleware.latch.await()
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     @Test
@@ -90,7 +94,7 @@ internal class OrbitRxJava2PluginDslThreadingTest {
         middleware.completable(action)
 
         testStreamObserver.awaitCount(2)
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     @Test
@@ -103,7 +107,7 @@ internal class OrbitRxJava2PluginDslThreadingTest {
         middleware.observable(action)
 
         testStreamObserver.awaitCount(5)
-        assertThat(middleware.threadName).startsWith("IO")
+        assertThat(middleware.threadName).startsWith(BACKGROUND_THREAD_PREFIX)
     }
 
     private data class TestState(val id: Int)
@@ -113,8 +117,7 @@ internal class OrbitRxJava2PluginDslThreadingTest {
         override val container: Container<TestState, String> = RealContainer(
             initialState = TestState(42),
             settings = Container.Settings(),
-            backgroundDispatcher = Executors.newSingleThreadExecutor { Thread(it, "IO") }
-                .asCoroutineDispatcher()
+            backgroundDispatcher = newSingleThreadContext(BACKGROUND_THREAD_PREFIX)
         )
         lateinit var threadName: String
         val latch = CountDownLatch(1)