Add GameStateRepository.getUntil(event)

Add Cache and Snapshot to the GameStateRepository
Add eventId on events
This commit is contained in:
2025-03-09 04:44:14 +01:00
parent 19e425d684
commit 3a685496fd
14 changed files with 110 additions and 43 deletions

View File

@@ -3,6 +3,7 @@ package eventDemo.app.event.event
import eventDemo.app.entity.Card import eventDemo.app.entity.Card
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import java.util.UUID
/** /**
* An [GameEvent] to represent a played card. * An [GameEvent] to represent a played card.
@@ -11,4 +12,5 @@ data class CardIsPlayedEvent(
override val gameId: GameId, override val gameId: GameId,
val card: Card, val card: Card,
val player: Player, val player: Player,
override val eventId: UUID = UUID.randomUUID(),
) : GameEvent ) : GameEvent

View File

@@ -3,11 +3,13 @@ package eventDemo.app.event.event
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.libs.event.Event import eventDemo.libs.event.Event
import kotlinx.serialization.Serializable import kotlinx.serialization.Serializable
import java.util.UUID
/** /**
* An [Event] of a Game. * An [Event] of a Game.
*/ */
@Serializable @Serializable
sealed interface GameEvent : Event<GameId> { sealed interface GameEvent : Event<GameId> {
override val eventId: UUID
override val gameId: GameId override val gameId: GameId
} }

View File

@@ -4,6 +4,7 @@ import eventDemo.app.entity.Deck
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import eventDemo.app.entity.initHands import eventDemo.app.entity.initHands
import java.util.UUID
/** /**
* This [GameEvent] is sent when all players are ready. * This [GameEvent] is sent when all players are ready.
@@ -13,6 +14,8 @@ data class GameStartedEvent(
val firstPlayer: Player, val firstPlayer: Player,
val deck: Deck, val deck: Deck,
) : GameEvent { ) : GameEvent {
override val eventId: UUID = UUID.randomUUID()
companion object { companion object {
fun new( fun new(
id: GameId, id: GameId,

View File

@@ -2,6 +2,7 @@ package eventDemo.app.event.event
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import java.util.UUID
/** /**
* An [GameEvent] to represent a new player joining the game. * An [GameEvent] to represent a new player joining the game.
@@ -9,4 +10,6 @@ import eventDemo.app.entity.Player
data class NewPlayerEvent( data class NewPlayerEvent(
override val gameId: GameId, override val gameId: GameId,
val player: Player, val player: Player,
) : GameEvent ) : GameEvent {
override val eventId: UUID = UUID.randomUUID()
}

View File

@@ -3,6 +3,7 @@ package eventDemo.app.event.event
import eventDemo.app.entity.Card import eventDemo.app.entity.Card
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import java.util.UUID
/** /**
* This [GameEvent] is sent when a player chose a color. * This [GameEvent] is sent when a player chose a color.
@@ -11,4 +12,6 @@ data class PlayerChoseColorEvent(
override val gameId: GameId, override val gameId: GameId,
val player: Player, val player: Player,
val color: Card.Color, val color: Card.Color,
) : GameEvent ) : GameEvent {
override val eventId: UUID = UUID.randomUUID()
}

View File

@@ -3,6 +3,7 @@ package eventDemo.app.event.event
import eventDemo.app.entity.Card import eventDemo.app.entity.Card
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import java.util.UUID
/** /**
* This [GameEvent] is sent when a player can play. * This [GameEvent] is sent when a player can play.
@@ -11,4 +12,6 @@ data class PlayerHavePassEvent(
override val gameId: GameId, override val gameId: GameId,
val player: Player, val player: Player,
val takenCard: Card, val takenCard: Card,
) : GameEvent ) : GameEvent {
override val eventId: UUID = UUID.randomUUID()
}

View File

@@ -2,6 +2,7 @@ package eventDemo.app.event.event
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import java.util.UUID
/** /**
* This [GameEvent] is sent when a player is ready. * This [GameEvent] is sent when a player is ready.
@@ -9,4 +10,6 @@ import eventDemo.app.entity.Player
data class PlayerReadyEvent( data class PlayerReadyEvent(
override val gameId: GameId, override val gameId: GameId,
val player: Player, val player: Player,
) : GameEvent ) : GameEvent {
override val eventId: UUID = UUID.randomUUID()
}

View File

@@ -2,6 +2,7 @@ package eventDemo.app.event.event
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import java.util.UUID
/** /**
* This [GameEvent] is sent when a player is ready. * This [GameEvent] is sent when a player is ready.
@@ -9,4 +10,6 @@ import eventDemo.app.entity.Player
data class PlayerWinEvent( data class PlayerWinEvent(
override val gameId: GameId, override val gameId: GameId,
val player: Player, val player: Player,
) : GameEvent ) : GameEvent {
override val eventId: UUID = UUID.randomUUID()
}

View File

@@ -12,28 +12,18 @@ import eventDemo.app.event.event.PlayerHavePassEvent
import eventDemo.app.event.event.PlayerReadyEvent import eventDemo.app.event.event.PlayerReadyEvent
import eventDemo.app.event.event.PlayerWinEvent import eventDemo.app.event.event.PlayerWinEvent
fun GameId.buildStateFromEventStream(eventStream: GameEventStream): GameState = fun GameId.buildStateFromEventStream(eventStream: GameEventStream): GameState {
buildStateFromEvents( val events = eventStream.readAll(this)
eventStream.readAll(this), if (events.isEmpty()) return GameState(this)
) return events.buildStateFromEvents()
}
/** fun List<GameEvent>.buildStateFromEvents(): GameState {
* Build the state to the specific event val gameId = this.firstOrNull()?.gameId ?: error("Cannot build GameState from an empty list")
*/ return fold(GameState(gameId)) { state, event ->
fun GameEvent.buildStateFromEventStreamTo(eventStream: GameEventStream): GameState =
gameId.buildStateFromEvents(
eventStream.readAll(gameId).takeWhile { it != this } + this,
)
private fun GameId.buildStateFromEvents(events: List<GameEvent>): GameState =
events.fold(GameState(this)) { state, event ->
state.apply(event)
}
fun List<GameEvent>.buildStateFromEvents(): GameState =
fold(GameState(this.first().gameId)) { state, event ->
state.apply(event) state.apply(event)
} }
}
fun GameState.apply(event: GameEvent): GameState = fun GameState.apply(event: GameEvent): GameState =
let { state -> let { state ->

View File

@@ -3,32 +3,92 @@ package eventDemo.app.event.projection
import eventDemo.app.entity.GameId import eventDemo.app.entity.GameId
import eventDemo.app.event.GameEventHandler import eventDemo.app.event.GameEventHandler
import eventDemo.app.event.GameEventStream import eventDemo.app.event.GameEventStream
import eventDemo.app.event.event.GameEvent
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger
class GameStateRepository( class GameStateRepository(
private val eventStream: GameEventStream, private val eventStream: GameEventStream,
eventHandler: GameEventHandler, eventHandler: GameEventHandler,
private val maxSnapshotCacheSize: Int = 20,
) { ) {
private val projections: ConcurrentHashMap<GameId, GameState> = ConcurrentHashMap() private val projections: MutableMap<GameId, GameState> = ConcurrentHashMap()
private val version: AtomicInteger = AtomicInteger(0)
private val projectionsSnapshot: MutableMap<GameEvent, GameState> = ConcurrentHashMap()
private val sortedSnapshotByVersion: MutableMap<GameEvent, Int> = ConcurrentHashMap()
init { init {
eventHandler.registerProjectionBuilder { event -> eventHandler.registerProjectionBuilder { event ->
val projection = projections[event.gameId] val projection = projections[event.gameId]
if (projection == null) { if (projection == null) {
event.gameId event
.gameId
.buildStateFromEventStream(eventStream) .buildStateFromEventStream(eventStream)
.update() .update()
} else { } else {
projection projection
.apply(event) .apply(event)
.let { projections.put(it.gameId, it) } .also { projections[it.gameId] = it }
.also { state ->
val newVersion = version.addAndGet(1)
saveSnapshot(event, state, newVersion)
removeOldSnapshot()
}
} }
} }
} }
fun get(gameId: GameId): GameState = gameId.buildStateFromEventStream(eventStream) private fun removeOldSnapshot() {
if (projectionsSnapshot.size > maxSnapshotCacheSize) {
val numberToRemove = projectionsSnapshot.size - maxSnapshotCacheSize
sortedSnapshotByVersion
.toList()
.sortedBy { it.second }
.take(numberToRemove)
.toMap()
.keys
.forEach { event ->
sortedSnapshotByVersion.remove(event)
projectionsSnapshot.remove(event)
}
}
}
private fun saveSnapshot(
event: GameEvent,
state: GameState,
newVersion: Int,
) {
projectionsSnapshot[event] = state
sortedSnapshotByVersion[event] = newVersion
}
/**
* Get the last version of the [GameState] from the all eventStream.
*
* It fetches it from the local cache if possible, otherwise it builds it.
*/
fun get(gameId: GameId): GameState =
projections[gameId]
?: gameId.buildStateFromEventStream(eventStream)
/**
* Get the [GameState] to the specific [event][GameEvent].
* It does not contain the [events][GameEvent] it after this one.
*
* It fetches it from the local cache if possible, otherwise it builds it.
*/
fun getUntil(event: GameEvent): GameState =
projectionsSnapshot[event]
?: event.buildStateFromEventStreamTo(eventStream)
private fun GameState.update() { private fun GameState.update() {
projections[gameId] = this projections[gameId] = this
} }
/**
* Build the state to the specific event
*/
private fun GameEvent.buildStateFromEventStreamTo(eventStream: GameEventStream): GameState =
run { eventStream.readAll(gameId).takeWhile { it != this } + this }.buildStateFromEvents()
} }

View File

@@ -2,7 +2,6 @@ package eventDemo.app.eventListener
import eventDemo.app.entity.Player import eventDemo.app.entity.Player
import eventDemo.app.event.GameEventBus import eventDemo.app.event.GameEventBus
import eventDemo.app.event.GameEventStream
import eventDemo.app.event.event.CardIsPlayedEvent import eventDemo.app.event.event.CardIsPlayedEvent
import eventDemo.app.event.event.GameEvent import eventDemo.app.event.event.GameEvent
import eventDemo.app.event.event.GameStartedEvent import eventDemo.app.event.event.GameStartedEvent
@@ -11,7 +10,7 @@ import eventDemo.app.event.event.PlayerChoseColorEvent
import eventDemo.app.event.event.PlayerHavePassEvent import eventDemo.app.event.event.PlayerHavePassEvent
import eventDemo.app.event.event.PlayerReadyEvent import eventDemo.app.event.event.PlayerReadyEvent
import eventDemo.app.event.event.PlayerWinEvent import eventDemo.app.event.event.PlayerWinEvent
import eventDemo.app.event.projection.buildStateFromEventStreamTo import eventDemo.app.event.projection.GameStateRepository
import eventDemo.app.notification.PlayerAsJoinTheGameNotification import eventDemo.app.notification.PlayerAsJoinTheGameNotification
import eventDemo.app.notification.PlayerAsPlayACardNotification import eventDemo.app.notification.PlayerAsPlayACardNotification
import eventDemo.app.notification.PlayerHavePassNotification import eventDemo.app.notification.PlayerHavePassNotification
@@ -29,7 +28,7 @@ import kotlinx.coroutines.channels.trySendBlocking
class GameEventPlayerNotificationListener( class GameEventPlayerNotificationListener(
private val eventBus: GameEventBus, private val eventBus: GameEventBus,
private val eventStream: GameEventStream, private val gameStateRepository: GameStateRepository,
) { ) {
private val logger = KotlinLogging.logger {} private val logger = KotlinLogging.logger {}
@@ -38,7 +37,7 @@ class GameEventPlayerNotificationListener(
currentPlayer: Player, currentPlayer: Player,
) { ) {
eventBus.subscribe { event: GameEvent -> eventBus.subscribe { event: GameEvent ->
val currentState = event.buildStateFromEventStreamTo(eventStream) val currentState = gameStateRepository.getUntil(event)
val notification = val notification =
when (event) { when (event) {
is NewPlayerEvent -> { is NewPlayerEvent -> {

View File

@@ -2,18 +2,17 @@ package eventDemo.app.eventListener
import eventDemo.app.event.GameEventBus import eventDemo.app.event.GameEventBus
import eventDemo.app.event.GameEventHandler import eventDemo.app.event.GameEventHandler
import eventDemo.app.event.GameEventStream
import eventDemo.app.event.event.GameEvent import eventDemo.app.event.event.GameEvent
import eventDemo.app.event.event.GameStartedEvent import eventDemo.app.event.event.GameStartedEvent
import eventDemo.app.event.event.PlayerWinEvent import eventDemo.app.event.event.PlayerWinEvent
import eventDemo.app.event.projection.GameState import eventDemo.app.event.projection.GameState
import eventDemo.app.event.projection.buildStateFromEventStreamTo import eventDemo.app.event.projection.GameStateRepository
import io.github.oshai.kotlinlogging.KotlinLogging import io.github.oshai.kotlinlogging.KotlinLogging
class GameEventReactionListener( class GameEventReactionListener(
private val eventBus: GameEventBus, private val eventBus: GameEventBus,
private val eventHandler: GameEventHandler, private val eventHandler: GameEventHandler,
private val eventStream: GameEventStream, private val gameStateRepository: GameStateRepository,
private val priority: Int = DEFAULT_PRIORITY, private val priority: Int = DEFAULT_PRIORITY,
) { ) {
companion object Config { companion object Config {
@@ -24,7 +23,7 @@ class GameEventReactionListener(
fun init() { fun init() {
eventBus.subscribe(priority) { event: GameEvent -> eventBus.subscribe(priority) { event: GameEvent ->
val state = event.buildStateFromEventStreamTo(eventStream) val state = gameStateRepository.getUntil(event)
sendStartGameEvent(state, event) sendStartGameEvent(state, event)
sendWinnerEvent(state, event) sendWinnerEvent(state, event)
} }
@@ -71,7 +70,7 @@ class GameEventReactionListener(
"reactionEvent" to reactionEvent, "reactionEvent" to reactionEvent,
) )
} }
eventStream.publish(reactionEvent) eventHandler.handle(reactionEvent)
} }
} }
} }

View File

@@ -33,12 +33,8 @@ val appKoinModule =
single { single {
GameStateRepository(get(), get()) GameStateRepository(get(), get())
} }
single {
GameEventHandler(get(), get())
}
single {
GameCommandHandler(get(), get())
}
singleOf(::GameEventHandler)
singleOf(::GameCommandHandler)
singleOf(::GameEventPlayerNotificationListener) singleOf(::GameEventPlayerNotificationListener)
} }

View File

@@ -15,5 +15,6 @@ interface AggregateId {
* @see EventStream * @see EventStream
*/ */
interface Event<ID : AggregateId> { interface Event<ID : AggregateId> {
val eventId: UUID
val gameId: ID val gameId: ID
} }