Sagas rework

6 files changed, 91 insertions, 83 deletions

diff --git a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/SevenWondersUi.kt b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/SevenWondersUi.kt
index fafa2333..d8dee5df 100644
--- a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/SevenWondersUi.kt
+++ b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/SevenWondersUi.kt
@@ -1,5 +1,7 @@
 package org.luxons.sevenwonders.ui
 
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.promise
 import org.luxons.sevenwonders.ui.components.application
 import org.luxons.sevenwonders.ui.redux.SwState
 import org.luxons.sevenwonders.ui.redux.configureStore
@@ -38,6 +40,10 @@ private fun initializeAndRender(rootElement: Element) {
 private fun initRedux(): Store<SwState, RAction, WrapperAction> {
     val sagaManager = SagaManager<SwState, RAction, WrapperAction>()
     val store = configureStore(sagaManager = sagaManager)
-    sagaManager.startSaga(rootSaga())
+    GlobalScope.promise {
+        sagaManager.launchSaga(this) {
+            rootSaga()
+        }
+    }
     return store
 }
diff --git a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt
index cd9e0cf8..b4c99827 100644
--- a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt
+++ b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt
@@ -1,28 +1,29 @@
 package org.luxons.sevenwonders.ui.redux.sagas
 
-import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
 import org.luxons.sevenwonders.client.SevenWondersSession
 import org.luxons.sevenwonders.model.api.LobbyDTO
 import org.luxons.sevenwonders.ui.redux.EnterLobbyAction
 import org.luxons.sevenwonders.ui.redux.RequestCreateGameAction
 import org.luxons.sevenwonders.ui.redux.RequestJoinGameAction
-import org.luxons.sevenwonders.ui.redux.SwState
 import org.luxons.sevenwonders.ui.redux.UpdateGameListAction
 import org.luxons.sevenwonders.ui.redux.UpdateLobbyAction
-import redux.RAction
-import redux.WrapperAction
+import kotlin.coroutines.coroutineContext
 
-fun gameBrowserSaga(session: SevenWondersSession) = saga<SwState, RAction, WrapperAction> {
-    val watchGamesJob = fork(watchGames(session))
-    val watchCreateGameJob = fork(watchCreateGame(session))
-    val watchJoinGameJob = fork(watchJoinGame(session))
+suspend fun SwSagaContext.gameBrowserSaga(session: SevenWondersSession) {
+    coroutineScope {
+        launch { watchGames(session) }
+        launch { watchCreateGame(session) }
+        launch { watchJoinGame(session) }
+    }
 }
 
-private fun watchGames(session: SevenWondersSession) = saga<SwState, RAction, WrapperAction> {
+private suspend fun SwSagaContext.watchGames(session: SevenWondersSession) {
     val gamesSubscription = session.watchGames()
     for (lobbies in gamesSubscription.messages) {
-        if (!isActive) {
+        if (!coroutineContext.isActive) {
             gamesSubscription.unsubscribe()
             break
         }
@@ -30,23 +31,26 @@ private fun watchGames(session: SevenWondersSession) = saga<SwState, RAction, Wr
     }
 }
 
-private fun watchCreateGame(session: SevenWondersSession) =
-        actionHandlerSaga<SwState, RAction, WrapperAction, RequestCreateGameAction> {
+private suspend fun SwSagaContext.watchCreateGame(session: SevenWondersSession) =
+        onEach<RequestCreateGameAction> {
             val lobby = session.createGame(it.gameName)
             handleGameJoined(session, lobby)
         }
 
-private fun watchJoinGame(session: SevenWondersSession) = actionHandlerSaga<SwState, RAction, WrapperAction, RequestJoinGameAction> {
-    val lobby = session.joinGame(it.gameId)
-    handleGameJoined(session, lobby)
-}
+private suspend fun SwSagaContext.watchJoinGame(session: SevenWondersSession) =
+        onEach<RequestJoinGameAction> {
+            val lobby = session.joinGame(it.gameId)
+            handleGameJoined(session, lobby)
+        }
 
-private fun SagaContext<SwState, RAction, WrapperAction>.handleGameJoined(
+private suspend fun SwSagaContext.handleGameJoined(
     session: SevenWondersSession,
     lobby: LobbyDTO
 ) {
     dispatch(UpdateLobbyAction(lobby))
     dispatch(EnterLobbyAction(lobby.id))
-    fork(lobbySaga(session, lobby.id))
-    // TODO push /lobby/{lobby.id}
+    coroutineScope {
+        launch { lobbySaga(session, lobby.id) }
+        // TODO push /lobby/{lobby.id}
+    }
 }
diff --git a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt
index 24a9e1b4..29242782 100644
--- a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt
+++ b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt
@@ -1,15 +1,19 @@
 package org.luxons.sevenwonders.ui.redux.sagas
 
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import org.luxons.sevenwonders.client.SevenWondersSession
 import org.luxons.sevenwonders.ui.redux.SwState
 import redux.RAction
 import redux.WrapperAction
 
-fun gameSaga(session: SevenWondersSession, gameId: Long) = saga<SwState, RAction, WrapperAction> {
-    fork(watchPlayerReady(session, gameId))
+suspend fun SwSagaContext.gameSaga(session: SevenWondersSession, gameId: Long) {
+    coroutineScope {
+        launch { watchPlayerReady(session, gameId) }
+    }
 }
 
-fun watchPlayerReady(session: SevenWondersSession, gameId: Long) = saga<SwState, RAction, WrapperAction> {
+private suspend fun SwSagaContext.watchPlayerReady(session: SevenWondersSession, gameId: Long) {
     session.watchPlayerReady(gameId)
 }
 
diff --git a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt
index 89e8e890..a58aeaa9 100644
--- a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt
+++ b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt
@@ -1,33 +1,39 @@
 package org.luxons.sevenwonders.ui.redux.sagas
 
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import org.luxons.sevenwonders.client.SevenWondersSession
 import org.luxons.sevenwonders.ui.redux.EnterGameAction
 import org.luxons.sevenwonders.ui.redux.RequestStartGameAction
-import org.luxons.sevenwonders.ui.redux.SwState
 import org.luxons.sevenwonders.ui.redux.UpdateLobbyAction
-import redux.RAction
-import redux.WrapperAction
 
-fun lobbySaga(session: SevenWondersSession, lobbyId: Long) = saga<SwState, RAction, WrapperAction> {
-    fork(watchLobbyUpdates(session, lobbyId))
-    fork(watchGameStart(session, lobbyId))
-    fork(startGame(session))
+suspend fun SwSagaContext.lobbySaga(session: SevenWondersSession, lobbyId: Long) {
+    coroutineScope {
+        launch { watchLobbyUpdates(session, lobbyId) }
+        launch { handleGameStart(session, lobbyId) }
+        launch { watchStartGame(session) }
+    }
 }
 
-private fun watchLobbyUpdates(session: SevenWondersSession, lobbyId: Long) = saga<SwState, RAction, WrapperAction> {
+private suspend fun SwSagaContext.watchLobbyUpdates(session: SevenWondersSession, lobbyId: Long) {
     val lobbyUpdates = session.watchLobbyUpdates(lobbyId)
     for (lobby in lobbyUpdates.messages) {
         dispatch(UpdateLobbyAction(lobby))
     }
 }
 
-private fun watchGameStart(session: SevenWondersSession, lobbyId: Long) = saga<SwState, RAction, WrapperAction> {
+private suspend fun SwSagaContext.handleGameStart(session: SevenWondersSession, lobbyId: Long) {
     val gameStartSubscription = session.watchGameStart(lobbyId)
     gameStartSubscription.messages.receive()
     dispatch(EnterGameAction(lobbyId))
-    // TODO push /game/{lobby.id}
+
+    coroutineScope {
+        launch { gameSaga(session, lobbyId) }
+
+        // TODO push /game/{lobby.id}
+    }
 }
 
-private fun startGame(session: SevenWondersSession) = actionHandlerSaga<SwState, RAction, WrapperAction, RequestStartGameAction> {
+private suspend fun SwSagaContext.watchStartGame(session: SevenWondersSession) = onEach<RequestStartGameAction> {
     session.startGame()
 }
diff --git a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt
index 96311f98..08769202 100644
--- a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt
+++ b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt
@@ -1,5 +1,7 @@
 package org.luxons.sevenwonders.ui.redux.sagas
 
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import org.luxons.sevenwonders.client.SevenWondersClient
 import org.luxons.sevenwonders.model.api.SEVEN_WONDERS_WS_ENDPOINT
 import org.luxons.sevenwonders.ui.redux.RequestChooseName
@@ -8,13 +10,17 @@ import org.luxons.sevenwonders.ui.redux.SwState
 import redux.RAction
 import redux.WrapperAction
 
-fun rootSaga() = saga<SwState, RAction, WrapperAction> {
+typealias SwSagaContext = SagaContext<SwState, RAction, WrapperAction>
+
+suspend fun SwSagaContext.rootSaga() {
     val action = next<RequestChooseName>()
     val session = SevenWondersClient().connect(SEVEN_WONDERS_WS_ENDPOINT)
 
-    fork(gameBrowserSaga(session))
+    coroutineScope {
+        launch { gameBrowserSaga(session) }
 
-    val player = session.chooseName(action.playerName)
-    dispatch(SetCurrentPlayerAction(player))
-    // push /games
+        val player = session.chooseName(action.playerName)
+        dispatch(SetCurrentPlayerAction(player))
+        // push /games
+    }
 }
diff --git a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt
index 2aa677fe..c7184cea 100644
--- a/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt
+++ b/sw-ui-kt/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt
@@ -1,5 +1,6 @@
 package org.luxons.sevenwonders.ui.redux.sagas
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.FlowPreview
@@ -12,7 +13,7 @@ import kotlinx.coroutines.promise
 import redux.Middleware
 import redux.MiddlewareApi
 import redux.RAction
-import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.coroutineContext
 
 @UseExperimental(ExperimentalCoroutinesApi::class)
 class SagaManager<S, A : RAction, R>(
@@ -44,61 +45,43 @@ class SagaManager<S, A : RAction, R>(
         GlobalScope.promise { actions.send(action) }
     }
 
-    fun startSaga(saga: Saga<S, A, R>): Job {
-        check(::context.isInitialized) {
-            "Before running a Saga, you must mount the Saga middleware on the Store using applyMiddleware"
-        }
-        return context.launch {
-            saga.execute(context)
+    fun launchSaga(coroutineScope: CoroutineScope, saga: suspend SagaContext<S, A, R>.() -> Unit): Job {
+        checkMiddlewareApplied()
+        return coroutineScope.launch {
+            context.saga()
         }
     }
 
-    suspend fun runSaga(saga: Saga<S, A, R>) {
+    suspend fun runSaga(saga: suspend SagaContext<S, A, R>.() -> Unit) {
+        checkMiddlewareApplied()
+        context.saga()
+    }
+
+    private fun checkMiddlewareApplied() {
         check(::context.isInitialized) {
             "Before running a Saga, you must mount the Saga middleware on the Store using applyMiddleware"
         }
-        saga.execute(context)
     }
 }
 
-fun <S, A : RAction, R> saga(block: suspend SagaContext<S, A, R>.() -> Unit) = Saga(block)
-
-inline fun <S, A : RAction, R, reified AT : A> actionHandlerSaga(
-    noinline block: suspend SagaContext<S, A, R>.(AT) -> Unit
-) = saga<S, A, R> {
-    onEach<AT> {
-        block(it)
-    }
-}
-
-class Saga<S, A : RAction, R>(private val body: suspend SagaContext<S, A, R>.() -> Unit) {
-
-    internal suspend fun execute(context: SagaContext<S, A, R>) = context.body()
-}
-
 @UseExperimental(FlowPreview::class, ExperimentalCoroutinesApi::class)
 class SagaContext<S, A : RAction, R>(
-    private val reduxApi: MiddlewareApi<S, A, R>,
-    private val actions: BroadcastChannel<A>
-): CoroutineScope {
-
-    private val job = Job()
-
-    override val coroutineContext: CoroutineContext
-        get() = job
-
+    private val reduxApi: MiddlewareApi<S, A, R>, private val actions: BroadcastChannel<A>
+) {
+    /**
+     * Dispatches the given redux [action].
+     */
     fun dispatch(action: A) {
         reduxApi.dispatch(action)
     }
 
     /**
      * Starts a concurrent coroutine that executes [handle] on every action dispatched.
-     * Returns an "unsubscribe" function.
      */
     suspend fun onEach(handle: suspend SagaContext<S, A, R>.(A) -> Unit) {
         val channel = actions.openSubscription()
         for (a in channel) {
-            if (!isActive) {
+            if (!coroutineContext.isActive) {
                 channel.cancel()
                 break
             }
@@ -107,13 +90,12 @@ class SagaContext<S, A : RAction, R>(
     }
 
     /**
-     * Starts a concurrent coroutine that executes [handle] on every action dispatched of the given type.
-     * Returns an "unsubscribe" function.
+     * Starts a concurrent coroutine that executes [handle] on every action dispatched of the type [T].
      */
-    suspend inline fun <reified AT : A> onEach(
-        crossinline handle: suspend SagaContext<S, A, R>.(AT) -> Unit
+    suspend inline fun <reified T : A> onEach(
+        crossinline handle: suspend SagaContext<S, A, R>.(T) -> Unit
     ) = onEach {
-        if (it is AT) {
+        if (it is T) {
             handle(it)
         }
     }
@@ -124,6 +106,10 @@ class SagaContext<S, A : RAction, R>(
     suspend fun next(predicate: (A) -> Boolean): A {
         val channel = actions.openSubscription()
         for (a in channel) {
+            if (!coroutineContext.isActive) {
+                channel.cancel()
+                throw CancellationException("The expected action was not received before cancellation")
+            }
             if (predicate(a)) {
                 channel.cancel()
                 return a
@@ -136,8 +122,4 @@ class SagaContext<S, A : RAction, R>(
      * Suspends until the next action of type [T] is dispatched, and returns that action.
      */
     suspend inline fun <reified T : A> next(): T = next { it is T } as T
-
-    fun fork(saga: Saga<S, A, R>) = SagaContext(reduxApi, actions).launch {
-        saga.execute(this@SagaContext)
-    }
 }