Decouple some sagas from routes

Resolves:
https://github.com/joffrey-bion/seven-wonders/issues/13

9 files changed, 128 insertions, 192 deletions

diff --git a/sw-bot/src/main/kotlin/org/luxons/sevenwonders/bot/SevenWondersBot.kt b/sw-bot/src/main/kotlin/org/luxons/sevenwonders/bot/SevenWondersBot.kt
index 48a96f08..01106d21 100644
--- a/sw-bot/src/main/kotlin/org/luxons/sevenwonders/bot/SevenWondersBot.kt
+++ b/sw-bot/src/main/kotlin/org/luxons/sevenwonders/bot/SevenWondersBot.kt
@@ -9,6 +9,7 @@ import kotlinx.coroutines.flow.takeWhile
 import kotlinx.coroutines.withTimeout
 import org.luxons.sevenwonders.client.SevenWondersClient
 import org.luxons.sevenwonders.client.SevenWondersSession
+import org.luxons.sevenwonders.client.joinGameAndWaitLobby
 import org.luxons.sevenwonders.model.Action
 import org.luxons.sevenwonders.model.MoveType
 import org.luxons.sevenwonders.model.PlayerMove
@@ -37,7 +38,7 @@ class SevenWondersBot(
     suspend fun play(serverUrl: String, gameId: Long) = withTimeout(botConfig.globalTimeout) {
         val session = client.connect(serverUrl)
         session.chooseName(displayName, Icon("desktop"))
-        session.joinGame(gameId)
+        session.joinGameAndWaitLobby(gameId)
         val firstTurn = session.awaitGameStart(gameId)
 
         session.watchTurns()
diff --git a/sw-client/src/commonMain/kotlin/org/luxons/sevenwonders/client/SevenWondersClient.kt b/sw-client/src/commonMain/kotlin/org/luxons/sevenwonders/client/SevenWondersClient.kt
index 9a9fe356..5dda0292 100644
--- a/sw-client/src/commonMain/kotlin/org/luxons/sevenwonders/client/SevenWondersClient.kt
+++ b/sw-client/src/commonMain/kotlin/org/luxons/sevenwonders/client/SevenWondersClient.kt
@@ -22,15 +22,7 @@ import org.luxons.sevenwonders.model.Settings
 import org.luxons.sevenwonders.model.api.ConnectedPlayer
 import org.luxons.sevenwonders.model.api.LobbyDTO
 import org.luxons.sevenwonders.model.api.SEVEN_WONDERS_WS_ENDPOINT
-import org.luxons.sevenwonders.model.api.actions.AddBotAction
-import org.luxons.sevenwonders.model.api.actions.ChooseNameAction
-import org.luxons.sevenwonders.model.api.actions.CreateGameAction
-import org.luxons.sevenwonders.model.api.actions.Icon
-import org.luxons.sevenwonders.model.api.actions.JoinGameAction
-import org.luxons.sevenwonders.model.api.actions.PrepareMoveAction
-import org.luxons.sevenwonders.model.api.actions.ReassignWondersAction
-import org.luxons.sevenwonders.model.api.actions.ReorderPlayersAction
-import org.luxons.sevenwonders.model.api.actions.UpdateSettingsAction
+import org.luxons.sevenwonders.model.api.actions.*
 import org.luxons.sevenwonders.model.api.errors.ErrorDTO
 import org.luxons.sevenwonders.model.cards.PreparedCard
 import org.luxons.sevenwonders.model.wonders.AssignedWonder
@@ -80,21 +72,16 @@ class SevenWondersSession(private val stompSession: StompSessionWithKxSerializat
     suspend fun watchGames(): Flow<List<LobbyDTO>> =
         stompSession.subscribe("/topic/games", ListSerializer(LobbyDTO.serializer()))
 
-    suspend fun createGame(gameName: String): LobbyDTO = stompSession.request(
-        sendDestination = "/app/lobby/create",
-        receiveDestination = "/user/queue/lobby/joined",
-        payload = CreateGameAction(gameName),
-        serializer = CreateGameAction.serializer(),
-        deserializer = LobbyDTO.serializer(),
-    )
+    suspend fun createGame(gameName: String) {
+        stompSession.convertAndSend("/app/lobby/create", CreateGameAction(gameName), CreateGameAction.serializer())
+    }
 
-    suspend fun joinGame(gameId: Long): LobbyDTO = stompSession.request(
-        sendDestination = "/app/lobby/join",
-        receiveDestination = "/user/queue/lobby/joined",
-        payload = JoinGameAction(gameId),
-        serializer = JoinGameAction.serializer(),
-        deserializer = LobbyDTO.serializer(),
-    )
+    suspend fun joinGame(gameId: Long) {
+        stompSession.convertAndSend("/app/lobby/join", JoinGameAction(gameId), JoinGameAction.serializer())
+    }
+
+    suspend fun watchLobbyJoined(): Flow<LobbyDTO> =
+        stompSession.subscribe("/user/queue/lobby/joined", LobbyDTO.serializer())
 
     suspend fun leaveLobby() {
         stompSession.sendEmptyMsg("/app/lobby/leave")
@@ -172,3 +159,9 @@ class SevenWondersSession(private val stompSession: StompSessionWithKxSerializat
         stompSession.sendEmptyMsg("/app/game/leave")
     }
 }
+
+suspend fun SevenWondersSession.joinGameAndWaitLobby(gameId: Long): LobbyDTO {
+    val joinedLobbies = watchLobbyJoined()
+    joinGame(gameId)
+    return joinedLobbies.first()
+}
\ No newline at end of file
diff --git a/sw-server/src/test/kotlin/org/luxons/sevenwonders/server/SevenWondersTest.kt b/sw-server/src/test/kotlin/org/luxons/sevenwonders/server/SevenWondersTest.kt
index 4d0b19f4..db50609a 100644
--- a/sw-server/src/test/kotlin/org/luxons/sevenwonders/server/SevenWondersTest.kt
+++ b/sw-server/src/test/kotlin/org/luxons/sevenwonders/server/SevenWondersTest.kt
@@ -1,6 +1,7 @@
 package org.luxons.sevenwonders.server
 
 import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.produceIn
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeout
@@ -8,6 +9,8 @@ import kotlinx.coroutines.withTimeoutOrNull
 import org.junit.runner.RunWith
 import org.luxons.sevenwonders.client.SevenWondersClient
 import org.luxons.sevenwonders.client.SevenWondersSession
+import org.luxons.sevenwonders.client.joinGameAndWaitLobby
+import org.luxons.sevenwonders.model.api.LobbyDTO
 import org.luxons.sevenwonders.server.test.runAsyncTest
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.boot.test.context.SpringBootTest.WebEnvironment
@@ -58,9 +61,11 @@ class SevenWondersTest {
         val session1 = newPlayer("Player1")
         val session2 = newPlayer("Player2")
         val gameName = "Test Game"
-        val lobby = ownerSession.createGame(gameName)
-        session1.joinGame(lobby.id)
-        session2.joinGame(lobby.id)
+
+        val lobby = ownerSession.createGameAndWaitLobby(gameName)
+
+        session1.joinGameAndWaitLobby(lobby.id)
+        session2.joinGameAndWaitLobby(lobby.id)
 
         val outsiderSession = newPlayer("Outsider")
         val started = launch { outsiderSession.awaitGameStart(lobby.id) }
@@ -77,7 +82,7 @@ class SevenWondersTest {
         val ownerSession = newPlayer("GameOwner")
 
         val gameName = "Test Game"
-        val lobby = ownerSession.createGame(gameName)
+        val lobby = ownerSession.createGameAndWaitLobby(gameName)
         assertNotNull(lobby)
         assertEquals(gameName, lobby.name)
 
@@ -95,7 +100,7 @@ class SevenWondersTest {
 
         val ownerSession = newPlayer("GameOwner")
         val gameName = "Test Game"
-        val createdLobby = ownerSession.createGame(gameName)
+        val createdLobby = ownerSession.createGameAndWaitLobby(gameName)
 
         receivedLobbies = withTimeout(500) { games.receive() }
         assertNotNull(receivedLobbies)
@@ -115,13 +120,13 @@ class SevenWondersTest {
         val session2 = newPlayer("Player2")
         println("startGame_3players: after player 2")
 
-        val lobby = session1.createGame("Test Game")
+        val lobby = session1.createGameAndWaitLobby("Test Game")
         println("startGame_3players: after player 1 creates game")
-        session2.joinGame(lobby.id)
+        session2.joinGameAndWaitLobby(lobby.id)
         println("startGame_3players: after player 2 joins game")
 
         val session3 = newPlayer("Player3")
-        session3.joinGame(lobby.id)
+        session3.joinGameAndWaitLobby(lobby.id)
 
         listOf(session1, session2, session3).forEachIndexed { i, session ->
             launch {
@@ -144,3 +149,9 @@ class SevenWondersTest {
         println("startGame_3players: end of test method (main body)")
     }
 }
+
+private suspend fun SevenWondersSession.createGameAndWaitLobby(gameName: String): LobbyDTO {
+    val joinedLobbies = watchLobbyJoined()
+    createGame(gameName)
+    return joinedLobbies.first()
+}
diff --git a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt
deleted file mode 100644
index b6f3662a..00000000
--- a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameBrowserSagas.kt
+++ /dev/null
@@ -1,52 +0,0 @@
-package org.luxons.sevenwonders.ui.redux.sagas
-
-import kotlinx.coroutines.cancelAndJoin
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.map
-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.RequestCreateGame
-import org.luxons.sevenwonders.ui.redux.RequestJoinGame
-import org.luxons.sevenwonders.ui.redux.UpdateGameListAction
-import org.luxons.sevenwonders.ui.router.Navigate
-import org.luxons.sevenwonders.ui.router.Route
-import org.luxons.sevenwonders.ui.utils.awaitFirst
-
-suspend fun SwSagaContext.gameBrowserSaga(session: SevenWondersSession) {
-    GameBrowserSaga(session, this).run()
-}
-
-private class GameBrowserSaga(
-    private val session: SevenWondersSession,
-    private val sagaContext: SwSagaContext,
-) {
-    suspend fun run() {
-        coroutineScope {
-            val gamesDispatch = launch { dispatchGameUpdates() }
-            val lobby = awaitCreateOrJoinGame()
-            gamesDispatch.cancelAndJoin()
-            sagaContext.dispatch(EnterLobbyAction(lobby))
-            sagaContext.dispatch(Navigate(Route.LOBBY))
-        }
-    }
-
-    private suspend fun dispatchGameUpdates() {
-        with(sagaContext) {
-            session.watchGames().map { UpdateGameListAction(it) }.dispatchAll()
-        }
-    }
-
-    private suspend fun awaitCreateOrJoinGame(): LobbyDTO = awaitFirst(this::awaitCreateGame, this::awaitJoinGame)
-
-    private suspend fun awaitCreateGame(): LobbyDTO {
-        val action = sagaContext.next<RequestCreateGame>()
-        return session.createGame(action.gameName)
-    }
-
-    private suspend fun awaitJoinGame(): LobbyDTO {
-        val action = sagaContext.next<RequestJoinGame>()
-        return session.joinGame(action.gameId)
-    }
-}
diff --git a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt
deleted file mode 100644
index fb9bdfe2..00000000
--- a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/GameSagas.kt
+++ /dev/null
@@ -1,35 +0,0 @@
-package org.luxons.sevenwonders.ui.redux.sagas
-
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.launch
-import org.luxons.sevenwonders.client.SevenWondersSession
-import org.luxons.sevenwonders.ui.redux.PlayerReadyEvent
-import org.luxons.sevenwonders.ui.redux.PreparedCardEvent
-import org.luxons.sevenwonders.ui.redux.PreparedMoveEvent
-import org.luxons.sevenwonders.ui.redux.RequestLeaveGame
-import org.luxons.sevenwonders.ui.redux.RequestPrepareMove
-import org.luxons.sevenwonders.ui.redux.RequestSayReady
-import org.luxons.sevenwonders.ui.redux.RequestUnprepareMove
-import org.luxons.sevenwonders.ui.redux.TurnInfoEvent
-import org.luxons.sevenwonders.ui.router.Navigate
-import org.luxons.sevenwonders.ui.router.Route
-
-suspend fun SwSagaContext.gameSaga(session: SevenWondersSession) {
-    val game = getState().gameState ?: error("Game saga run without a current game")
-    coroutineScope {
-        session.watchPlayerReady(game.id).map { PlayerReadyEvent(it) }.dispatchAllIn(this)
-        session.watchPreparedCards(game.id).map { PreparedCardEvent(it) }.dispatchAllIn(this)
-        session.watchOwnMoves().map { PreparedMoveEvent(it) }.dispatchAllIn(this)
-        session.watchTurns().map { TurnInfoEvent(it) }.dispatchAllIn(this)
-
-        launch { onEach<RequestSayReady> { session.sayReady() } }
-        launch { onEach<RequestPrepareMove> { session.prepareMove(it.move) } }
-        launch { onEach<RequestUnprepareMove> { session.unprepareMove() } }
-
-        next<RequestLeaveGame>()
-        session.leaveGame()
-        dispatch(Navigate(Route.GAME_BROWSER))
-    }
-    console.log("End of game saga")
-}
diff --git a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt
deleted file mode 100644
index 37872017..00000000
--- a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/LobbySagas.kt
+++ /dev/null
@@ -1,65 +0,0 @@
-package org.luxons.sevenwonders.ui.redux.sagas
-
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.launch
-import org.luxons.sevenwonders.client.SevenWondersSession
-import org.luxons.sevenwonders.ui.redux.EnterGameAction
-import org.luxons.sevenwonders.ui.redux.RequestAddBot
-import org.luxons.sevenwonders.ui.redux.RequestLeaveLobby
-import org.luxons.sevenwonders.ui.redux.RequestReassignWonders
-import org.luxons.sevenwonders.ui.redux.RequestReorderPlayers
-import org.luxons.sevenwonders.ui.redux.RequestStartGame
-import org.luxons.sevenwonders.ui.redux.UpdateLobbyAction
-import org.luxons.sevenwonders.ui.router.Navigate
-import org.luxons.sevenwonders.ui.router.Route
-import org.luxons.sevenwonders.ui.utils.awaitFirst
-
-suspend fun SwSagaContext.lobbySaga(session: SevenWondersSession) {
-    val lobby = getState().currentLobby ?: error("Lobby saga run without a current lobby")
-    coroutineScope {
-        val lobbyUpdatesSubscription = session.watchLobbyUpdates().map { UpdateLobbyAction(it) }.dispatchAllIn(this)
-
-        launch {
-            onEach<RequestAddBot> { session.addBot(it.botDisplayName) }
-        }
-        launch {
-            onEach<RequestReorderPlayers> { session.reorderPlayers(it.orderedPlayers) }
-        }
-        launch {
-            onEach<RequestReassignWonders> { session.reassignWonders(it.wonders) }
-        }
-        val startGameJob = launch { awaitStartGame(session) }
-
-        awaitFirst(
-            {
-                awaitLeaveLobby(session)
-                lobbyUpdatesSubscription.cancel()
-                startGameJob.cancel()
-                dispatch(Navigate(Route.GAME_BROWSER))
-            },
-            {
-                awaitGameStart(session, lobby.id)
-                lobbyUpdatesSubscription.cancel()
-                startGameJob.cancel()
-                dispatch(Navigate(Route.GAME))
-            },
-        )
-    }
-}
-
-private suspend fun SwSagaContext.awaitGameStart(session: SevenWondersSession, lobbyId: Long) {
-    val turnInfo = session.awaitGameStart(lobbyId)
-    val lobby = getState().currentLobby!!
-    dispatch(EnterGameAction(lobby, turnInfo))
-}
-
-private suspend fun SwSagaContext.awaitStartGame(session: SevenWondersSession) {
-    next<RequestStartGame>()
-    session.startGame()
-}
-
-private suspend fun SwSagaContext.awaitLeaveLobby(session: SevenWondersSession) {
-    next<RequestLeaveLobby>()
-    session.leaveLobby()
-}
diff --git a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/RouteBasedSagas.kt b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/RouteBasedSagas.kt
new file mode 100644
index 00000000..d4329f2f
--- /dev/null
+++ b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/RouteBasedSagas.kt
@@ -0,0 +1,41 @@
+package org.luxons.sevenwonders.ui.redux.sagas
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.map
+import org.luxons.sevenwonders.client.SevenWondersSession
+import org.luxons.sevenwonders.ui.redux.*
+import org.luxons.sevenwonders.ui.router.Navigate
+import org.luxons.sevenwonders.ui.router.Route
+
+suspend fun SwSagaContext.homeSaga(session: SevenWondersSession) {
+    val action = next<RequestChooseName>()
+    val player = session.chooseName(action.playerName)
+    dispatch(SetCurrentPlayerAction(player))
+    dispatch(Navigate(Route.GAME_BROWSER))
+}
+
+suspend fun SwSagaContext.gameBrowserSaga(session: SevenWondersSession) {
+    session.watchGames().map { UpdateGameListAction(it) }.dispatchAll()
+}
+
+suspend fun SwSagaContext.lobbySaga(session: SevenWondersSession) {
+    val lobby = getState().currentLobby ?: error("Lobby saga run without a current lobby")
+    coroutineScope {
+        session.watchLobbyUpdates().map { UpdateLobbyAction(it) }.dispatchAllIn(this)
+
+        val turnInfo = session.awaitGameStart(lobby.id)
+        dispatch(EnterGameAction(getState().currentLobby!!, turnInfo))
+        dispatch(Navigate(Route.GAME))
+    }
+}
+
+suspend fun SwSagaContext.gameSaga(session: SevenWondersSession) {
+    val game = getState().gameState ?: error("Game saga run without a current game")
+    coroutineScope {
+        session.watchPlayerReady(game.id).map { PlayerReadyEvent(it) }.dispatchAllIn(this)
+        session.watchPreparedCards(game.id).map { PreparedCardEvent(it) }.dispatchAllIn(this)
+        session.watchOwnMoves().map { PreparedMoveEvent(it) }.dispatchAllIn(this)
+        session.watchTurns().map { TurnInfoEvent(it) }.dispatchAllIn(this)
+    }
+    console.log("End of game saga")
+}
diff --git a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt
index 6339f051..c2d26e0f 100644
--- a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt
+++ b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/Sagas.kt
@@ -8,10 +8,8 @@ import org.hildan.krossbow.stomp.MissingHeartBeatException
 import org.hildan.krossbow.stomp.WebSocketClosedUnexpectedly
 import org.luxons.sevenwonders.client.SevenWondersClient
 import org.luxons.sevenwonders.client.SevenWondersSession
-import org.luxons.sevenwonders.ui.redux.FatalError
-import org.luxons.sevenwonders.ui.redux.RequestChooseName
-import org.luxons.sevenwonders.ui.redux.SetCurrentPlayerAction
-import org.luxons.sevenwonders.ui.redux.SwState
+import org.luxons.sevenwonders.ui.redux.*
+import org.luxons.sevenwonders.ui.router.Navigate
 import org.luxons.sevenwonders.ui.router.Route
 import org.luxons.sevenwonders.ui.router.routerSaga
 import redux.RAction
@@ -32,6 +30,9 @@ suspend fun SwSagaContext.rootSaga() = try {
             serverErrorSaga(session)
         }
 
+        launchApiActionHandlersIn(this, session)
+        launchNavigationHandlers(this, session)
+
         val player = session.chooseName(action.playerName, null)
         dispatch(SetCurrentPlayerAction(player))
 
@@ -75,8 +76,41 @@ private suspend fun serverErrorSaga(session: SevenWondersSession) {
     }
 }
 
-private suspend fun SwSagaContext.homeSaga(session: SevenWondersSession) {
-    val action = next<RequestChooseName>()
-    val player = session.chooseName(action.playerName)
-    dispatch(SetCurrentPlayerAction(player))
+private fun SwSagaContext.launchApiActionHandlersIn(scope: CoroutineScope, session: SevenWondersSession) {
+
+    scope.launchOnEach<RequestCreateGame> { session.createGame(it.gameName) }
+    scope.launchOnEach<RequestJoinGame> { session.joinGame(it.gameId) }
+
+    scope.launchOnEach<RequestAddBot> { session.addBot(it.botDisplayName) }
+    scope.launchOnEach<RequestReorderPlayers> { session.reorderPlayers(it.orderedPlayers) }
+    scope.launchOnEach<RequestReassignWonders> { session.reassignWonders(it.wonders) }
+    // mapAction<RequestUpdateSettings> { updateSettings(it.settings) }
+    scope.launchOnEach<RequestStartGame> { session.startGame() }
+
+    scope.launchOnEach<RequestSayReady> { session.sayReady() }
+    scope.launchOnEach<RequestPrepareMove> { session.prepareMove(it.move) }
+    scope.launchOnEach<RequestUnprepareMove> { session.unprepareMove() }
+}
+
+private fun SwSagaContext.launchNavigationHandlers(scope: CoroutineScope, session: SevenWondersSession) {
+
+    // FIXME map this actions like others and await server event instead
+    scope.launchOnEach<RequestLeaveLobby> {
+        session.leaveLobby()
+        dispatch(Navigate(Route.GAME_BROWSER))
+    }
+
+    // FIXME map this actions like others and await server event instead
+    scope.launchOnEach<RequestLeaveGame> {
+        session.leaveGame()
+        dispatch(Navigate(Route.GAME_BROWSER))
+    }
+
+    scope.launch {
+        session.watchLobbyJoined().collect { lobby ->
+            dispatch(EnterLobbyAction(lobby))
+            dispatch(Navigate(Route.LOBBY))
+        }
+    }
 }
+
diff --git a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt
index 3acf68e8..0a2d7fa5 100644
--- a/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt
+++ b/sw-ui/src/main/kotlin/org/luxons/sevenwonders/ui/redux/sagas/SagasFramework.kt
@@ -120,6 +120,14 @@ class SagaContext<S, A : RAction, R>(
     }
 
     /**
+     * Launches a coroutine in the receiver scope that executes [handle] on every action dispatched of the type [T].
+     * The returned [Job] can be used to cancel that coroutine (just like a regular [launch])
+     */
+    inline fun <reified T : A> CoroutineScope.launchOnEach(
+        crossinline handle: suspend SagaContext<S, A, R>.(T) -> Unit,
+    ): Job = launch { onEach(handle) }
+
+    /**
      * Suspends until the next action matching the given [predicate] is dispatched, and returns that action.
      */
     suspend fun next(predicate: (A) -> Boolean): A {