flecomte/event-demo
1
1
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Improve concurrency

Browse Source 
Fix GameState.currentPlayerTurn and nextPlayer
Add ItsTheTurnOfNotification
Improve test
This commit is contained in:
Fabrice Lecomte
2025-03-10 22:18:06 +01:00
parent 3a685496fd
commit 6028846828
23 changed files with 344 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/main/kotlin/eventDemo/app/command/GameCommandHandler.kt
View File

@@ -15,7 +15,6 @@ import io.github.oshai.kotlinlogging.KotlinLogging
import io.ktor.websocket.Frame
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.trySendBlocking
/**
* Listen [GameCommand] on [GameCommandStream], check the validity and execute an action.
@@ -33,39 +32,29 @@ class GameCommandHandler(
*/
suspend fun handle(
player: Player,
incoming: ReceiveChannel<Frame>,
outgoing: SendChannel<Frame>,
) {
val commandStream = GameCommandStream(incoming)
val playerErrorNotifier: (String) -> Unit = {
incomingCommandChannel: ReceiveChannel<Frame>,
outgoingErrorChannelNotification: SendChannel<Frame>,
) = GameCommandStream(incomingCommandChannel).process { command ->
if (command.payload.player.id != player.id) {
nack()
}
val playerErrorNotifier: suspend (String) -> Unit = {
val notification = ErrorNotification(message = it)
logger.atInfo {
logger.atWarn {
message = "Notification send ERROR: ${notification.message}"
payload = mapOf("notification" to notification)
}
outgoing.trySendBlocking(notification.toFrame())
outgoingErrorChannelNotification.send(notification.toFrame())
}
return init(player, commandStream, playerErrorNotifier)
}
private suspend fun init(
player: Player,
commandStream: GameCommandStream,
playerErrorNotifier: (String) -> Unit,
) {
commandStream.process { command ->
if (command.payload.player.id != player.id) {
nack()
}
val gameState = gameStateRepository.get(command.payload.gameId)
val gameState = gameStateRepository.get(command.payload.gameId)
when (command) {
is IWantToPlayCardCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
is IamReadyToPlayCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
is IWantToJoinTheGameCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
is ICantPlayCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
}
when (command) {
is IWantToPlayCardCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
is IamReadyToPlayCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
is IWantToJoinTheGameCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
is ICantPlayCommand -> command.run(gameState, playerErrorNotifier, eventHandler)
}
}
}

8
src/main/kotlin/eventDemo/app/command/command/ICantPlayCommand.kt
View File

@@ -23,11 +23,15 @@ data class ICantPlayCommand(
override val player: Player,
) : GameCommand.Payload
fun run(
suspend fun run(
state: GameState,
playerErrorNotifier: (String) -> Unit,
playerErrorNotifier: suspend (String) -> Unit,
eventHandler: GameEventHandler,
) {
if (state.currentPlayerTurn != payload.player) {
playerErrorNotifier("Its not your turn!")
return
}
val playableCards = state.playableCards(payload.player)
if (playableCards.isEmpty()) {
val takenCard = state.deck.stack.first()

4
src/main/kotlin/eventDemo/app/command/command/IWantToJoinTheGameCommand.kt
View File

@@ -24,9 +24,9 @@ data class IWantToJoinTheGameCommand(
override val player: Player,
) : GameCommand.Payload
fun run(
suspend fun run(
state: GameState,
playerErrorNotifier: (String) -> Unit,
playerErrorNotifier: suspend (String) -> Unit,
eventHandler: GameEventHandler,
) {
val logger = KotlinLogging.logger {}

8
src/main/kotlin/eventDemo/app/command/command/IWantToPlayCardCommand.kt
View File

@@ -25,15 +25,19 @@ data class IWantToPlayCardCommand(
val card: Card,
) : GameCommand.Payload
fun run(
suspend fun run(
state: GameState,
playerErrorNotifier: (String) -> Unit,
playerErrorNotifier: suspend (String) -> Unit,
eventHandler: GameEventHandler,
) {
if (!state.isStarted) {
playerErrorNotifier("The game is Not started")
return
}
if (state.currentPlayerTurn != payload.player) {
playerErrorNotifier("Its not your turn!")
return
}
if (state.canBePlayThisCard(payload.player, payload.card)) {
eventHandler.handle(

4
src/main/kotlin/eventDemo/app/command/command/IamReadyToPlayCommand.kt
View File

@@ -23,9 +23,9 @@ data class IamReadyToPlayCommand(
override val player: Player,
) : GameCommand.Payload
fun run(
suspend fun run(
state: GameState,
playerErrorNotifier: (String) -> Unit,
playerErrorNotifier: suspend (String) -> Unit,
eventHandler: GameEventHandler,
) {
val playerExist: Boolean = state.players.contains(payload.player)

5
src/main/kotlin/eventDemo/app/event/event/CardIsPlayedEvent.kt
View File

@@ -11,6 +11,7 @@ import java.util.UUID
data class CardIsPlayedEvent(
override val gameId: GameId,
val card: Card,
val player: Player,
override val player: Player,
override val eventId: UUID = UUID.randomUUID(),
) : GameEvent
) : GameEvent,
PlayerActionEvent

7
src/main/kotlin/eventDemo/app/event/event/PlayerActionEvent.kt Normal file
View File

@@ -0,0 +1,7 @@
package eventDemo.app.event.event
import eventDemo.app.entity.Player
sealed interface PlayerActionEvent {
val player: Player
}

5
src/main/kotlin/eventDemo/app/event/event/PlayerChoseColorEvent.kt
View File

@@ -10,8 +10,9 @@ import java.util.UUID
*/
data class PlayerChoseColorEvent(
override val gameId: GameId,
val player: Player,
override val player: Player,
val color: Card.Color,
) : GameEvent {
) : GameEvent,
PlayerActionEvent {
override val eventId: UUID = UUID.randomUUID()
}

5
src/main/kotlin/eventDemo/app/event/event/PlayerHavePassEvent.kt
View File

@@ -10,8 +10,9 @@ import java.util.UUID
*/
data class PlayerHavePassEvent(
override val gameId: GameId,
val player: Player,
override val player: Player,
val takenCard: Card,
) : GameEvent {
) : GameEvent,
PlayerActionEvent {
override val eventId: UUID = UUID.randomUUID()
}

68
src/main/kotlin/eventDemo/app/event/projection/GameState.kt
View File

@@ -10,9 +10,9 @@ import kotlinx.serialization.Serializable
data class GameState(
val gameId: GameId,
val players: Set<Player> = emptySet(),
val lastPlayer: Player? = null,
val lastCard: LastCard? = null,
val lastColor: Card.Color? = null,
val currentPlayerTurn: Player? = null,
val cardOnCurrentStack: LastCard? = null,
val colorOnCurrentStack: Card.Color? = null,
val direction: Direction = Direction.CLOCKWISE,
val readyPlayers: Set<Player> = emptySet(),
val deck: Deck = Deck(players),
@@ -42,8 +42,8 @@ data class GameState(
return players.size == readyPlayers.size && players.all { readyPlayers.contains(it) }
}
private val lastPlayerIndex: Int? get() {
val i = players.indexOf(lastPlayer)
private val currentPlayerIndex: Int? get() {
val i = players.indexOf(currentPlayerTurn)
return if (i == -1) {
null
} else {
@@ -51,28 +51,42 @@ data class GameState(
}
}
private val nextPlayerIndex: Int get() {
if (players.size == 0) return 0
private fun nextPlayerIndex(direction: Direction): Int {
if (players.isEmpty()) return 0
val y =
if (direction == Direction.CLOCKWISE) {
+1
} else {
-1
}
return ((lastPlayerIndex ?: 0) + y) % players.size
}
val nextPlayer: Player? by lazy {
if (players.isEmpty()) {
null
return if (direction == Direction.CLOCKWISE) {
sidePlayerIndexClockwise
} else {
players.elementAt(nextPlayerIndex)
sidePlayerIndexCounterClockwise
}
}
val Player.currentIndex: Int get() = players.indexOf(this)
fun nextPlayer(direction: Direction): Player = players.elementAt(nextPlayerIndex(direction))
private val sidePlayerIndexClockwise: Int by lazy {
if (players.isEmpty()) {
0
} else {
((currentPlayerIndex ?: 0) + 1) % players.size
}
}
private val sidePlayerIndexCounterClockwise: Int by lazy {
if (players.isEmpty()) {
0
} else {
((currentPlayerIndex ?: 0) - 1) % players.size
}
}
val nextPlayerTurn: Player? by lazy {
if (players.isEmpty()) {
null
} else {
nextPlayer(direction)
}
}
private val Player.currentIndex: Int get() = players.indexOf(this)
fun Player.playerDiffIndex(nextPlayer: Player): Int =
if (direction == Direction.CLOCKWISE) {
@@ -82,8 +96,8 @@ data class GameState(
}.let { it % players.size }
val Player.cardOnBoardIsForYou: Boolean get() {
if (lastCard == null) error("No card")
return this.playerDiffIndex(lastCard.player) == 1
if (cardOnCurrentStack == null) error("No card")
return this.playerDiffIndex(cardOnCurrentStack.player) == 1
}
fun playableCards(player: Player): List<Card> =
@@ -102,7 +116,7 @@ data class GameState(
player: Player,
card: Card,
): Boolean {
val cardOnBoard = lastCard?.card ?: return false
val cardOnBoard = cardOnCurrentStack?.card ?: return false
return when (cardOnBoard) {
is Card.NumericCard -> {
when (card) {
@@ -134,7 +148,7 @@ data class GameState(
is Card.ChangeColorCard -> {
when (card) {
is Card.AllColorCard -> true
is Card.ColorCard -> card.color == lastColor
is Card.ColorCard -> card.color == colorOnCurrentStack
}
}
@@ -156,7 +170,7 @@ data class GameState(
} else {
when (card) {
is Card.AllColorCard -> true
is Card.ColorCard -> card.color == lastColor
is Card.ColorCard -> card.color == colorOnCurrentStack
}
}
}

44
src/main/kotlin/eventDemo/app/event/projection/GameStateBuilder.kt
View File

@@ -7,18 +7,22 @@ import eventDemo.app.event.event.CardIsPlayedEvent
import eventDemo.app.event.event.GameEvent
import eventDemo.app.event.event.GameStartedEvent
import eventDemo.app.event.event.NewPlayerEvent
import eventDemo.app.event.event.PlayerActionEvent
import eventDemo.app.event.event.PlayerChoseColorEvent
import eventDemo.app.event.event.PlayerHavePassEvent
import eventDemo.app.event.event.PlayerReadyEvent
import eventDemo.app.event.event.PlayerWinEvent
import io.github.oshai.kotlinlogging.KotlinLogging
fun GameId.buildStateFromEventStream(eventStream: GameEventStream): GameState {
val events = eventStream.readAll(this)
if (events.isEmpty()) return GameState(this)
return events.buildStateFromEvents()
return events.buildStateFromEvents().also {
KotlinLogging.logger {}.warn { "state is build from scratch for game: $this " }
}
}
fun List<GameEvent>.buildStateFromEvents(): GameState {
fun Collection<GameEvent>.buildStateFromEvents(): GameState {
val gameId = this.firstOrNull()?.gameId ?: error("Cannot build GameState from an empty list")
return fold(GameState(gameId)) { state, event ->
state.apply(event)
@@ -27,9 +31,14 @@ fun List<GameEvent>.buildStateFromEvents(): GameState {
fun GameState.apply(event: GameEvent): GameState =
let { state ->
if (event is PlayerActionEvent) {
if (state.currentPlayerTurn != event.player) {
error("inconsistent player turn. currentPlayerTurn: $currentPlayerTurn | player: ${event.player}")
}
}
when (event) {
is CardIsPlayedEvent -> {
val direction =
val nextDirectionAfterPlay =
when (event.card) {
is Card.ReverseCard -> state.direction.revert()
else -> state.direction
@@ -38,14 +47,21 @@ fun GameState.apply(event: GameEvent): GameState =
val color =
when (event.card) {
is Card.ColorCard -> event.card.color
else -> state.lastColor
is Card.AllColorCard -> null
}
val currentPlayerAfterThePlay =
if (event.card is Card.AllColorCard) {
currentPlayerTurn
} else {
nextPlayer(nextDirectionAfterPlay)
}
state.copy(
lastPlayer = event.player,
direction = direction,
lastColor = color,
lastCard = GameState.LastCard(event.card, event.player),
currentPlayerTurn = currentPlayerAfterThePlay,
direction = nextDirectionAfterPlay,
colorOnCurrentStack = color,
cardOnCurrentStack = GameState.LastCard(event.card, event.player),
deck = state.deck.putOneCardFromHand(event.player, event.card),
)
}
@@ -59,6 +75,7 @@ fun GameState.apply(event: GameEvent): GameState =
}
is PlayerReadyEvent -> {
if (state.isStarted) error("The game is already started")
state.copy(
readyPlayers = state.readyPlayers + event.player,
)
@@ -67,22 +84,23 @@ fun GameState.apply(event: GameEvent): GameState =
is PlayerHavePassEvent -> {
if (event.takenCard != state.deck.stack.first()) error("taken card is not ot top of the stack")
state.copy(
lastPlayer = event.player,
currentPlayerTurn = nextPlayerTurn,
deck = state.deck.takeOneCardFromStackTo(event.player),
)
}
is PlayerChoseColorEvent -> {
state.copy(
lastColor = event.color,
currentPlayerTurn = nextPlayerTurn,
colorOnCurrentStack = event.color,
)
}
is GameStartedEvent -> {
state.copy(
lastColor = (event.deck.discard.first() as? Card.ColorCard)?.color ?: state.lastColor,
lastCard = GameState.LastCard(event.deck.discard.first(), event.firstPlayer),
lastPlayer = event.firstPlayer,
colorOnCurrentStack = (event.deck.discard.first() as? Card.ColorCard)?.color ?: state.colorOnCurrentStack,
cardOnCurrentStack = GameState.LastCard(event.deck.discard.first(), event.firstPlayer),
currentPlayerTurn = event.firstPlayer,
deck = event.deck,
isStarted = true,
)

3
src/main/kotlin/eventDemo/app/event/projection/GameStateRepository.kt
View File

@@ -22,8 +22,7 @@ class GameStateRepository(
val projection = projections[event.gameId]
if (projection == null) {
event
.gameId
.buildStateFromEventStream(eventStream)
.buildStateFromEventStreamTo(eventStream)
.update()
} else {
projection

171
src/main/kotlin/eventDemo/app/eventListener/GameEventPlayerNotificationListener.kt
View File

@@ -1,5 +1,6 @@
package eventDemo.app.eventListener
import eventDemo.app.entity.Card
import eventDemo.app.entity.Player
import eventDemo.app.event.GameEventBus
import eventDemo.app.event.event.CardIsPlayedEvent
@@ -11,6 +12,8 @@ import eventDemo.app.event.event.PlayerHavePassEvent
import eventDemo.app.event.event.PlayerReadyEvent
import eventDemo.app.event.event.PlayerWinEvent
import eventDemo.app.event.projection.GameStateRepository
import eventDemo.app.notification.ItsTheTurnOfNotification
import eventDemo.app.notification.Notification
import eventDemo.app.notification.PlayerAsJoinTheGameNotification
import eventDemo.app.notification.PlayerAsPlayACardNotification
import eventDemo.app.notification.PlayerHavePassNotification
@@ -33,100 +36,110 @@ class GameEventPlayerNotificationListener(
private val logger = KotlinLogging.logger {}
fun startListening(
outgoing: SendChannel<Frame>,
outgoingNotificationChannel: SendChannel<Frame>,
currentPlayer: Player,
) {
eventBus.subscribe { event: GameEvent ->
val currentState = gameStateRepository.getUntil(event)
val notification =
when (event) {
is NewPlayerEvent -> {
if (currentPlayer != event.player) {
PlayerAsJoinTheGameNotification(
player = event.player,
)
} else {
WelcomeToTheGameNotification(
players = currentState.players,
)
}
}
is CardIsPlayedEvent -> {
if (currentPlayer != event.player) {
PlayerAsPlayACardNotification(
player = event.player,
card = event.card,
)
} else {
null
}
fun Notification.send() {
if (currentState.players.contains(currentPlayer)) {
// Only notify players who have already joined the game.
outgoingNotificationChannel.trySendBlocking(toFrame())
logger.atInfo {
message = "Notification for player ${currentPlayer.name} was SEND: ${this@send}"
payload = mapOf("notification" to this@send, "event" to event)
}
is GameStartedEvent -> {
TheGameWasStartedNotification(
hand =
event.deck.playersHands.getHand(currentPlayer)
?: error("You are not in the game"),
)
} else {
// Rare use case, when a connexion is created with the channel,
// but the player was not already join in the game
logger.atWarn {
message = "Notification for player ${currentPlayer.name} was SKIP, No player on the game: ${this@send}"
payload = mapOf("notification" to this@send, "event" to event)
}
}
}
is PlayerChoseColorEvent -> {
if (currentPlayer != event.player) {
PlayerWasChoseTheCardColorNotification(
player = event.player,
color = event.color,
)
} else {
null
}
}
fun sendNextTurnNotif() =
ItsTheTurnOfNotification(
player = currentState.currentPlayerTurn ?: error("No player turn defined"),
).send()
is PlayerHavePassEvent -> {
if (currentPlayer == event.player) {
YourNewCardNotification(
card = event.takenCard,
)
} else {
PlayerHavePassNotification(
player = event.player,
)
}
}
is PlayerReadyEvent -> {
if (currentPlayer != event.player) {
PlayerWasReadyNotification(
player = event.player,
)
} else {
null
}
}
is PlayerWinEvent -> {
PlayerWinNotification(
when (event) {
is NewPlayerEvent -> {
if (currentPlayer != event.player) {
PlayerAsJoinTheGameNotification(
player = event.player,
)
).send()
} else {
WelcomeToTheGameNotification(
players = currentState.players,
).send()
}
}
if (notification == null) {
logger.atInfo {
message = "Notification Ignore: $event"
payload = mapOf("event" to event)
is CardIsPlayedEvent -> {
if (currentPlayer != event.player) {
PlayerAsPlayACardNotification(
player = event.player,
card = event.card,
).send()
}
if (event.card !is Card.AllColorCard) {
ItsTheTurnOfNotification(
player = currentState.currentPlayerTurn ?: error("No player turn defined"),
).send()
}
}
} else if (currentState.players.contains(currentPlayer)) {
// Only notify players who have already joined the game.
outgoing.trySendBlocking(notification.toFrame())
logger.atInfo {
message = "Notification SEND: $notification"
payload = mapOf("notification" to notification, "event" to event)
is GameStartedEvent -> {
TheGameWasStartedNotification(
hand =
event.deck.playersHands.getHand(currentPlayer)
?: error("You are not in the game"),
).send()
sendNextTurnNotif()
}
} else {
logger.atInfo {
message = "Notification SKIP: $notification"
payload = mapOf("notification" to notification, "event" to event)
is PlayerChoseColorEvent -> {
if (currentPlayer != event.player) {
PlayerWasChoseTheCardColorNotification(
player = event.player,
color = event.color,
).send()
}
sendNextTurnNotif()
}
is PlayerHavePassEvent -> {
if (currentPlayer == event.player) {
YourNewCardNotification(
card = event.takenCard,
).send()
} else {
PlayerHavePassNotification(
player = event.player,
).send()
}
sendNextTurnNotif()
}
is PlayerReadyEvent -> {
if (currentPlayer != event.player) {
PlayerWasReadyNotification(
player = event.player,
).send()
}
}
is PlayerWinEvent -> {
PlayerWinNotification(
player = event.player,
).send()
}
}
}

11
src/main/kotlin/eventDemo/app/eventListener/GameEventReactionListener.kt
View File

@@ -4,6 +4,7 @@ import eventDemo.app.event.GameEventBus
import eventDemo.app.event.GameEventHandler
import eventDemo.app.event.event.GameEvent
import eventDemo.app.event.event.GameStartedEvent
import eventDemo.app.event.event.PlayerReadyEvent
import eventDemo.app.event.event.PlayerWinEvent
import eventDemo.app.event.projection.GameState
import eventDemo.app.event.projection.GameStateRepository
@@ -29,7 +30,7 @@ class GameEventReactionListener(
}
}
private fun sendStartGameEvent(
private suspend fun sendStartGameEvent(
state: GameState,
event: GameEvent,
) {
@@ -40,7 +41,7 @@ class GameEventReactionListener(
state.players,
)
logger.atInfo {
message = "Event Send on reaction of: $event"
message = "Reaction event was Send $reactionEvent on reaction of: $event"
payload =
mapOf(
"event" to event,
@@ -48,6 +49,10 @@ class GameEventReactionListener(
)
}
eventHandler.handle(reactionEvent)
} else {
if (event is PlayerReadyEvent) {
logger.info { "All players was not ready ${state.readyPlayers}" }
}
}
}
@@ -63,7 +68,7 @@ class GameEventReactionListener(
winner,
)
logger.atInfo {
message = "Event Send on reaction of: $event"
message = "Reaction event was Send $reactionEvent on reaction of: $event"
payload =
mapOf(
"event" to event,

13
src/main/kotlin/eventDemo/app/notification/ItsTheTurnOfNotification.kt Normal file
View File

@@ -0,0 +1,13 @@
package eventDemo.app.notification
import eventDemo.app.entity.Player
import eventDemo.shared.UUIDSerializer
import kotlinx.serialization.Serializable
import java.util.UUID
@Serializable
data class ItsTheTurnOfNotification(
@Serializable(with = UUIDSerializer::class)
override val id: UUID = UUID.randomUUID(),
val player: Player,
) : Notification

2
src/main/kotlin/eventDemo/app/query/ReadTheGameState.kt
View File

@@ -40,7 +40,7 @@ fun Route.readTheGameState(gameStateRepository: GameStateRepository) {
get<Game.Card> { body ->
gameStateRepository
.get(body.game.id)
.lastCard
.cardOnCurrentStack
?.card
?.let { call.respond(it) }
?: call.response.status(HttpStatusCode.BadRequest)

2
src/main/kotlin/eventDemo/libs/command/CommandStreamChannel.kt
View File

@@ -49,7 +49,7 @@ class CommandStreamChannel<C : Command>(
val actionResult = runCatching { status.action(command) }
if (actionResult.isFailure) {
logger.atInfo {
message = "Error on compute the Command"
message = "Error on compute the Command: $command"
payload = mapOf("command" to command)
cause = actionResult.exceptionOrNull()
}

2
src/main/kotlin/eventDemo/libs/event/EventBus.kt
View File

@@ -8,6 +8,6 @@ interface EventBus<E : Event<ID>, ID : AggregateId> {
*/
fun subscribe(
priority: Int = 0,
block: (E) -> Unit,
block: suspend (E) -> Unit,
)
}

12
src/main/kotlin/eventDemo/libs/event/EventBusInMemory.kt
View File

@@ -1,19 +1,23 @@
package eventDemo.libs.event
import kotlinx.coroutines.runBlocking
class EventBusInMemory<E : Event<ID>, ID : AggregateId> : EventBus<E, ID> {
private val subscribers: MutableList<Pair<Int, (E) -> Unit>> = mutableListOf()
private val subscribers: MutableList<Pair<Int, suspend (E) -> Unit>> = mutableListOf()
override fun publish(event: E) {
subscribers
.sortedByDescending { (priority, block) -> priority }
.sortedByDescending { (priority, _) -> priority }
.forEach { (_, block) ->
block(event)
runBlocking {
block(event)
}
}
}
override fun subscribe(
priority: Int,
block: (E) -> Unit,
block: suspend (E) -> Unit,
) {
subscribers.add(priority to block)
}

2
src/main/kotlin/eventDemo/libs/event/EventStream.kt
View File

@@ -22,5 +22,5 @@ interface EventStream<E : Event<ID>, ID : AggregateId> {
): R?
/** Reads all events associated with a given aggregate ID */
fun readAll(aggregateId: ID): List<E>
fun readAll(aggregateId: ID): Set<E>
}

6
src/main/kotlin/eventDemo/libs/event/EventStreamInMemory.kt
View File

@@ -1,6 +1,8 @@
package eventDemo.libs.event
import io.github.oshai.kotlinlogging.KotlinLogging
import java.util.Queue
import java.util.concurrent.ConcurrentLinkedQueue
import kotlin.reflect.KClass
/**
@@ -10,7 +12,7 @@ import kotlin.reflect.KClass
*/
class EventStreamInMemory<E : Event<ID>, ID : AggregateId> : EventStream<E, ID> {
private val logger = KotlinLogging.logger {}
private val events: MutableList<E> = mutableListOf()
private val events: Queue<E> = ConcurrentLinkedQueue()
override fun publish(event: E) {
events.add(event)
@@ -34,7 +36,7 @@ class EventStreamInMemory<E : Event<ID>, ID : AggregateId> : EventStream<E, ID>
.filterIsInstance(eventType.java)
.lastOrNull { it.gameId == aggregateId }
override fun readAll(aggregateId: ID): List<E> = events
override fun readAll(aggregateId: ID): Set<E> = events.toSet()
}
inline fun <reified R : E, E : Event<ID>, ID : AggregateId> EventStream<E, ID>.readLastOf(aggregateId: ID): R? =