issue #38bugfix issue #38 enforce same thread

buildSrc/src/main/kotlin/Libs.kt

@@ -180,4 +180,10 @@ object Libs {
 
   const val spek_runner_junit5: String = "org.spekframework.spek2:spek-runner-junit5:" +
       Versions.spek
+
+  const val kotlin_coroutines_jvm: String = "org.jetbrains.kotlinx:kotlinx-coroutines-core:" +
+          Versions.coroutines
+
+  const val kotlin_coroutines_test: String =  "org.jetbrains.kotlinx:kotlinx-coroutines-test:" +
+          Versions.coroutines
 }

buildSrc/src/main/kotlin/Versions.kt

@@ -43,6 +43,7 @@ object Versions {
 
   const val recycler_view = "1.0.0"
 
+  const val coroutines = "1.3.3"
 
   /**
    *

lib/build.gradle

@@ -7,7 +7,7 @@ apply plugin: 'kotlin-multiplatform'
 archivesBaseName = 'redux-kotlin'
 
 group 'org.reduxkotlin'
-version '0.3.1'
+version '0.3.2'
 
 kotlin {
     jvm()
@@ -59,6 +59,8 @@ kotlin {
             dependencies {
                 implementation kotlin("test")
                 implementation kotlin("test-junit")
+                implementation Libs.kotlin_coroutines_test
+                implementation Libs.kotlin_coroutines_jvm
                 implementation Libs.spek_dsl_jvm
                 implementation Libs.atrium_cc_en_gb_robstoll
                 implementation Libs.mockk

lib/src/commonMain/kotlin/org/reduxkotlin/CreateStore.kt

@@ -2,6 +2,7 @@ package org.reduxkotlin
 
 import org.reduxkotlin.utils.getThreadName
 import org.reduxkotlin.utils.isPlainObject
+import org.reduxkotlin.utils.stripCoroutineName
 
 /**
  * Creates a Redux store that holds the state tree.
@@ -45,8 +46,8 @@ fun <State> createStore(
     var currentListeners = mutableListOf<() -> Unit>()
     var nextListeners = currentListeners
     var isDispatching = false
-    val storeThreadName = getThreadName()
-    fun isSameThread() = getThreadName() == storeThreadName
+    val storeThreadName = stripCoroutineName(getThreadName())
+    fun isSameThread() = stripCoroutineName(getThreadName()) == storeThreadName
     fun checkSameThread() = check(isSameThread()) {
         """You may not call the store from a thread other than the thread on which it was created.
             |This includes: getState(), dispatch(), subscribe(), and replaceReducer()

lib/src/commonMain/kotlin/org/reduxkotlin/utils/ThreadUtil.kt

@@ -5,4 +5,14 @@ const val UNKNOWN_THREAD_NAME = "UNKNOWN_THREAD_NAME"
 /**
  * Returns the name of the current thread.
  */
-expect fun getThreadName(): String
+expect fun getThreadName(): String
+
+/**
+ * Thread name may have '@coroutine#n' appended to it.
+ * This strips the suffix so we can compare threads.
+ */
+fun stripCoroutineName(threadName: String): String {
+    val lastIndex = threadName.lastIndexOf('@')
+    return if (lastIndex < 0) threadName
+        else threadName.substring(0, lastIndex)
+}

lib/src/jvmTest/kotlin/org/reduxkotlin/util/ThreadUtilSpec.kt

@@ -1,14 +1,23 @@
 package org.reduxkotlin.util
 
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.test.setMain
 import org.reduxkotlin.*
 import org.spekframework.spek2.Spek
 import org.spekframework.spek2.style.specification.describe
 import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
 import kotlin.IllegalStateException
-import kotlin.test.assertNotNull
-import kotlin.test.assertNull
+import kotlin.system.measureTimeMillis
+import kotlin.test.*
+
 
 object ThreadUtilSpec : Spek({
+    val mainThreadSurrogate = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
+    Dispatchers.setMain(mainThreadSurrogate)
+
     describe("createStore") {
         val store = createStore(
             todos, TestState(
@@ -28,21 +37,75 @@ object ThreadUtilSpec : Spek({
             ensureSameThread { store.dispatch(Any()) }
         }
         it("ensure same thread on replaceReducer") {
-            ensureSameThread { store.replaceReducer { state, action ->  state } }
+            ensureSameThread { store.replaceReducer { state, action -> state } }
         }
         it("ensure same thread on subscribe") {
             ensureSameThread { store.subscribe { } }
         }
+        it("enforces same thread when thread name appends coroutine name") {
+            val middleware = TestMiddleware()
+
+            runBlocking {
+                CoroutineScope(Dispatchers.Main).async {
+                    val store = createStore(
+                        testReducer,
+                        TestState(),
+                        applyMiddleware(middleware.middleware)
+                    )
+
+                    store.dispatch(Any())
+                }.await()
+                Thread.sleep(2000)
+                assertFalse(middleware.failed)
+            }
+        }
+        it("increments massively") {
+            suspend fun massiveRun(action: suspend () -> Unit) {
+                val n = 100  // number of coroutines to launch
+                val k = 1000 // times an action is repeated by each coroutine
+                val time = measureTimeMillis {
+                    coroutineScope {
+                        // scope for coroutines
+                        repeat(n) {
+                            launch {
+                                repeat(k) { action() }
+                            }
+                        }
+                    }
+                }
+                println("Completed ${n * k} actions in $time ms")
+            }
+
+
+            val counterContext = newSingleThreadContext("CounterContext")
+
+            lateinit var store: Store<TestCounterState>
+            runBlocking {
+                withContext(counterContext) {
+                    store = createStore(counterReducer, TestCounterState())
+                }
+            }
+            runBlocking {
+                withContext(counterContext) {
+                    massiveRun {
+                        store.dispatch(Increment())
+                    }
+                }
+                withContext(counterContext) {
+                    assertEquals(100000, store.state.counter)
+                }
+            }
+        }
     }
 })
 
-private fun ensureSameThread(getState: () -> Any) {
+private fun ensureSameThread(testFun: () -> Any) {
     val latch = CountDownLatch(1)
     var exception: java.lang.IllegalStateException? = null
     var state: Any? = null
 
     val newThread = Thread {
-        state = getState()
+        state = testFun()
     }
 
     newThread.setUncaughtExceptionHandler { thread, throwable ->
@@ -55,4 +118,43 @@ private fun ensureSameThread(getState: () -> Any) {
 
     assertNotNull(exception)
     assertNull(state)
-}
+}
+
+val testReducer: Reducer<TestState> = { state, action -> state }
+
+/**
+ * Used as a test for when Thread.currentThread.name returns the
+ * thread name + '@coroutine#'.
+ * See issue #38 https://github.com/reduxkotlin/redux-kotlin/issues/38
+ */
+class TestMiddleware {
+    var failed = false
+    val middleware = middleware<TestState> { store, next, action ->
+        CoroutineScope(Dispatchers.Main).launch {
+            flow {
+                delay(1000) // simulate api call
+                emit("Text Response")
+            }.collect { response ->
+                store.dispatch("")
+            }
+        }
+        try {
+            next(action)
+        } catch (e: Exception) {
+            e.printStackTrace()
+            failed = true
+            Unit
+        }
+    }
+}
+
+class Increment
+
+data class TestCounterState(val counter: Int = 0)
+
+val counterReducer = { state: TestCounterState, action: Any ->
+    when (action) {
+        is Increment -> state.copy(counter = state.counter + 1)
+        else -> state
+    }
+}