diff options
Diffstat (limited to 'sw-ui/src')
5 files changed, 40 insertions, 38 deletions
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 index 7806bc98..aa81ae55 100644 --- 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 @@ -1,8 +1,9 @@ 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.hildan.krossbow.stomp.StompSubscription import org.luxons.sevenwonders.client.SevenWondersSession import org.luxons.sevenwonders.model.api.LobbyDTO import org.luxons.sevenwonders.ui.redux.EnterLobbyAction @@ -23,17 +24,18 @@ private class GameBrowserSaga( ) { suspend fun run() { coroutineScope { - val gamesSubscription = session.watchGames() - launch { dispatchGameUpdates(gamesSubscription) } + val gamesDispatch = launch { dispatchGameUpdates() } val lobby = awaitCreateOrJoinGame() - gamesSubscription.unsubscribe() + gamesDispatch.cancelAndJoin() sagaContext.dispatch(EnterLobbyAction(lobby)) sagaContext.dispatch(Navigate(Route.LOBBY)) } } - private suspend fun dispatchGameUpdates(gamesSubscription: StompSubscription<List<LobbyDTO>>) { - sagaContext.dispatchAll(gamesSubscription.messages) { UpdateGameListAction(it) } + private suspend fun dispatchGameUpdates() { + with(sagaContext) { + session.watchGames().map { UpdateGameListAction(it) }.dispatchAll() + } } private suspend fun awaitCreateOrJoinGame(): LobbyDTO = awaitFirst(this::awaitCreateGame, this::awaitJoinGame) 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 index 38bb7cb7..48592778 100644 --- 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 @@ -1,6 +1,7 @@ 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 @@ -14,22 +15,16 @@ import org.luxons.sevenwonders.ui.redux.TurnInfoEvent suspend fun SwSagaContext.gameSaga(session: SevenWondersSession) { val game = getState().gameState ?: error("Game saga run without a current game") coroutineScope { - val playerReadySub = session.watchPlayerReady(game.id) - val preparedCardsSub = session.watchPreparedCards(game.id) - val turnInfoSub = session.watchTurns() - val sayReadyJob = launch { onEach<RequestSayReady> { session.sayReady() } } - val unprepareJob = launch { onEach<RequestUnprepareMove> { session.unprepareMove() } } - val prepareMoveJob = launch { - onEach<RequestPrepareMove> { - val move = session.prepareMove(it.move) - dispatch(PreparedMoveEvent(move)) - } - } - launch { dispatchAll(playerReadySub.messages) { PlayerReadyEvent(it) } } - launch { dispatchAll(preparedCardsSub.messages) { PreparedCardEvent(it) } } - launch { dispatchAll(turnInfoSub.messages) { TurnInfoEvent(it) } } - // TODO await game end - // TODO unsubscribe all subs, cancel all jobs + 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() } } + + // TODO await game end and cancel this scope to unsubscribe everything } 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 index c3996af7..e2bf82c5 100644 --- 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 @@ -1,10 +1,9 @@ package org.luxons.sevenwonders.ui.redux.sagas import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.map import kotlinx.coroutines.launch -import org.hildan.krossbow.stomp.StompSubscription import org.luxons.sevenwonders.client.SevenWondersSession -import org.luxons.sevenwonders.model.api.LobbyDTO import org.luxons.sevenwonders.ui.redux.EnterGameAction import org.luxons.sevenwonders.ui.redux.RequestLeaveLobby import org.luxons.sevenwonders.ui.redux.RequestStartGame @@ -17,19 +16,21 @@ suspend fun SwSagaContext.lobbySaga(session: SevenWondersSession) { val lobby = getState().currentLobby ?: error("Lobby saga run without a current lobby") coroutineScope { val lobbyUpdatesSubscription = session.watchLobbyUpdates() - launch { watchLobbyUpdates(lobbyUpdatesSubscription) } + .map { UpdateLobbyAction(it) } + .dispatchAllIn(this) + val startGameJob = launch { awaitStartGame(session) } awaitFirst( { awaitLeaveLobby(session) - lobbyUpdatesSubscription.unsubscribe() + lobbyUpdatesSubscription.cancel() startGameJob.cancel() dispatch(Navigate(Route.GAME_BROWSER)) }, { awaitGameStart(session, lobby.id) - lobbyUpdatesSubscription.unsubscribe() + lobbyUpdatesSubscription.cancel() startGameJob.cancel() dispatch(Navigate(Route.GAME)) } @@ -37,10 +38,6 @@ suspend fun SwSagaContext.lobbySaga(session: SevenWondersSession) { } } -private suspend fun SwSagaContext.watchLobbyUpdates(lobbyUpdatesSubscription: StompSubscription<LobbyDTO>) { - dispatchAll(lobbyUpdatesSubscription.messages) { UpdateLobbyAction(it) } -} - private suspend fun SwSagaContext.awaitGameStart(session: SevenWondersSession, lobbyId: Long) { val turnInfo = session.awaitGameStart(lobbyId) val lobby = getState().currentLobby!! 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 3c66d31e..565cf7ec 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 @@ -1,6 +1,9 @@ package org.luxons.sevenwonders.ui.redux.sagas +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.launch import org.luxons.sevenwonders.client.SevenWondersClient import org.luxons.sevenwonders.client.SevenWondersSession @@ -38,8 +41,7 @@ suspend fun SwSagaContext.rootSaga() = coroutineScope { } private suspend fun serverErrorSaga(session: SevenWondersSession) { - val errorsSub = session.watchErrors() - for (err in errorsSub.messages) { + session.watchErrors().collect { err -> // TODO use blueprintjs toaster console.error("${err.code}: ${err.message}") console.error(JSON.stringify(err)) 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 1a57708e..df00f43f 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 @@ -6,7 +6,8 @@ import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.GlobalScope import kotlinx.coroutines.Job import kotlinx.coroutines.channels.BroadcastChannel -import kotlinx.coroutines.channels.ReceiveChannel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.launch import redux.Middleware import redux.MiddlewareApi @@ -79,15 +80,20 @@ class SagaContext<S, A : RAction, R>( } /** - * Dispatches an action given by [createAction] for each message received in [channel]. + * Dispatches all actions from this flow. */ - suspend fun <T> dispatchAll(channel: ReceiveChannel<T>, createAction: (T) -> A) { - for (msg in channel) { - reduxApi.dispatch(createAction(msg)) + suspend fun Flow<A>.dispatchAll() { + collect { + reduxApi.dispatch(it) } } /** + * Dispatches all actions from this flow in the provided [scope]. + */ + fun Flow<A>.dispatchAllIn(scope: CoroutineScope): Job = scope.launch { dispatchAll() } + + /** * Executes [handle] on every action dispatched. This runs forever until the current coroutine is cancelled. */ suspend fun onEach(handle: suspend SagaContext<S, A, R>.(A) -> Unit) { |