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

feat: remove snapshot on ProjectionRepository
Some checks failed
Tests / build (push) Failing after 6m58s
Details
Tests / lint (push) Has been skipped
Details
Tests / test (push) Has been skipped
Details

Browse Source
This commit is contained in:
Fabrice Lecomte
2025-04-17 01:24:25 +02:00
parent b6e8a2f347
commit cba9971ca1
37 changed files with 478 additions and 871 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/main/kotlin/eventDemo/adapter/infrastructureLayer/event/projection/GameListRepositoryInMemory.kt
View File

@@ -2,28 +2,20 @@ package eventDemo.adapter.infrastructureLayer.event.projection
import eventDemo.business.entity.GameId
import eventDemo.business.event.GameEventBus
import eventDemo.business.event.GameEventStore
import eventDemo.business.event.projection.GameList
import eventDemo.business.event.projection.GameListRepository
import eventDemo.business.event.projection.GameProjectionBus
import eventDemo.business.event.projection.GameState
import eventDemo.business.event.projection.apply
import eventDemo.libs.event.projection.ProjectionSnapshotRepositoryInMemory
import eventDemo.libs.event.projection.SnapshotConfig
import eventDemo.libs.event.projection.ProjectionRepositoryInMemory
import io.github.oshai.kotlinlogging.withLoggingContext
/**
* Manages [projections][GameList], their building and publication in the [bus][GameProjectionBus].
*/
class GameListRepositoryInMemory(
eventStore: GameEventStore,
snapshotConfig: SnapshotConfig = SnapshotConfig(),
) : GameListRepository {
private val projectionsSnapshot =
ProjectionSnapshotRepositoryInMemory(
name = GameListRepositoryInMemory::class,
eventStore = eventStore,
snapshotCacheConfig = snapshotConfig,
class GameListRepositoryInMemory : GameListRepository {
private val projectionsRepository =
ProjectionRepositoryInMemory(
applyToProjection = GameList::apply,
initialStateBuilder = { aggregateId: GameId -> GameList(aggregateId) },
)
@@ -32,11 +24,11 @@ class GameListRepositoryInMemory(
projectionBus: GameProjectionBus,
eventBus: GameEventBus,
) {
// On new event was received, build snapshot and publish it to the projection bus
// On new event was received, build projection and publish it to the projection bus
eventBus.subscribe { event ->
withLoggingContext("event" to event.toString()) {
projectionsSnapshot
.applyAndPutToCache(event)
projectionsRepository
.applyAndSave(event)
.also { projectionBus.publish(it) }
}
}
@@ -48,5 +40,5 @@ class GameListRepositoryInMemory(
* It fetches it from the local cache if possible, otherwise it builds it.
*/
override fun getList(): List<GameList> =
projectionsSnapshot.getList()
projectionsRepository.getList()
}

18
src/main/kotlin/eventDemo/adapter/infrastructureLayer/event/projection/GameListRepositoryInRedis.kt
View File

@@ -2,14 +2,12 @@ package eventDemo.adapter.infrastructureLayer.event.projection
import eventDemo.business.entity.GameId
import eventDemo.business.event.GameEventBus
import eventDemo.business.event.GameEventStore
import eventDemo.business.event.projection.GameList
import eventDemo.business.event.projection.GameListRepository
import eventDemo.business.event.projection.GameProjectionBus
import eventDemo.business.event.projection.GameState
import eventDemo.business.event.projection.apply
import eventDemo.libs.event.projection.ProjectionSnapshotRepositoryInRedis
import eventDemo.libs.event.projection.SnapshotConfig
import eventDemo.libs.event.projection.ProjectionRepositoryInRedis
import io.github.oshai.kotlinlogging.withLoggingContext
import kotlinx.serialization.json.Json
import redis.clients.jedis.UnifiedJedis
@@ -18,14 +16,10 @@ import redis.clients.jedis.UnifiedJedis
* Manages [projections][GameList], their building and publication in the [bus][GameProjectionBus].
*/
class GameListRepositoryInRedis(
eventStore: GameEventStore,
jedis: UnifiedJedis,
snapshotConfig: SnapshotConfig = SnapshotConfig(),
) : GameListRepository {
private val projectionsSnapshot =
ProjectionSnapshotRepositoryInRedis(
eventStore = eventStore,
snapshotCacheConfig = snapshotConfig,
private val projectionsRepository =
ProjectionRepositoryInRedis(
initialStateBuilder = { aggregateId: GameId -> GameList(aggregateId) },
projectionClass = GameList::class,
projectionToJson = { Json.encodeToString(GameList.serializer(), it) },
@@ -40,8 +34,8 @@ class GameListRepositoryInRedis(
) {
eventBus.subscribe { event ->
withLoggingContext("event" to event.toString()) {
projectionsSnapshot
.applyAndPutToCache(event)
projectionsRepository
.applyAndSave(event)
.also { projectionBus.publish(it) }
}
}
@@ -53,5 +47,5 @@ class GameListRepositoryInRedis(
* It fetches it from the local cache if possible, otherwise it builds it.
*/
override fun getList(): List<GameList> =
projectionsSnapshot.getList()
projectionsRepository.getList()
}

43
src/main/kotlin/eventDemo/adapter/infrastructureLayer/event/projection/GameStateRepositoryInMemory.kt
View File

@@ -2,28 +2,19 @@ package eventDemo.adapter.infrastructureLayer.event.projection
import eventDemo.business.entity.GameId
import eventDemo.business.event.GameEventBus
import eventDemo.business.event.GameEventStore
import eventDemo.business.event.event.GameEvent
import eventDemo.business.event.projection.GameProjectionBus
import eventDemo.business.event.projection.GameState
import eventDemo.business.event.projection.GameStateRepository
import eventDemo.business.event.projection.apply
import eventDemo.libs.event.projection.ProjectionSnapshotRepositoryInMemory
import eventDemo.libs.event.projection.SnapshotConfig
import eventDemo.libs.event.projection.ProjectionRepositoryInMemory
import io.github.oshai.kotlinlogging.withLoggingContext
/**
* Manages [projections][GameState], their building and publication in the [bus][GameProjectionBus].
*/
class GameStateRepositoryInMemory(
eventStore: GameEventStore,
snapshotConfig: SnapshotConfig = SnapshotConfig(),
) : GameStateRepository {
private val projectionsSnapshot =
ProjectionSnapshotRepositoryInMemory(
name = GameStateRepositoryInMemory::class,
eventStore = eventStore,
snapshotCacheConfig = snapshotConfig,
class GameStateRepositoryInMemory : GameStateRepository {
private val projectionsRepository =
ProjectionRepositoryInMemory(
applyToProjection = GameState::apply,
initialStateBuilder = { aggregateId: GameId -> GameState(aggregateId) },
)
@@ -32,33 +23,19 @@ class GameStateRepositoryInMemory(
projectionBus: GameProjectionBus,
eventBus: GameEventBus,
) {
// On new event was received, build snapshot and publish it to the projection bus
// On new event was received, build projection and publish it to the projection bus
eventBus.subscribe { event ->
withLoggingContext("event" to event.toString()) {
projectionsSnapshot
.applyAndPutToCache(event)
projectionsRepository
.applyAndSave(event)
.also { projectionBus.publish(it) }
}
}
}
/**
* Get the last version of the [GameState] from the all eventStream.
*
* It fetches it from the local cache if possible, otherwise it builds it.
* Get the [GameState].
*/
override fun getLast(gameId: GameId): GameState =
projectionsSnapshot.getLast(gameId)
/**
* 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.
*/
override fun getUntil(event: GameEvent): GameState =
projectionsSnapshot.getUntil(event)
override fun count(gameId: GameId): Int =
projectionsSnapshot.count(gameId)
override fun get(gameId: GameId): GameState =
projectionsRepository.get(gameId)
}

39
src/main/kotlin/eventDemo/adapter/infrastructureLayer/event/projection/GameStateRepositoryInRedis.kt
View File

@@ -2,14 +2,11 @@ package eventDemo.adapter.infrastructureLayer.event.projection
import eventDemo.business.entity.GameId
import eventDemo.business.event.GameEventBus
import eventDemo.business.event.GameEventStore
import eventDemo.business.event.event.GameEvent
import eventDemo.business.event.projection.GameProjectionBus
import eventDemo.business.event.projection.GameState
import eventDemo.business.event.projection.GameStateRepository
import eventDemo.business.event.projection.apply
import eventDemo.libs.event.projection.ProjectionSnapshotRepositoryInRedis
import eventDemo.libs.event.projection.SnapshotConfig
import eventDemo.libs.event.projection.ProjectionRepositoryInRedis
import io.github.oshai.kotlinlogging.withLoggingContext
import kotlinx.serialization.json.Json
import redis.clients.jedis.UnifiedJedis
@@ -18,14 +15,10 @@ import redis.clients.jedis.UnifiedJedis
* Manages [projections][GameState], their building and publication in the [bus][GameProjectionBus].
*/
class GameStateRepositoryInRedis(
eventStore: GameEventStore,
jedis: UnifiedJedis,
snapshotConfig: SnapshotConfig = SnapshotConfig(),
) : GameStateRepository {
private val projectionsSnapshot =
ProjectionSnapshotRepositoryInRedis(
eventStore = eventStore,
snapshotCacheConfig = snapshotConfig,
private val projectionsRepository =
ProjectionRepositoryInRedis(
initialStateBuilder = { aggregateId: GameId -> GameState(aggregateId) },
projectionClass = GameState::class,
projectionToJson = { Json.encodeToString(GameState.serializer(), it) },
@@ -38,33 +31,19 @@ class GameStateRepositoryInRedis(
projectionBus: GameProjectionBus,
eventBus: GameEventBus,
) {
// On new event was received, build snapshot and publish it to the projection bus
// On new event was received, build projection and publish it to the projection bus
eventBus.subscribe { event ->
withLoggingContext("event" to event.toString()) {
projectionsSnapshot
.applyAndPutToCache(event)
projectionsRepository
.applyAndSave(event)
.also { projectionBus.publish(it) }
}
}
}
/**
* Get the last version of the [GameState] from the all eventStream.
*
* It fetches it from the local cache if possible, otherwise it builds it.
* Get the [GameState].
*/
override fun getLast(gameId: GameId): GameState =
projectionsSnapshot.getLast(gameId)
/**
* 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.
*/
override fun getUntil(event: GameEvent): GameState =
projectionsSnapshot.getUntil(event)
override fun count(gameId: GameId): Int =
projectionsSnapshot.count(gameId)
override fun get(gameId: GameId): GameState =
projectionsRepository.get(gameId)
}

4
src/main/kotlin/eventDemo/adapter/interfaceLayer/query/ReadTheGameState.kt
View File

@@ -38,7 +38,7 @@ fun Route.readTheGameState(gameStateRepository: GameStateRepository) {
// Read the last played card on the game.
get<Game.Card> { body ->
gameStateRepository
.getLast(body.game.id)
.get(body.game.id)
.cardOnCurrentStack
?.let { call.respond(it) }
?: call.response.status(HttpStatusCode.BadRequest)
@@ -46,7 +46,7 @@ fun Route.readTheGameState(gameStateRepository: GameStateRepository) {
// Read the last played card on the game.
get<Game.State> { body ->
val state = gameStateRepository.getLast(body.game.id)
val state = gameStateRepository.get(body.game.id)
call.respond(state)
}
}

14
src/main/kotlin/eventDemo/business/command/GameCommandHandler.kt
View File

@@ -86,12 +86,12 @@ class GameCommandHandler(
* If the command fail, send an [error notification][CommandErrorNotification],
* if success, send a [success notification][CommandSuccessNotification]
*/
suspend fun handle(
fun handle(
player: Player,
gameId: GameId,
command: GameCommand,
sendSuccess: suspend () -> Unit,
sendError: suspend (message: String) -> Unit,
sendSuccess: () -> Unit,
sendError: (message: String) -> Unit,
) {
if (command.payload.aggregateId.id != gameId.id) {
logger.warn { "Handle command Refuse, the gameId of the command is not the same" }
@@ -114,26 +114,26 @@ class GameCommandHandler(
}
}
private fun SendChannel<Notification>.sendSuccess(command: GameCommand): suspend () -> Unit =
private fun SendChannel<Notification>.sendSuccess(command: GameCommand): () -> Unit =
{
val logger = KotlinLogging.logger { }
CommandSuccessNotification(commandId = command.id)
.also { notification ->
withLoggingContext("notification" to notification.toString(), "commandId" to command.id.toString()) {
logger.debug { "Notification SUCCESS sent" }
send(notification)
trySend(notification)
}
}
}
private fun SendChannel<Notification>.sendError(command: GameCommand): suspend (message: String) -> Unit =
private fun SendChannel<Notification>.sendError(command: GameCommand): (message: String) -> Unit =
{
val logger = KotlinLogging.logger { }
CommandErrorNotification(message = it, command = command)
.also { notification ->
withLoggingContext("notification" to notification.toString(), "command" to command.toString()) {
logger.warn { "Notification ERROR sent: ${notification.message}" }
send(notification)
trySend(notification)
}
}
}

2
src/main/kotlin/eventDemo/business/command/action/ICantPlay.kt
View File

@@ -12,7 +12,7 @@ data class ICantPlay(
private val gameStateRepository: GameStateRepository,
) : CommandAction<ICantPlayCommand, PlayerHavePassEvent> {
override fun run(command: ICantPlayCommand): (version: Int) -> PlayerHavePassEvent {
val state = gameStateRepository.getLast(command.payload.aggregateId)
val state = gameStateRepository.get(command.payload.aggregateId)
if (state.currentPlayerTurn != command.payload.player) {
throw CommandException("Its not your turn!")

2
src/main/kotlin/eventDemo/business/command/action/IWantToJoinTheGame.kt
View File

@@ -12,7 +12,7 @@ data class IWantToJoinTheGame(
private val gameStateRepository: GameStateRepository,
) : CommandAction<IWantToJoinTheGameCommand, NewPlayerEvent> {
override fun run(command: IWantToJoinTheGameCommand): (version: Int) -> NewPlayerEvent {
val state = gameStateRepository.getLast(command.payload.aggregateId)
val state = gameStateRepository.get(command.payload.aggregateId)
if (!state.isStarted) {
return {
NewPlayerEvent(

2
src/main/kotlin/eventDemo/business/command/action/IWantToPlayCard.kt
View File

@@ -12,7 +12,7 @@ data class IWantToPlayCard(
private val gameStateRepository: GameStateRepository,
) : CommandAction<IWantToPlayCardCommand, CardIsPlayedEvent> {
override fun run(command: IWantToPlayCardCommand): (version: Int) -> CardIsPlayedEvent {
val state = gameStateRepository.getLast(command.payload.aggregateId)
val state = gameStateRepository.get(command.payload.aggregateId)
if (!state.isStarted) {
throw CommandException("The game is Not started")

2
src/main/kotlin/eventDemo/business/command/action/IamReadyToPlay.kt
View File

@@ -13,7 +13,7 @@ class IamReadyToPlay(
) : CommandAction<IamReadyToPlayCommand, PlayerReadyEvent> {
@Throws(CommandException::class)
override fun run(command: IamReadyToPlayCommand): (version: Int) -> PlayerReadyEvent {
val state = gameStateRepository.getLast(command.payload.aggregateId)
val state = gameStateRepository.get(command.payload.aggregateId)
val playerExist: Boolean = state.players.contains(command.payload.player)
val playerIsAlreadyReady: Boolean = state.readyPlayers.contains(command.payload.player)

7
src/main/kotlin/eventDemo/business/event/projection/gameState/GameStateRepository.kt
View File

@@ -1,12 +1,7 @@
package eventDemo.business.event.projection
import eventDemo.business.entity.GameId
import eventDemo.business.event.event.GameEvent
interface GameStateRepository {
fun getLast(gameId: GameId): GameState
fun getUntil(event: GameEvent): GameState
fun count(gameId: GameId): Int
fun get(gameId: GameId): GameState
}

4
src/main/kotlin/eventDemo/business/event/projection/projectionListener/ReactionListener.kt
View File

@@ -37,7 +37,7 @@ class ReactionListener(
}
}
private suspend fun sendStartGameEvent(state: GameState) {
private fun sendStartGameEvent(state: GameState) {
if (state.isReady && !state.isStarted) {
val reactionEvent =
eventHandler.handle(state.aggregateId) {
@@ -54,7 +54,7 @@ class ReactionListener(
}
}
private suspend fun sendWinnerEvent(state: GameState) {
private fun sendWinnerEvent(state: GameState) {
val winner = state.playerHasNoCardLeft().firstOrNull()
if (winner != null) {
val reactionEvent =

5
src/main/kotlin/eventDemo/configuration/injection/ConfigureDIInfrastructure.kt
View File

@@ -13,7 +13,6 @@ import eventDemo.business.event.GameEventStore
import eventDemo.business.event.projection.GameListRepository
import eventDemo.business.event.projection.GameProjectionBus
import eventDemo.business.event.projection.GameStateRepository
import eventDemo.libs.event.projection.SnapshotConfig
import org.koin.core.module.Module
import org.koin.core.module.dsl.singleOf
import org.koin.core.scope.Scope
@@ -65,10 +64,10 @@ fun Module.configureDIInfrastructure(config: Configuration) {
singleOf(::GameProjectionBusInRabbitMQ) bind GameProjectionBus::class
single {
GameStateRepositoryInRedis(get(), get(), snapshotConfig = SnapshotConfig())
GameStateRepositoryInRedis(get())
} bind GameStateRepository::class
single {
GameListRepositoryInRedis(get(), get(), snapshotConfig = SnapshotConfig())
GameListRepositoryInRedis(get())
} bind GameListRepository::class
}

15
src/main/kotlin/eventDemo/libs/bus/Bus.kt
View File

@@ -1,13 +1,22 @@
package eventDemo.libs.bus
interface Bus<T> {
suspend fun publish(item: T)
/**
* Publish a new [message][item] to the bus.
*/
fun publish(item: T)
/**
* @param priority The higher the priority, the more it will be called first
* Subscribe a [lambda][block] to the bus.
*
* When a message is sent to the bus, the [block] is executed.
*/
fun subscribe(block: suspend (T) -> Unit): Subscription
fun subscribe(block: (T) -> Unit): Subscription
/**
* The returns of the [subscribe] method.
* It can be called to [cancel][close] the subscription.
*/
interface Subscription : AutoCloseable {
override fun close()
}

8
src/main/kotlin/eventDemo/libs/bus/BusInMemory.kt
View File

@@ -9,21 +9,19 @@ class BusInMemory<E>(
val name: KClass<*> = BusInMemory::class,
) : Bus<E> {
private val logger = KotlinLogging.logger(name.qualifiedName.toString())
private val subscribers: MutableList<suspend (E) -> Unit> = mutableListOf()
private val subscribers: MutableList<(E) -> Unit> = mutableListOf()
override suspend fun publish(item: E) {
override fun publish(item: E) {
withLoggingContext("busItem" to item.toString()) {
logger.info { "Item sent to the bus" }
subscribers
.forEach {
coroutineScope {
it(item)
}
}
}
}
override fun subscribe(block: suspend (E) -> Unit): Bus.Subscription {
override fun subscribe(block: (E) -> Unit): Bus.Subscription {
subscribers.add(block)
return object : Bus.Subscription {
override fun close() {

8
src/main/kotlin/eventDemo/libs/bus/BusInRabbitMQ.kt
View File

@@ -42,21 +42,19 @@ class BusInRabbitMQ<E>(
}
}
override suspend fun publish(item: E) {
override fun publish(item: E) {
connection
.createChannel()
.use {
it.basicPublish(
.basicPublish(
exchangeName,
routingKey,
AMQP.BasicProperties(),
objectToString(item).toByteArray(),
)
}
logger.info { "Item sent to the bus" }
}
override fun subscribe(block: suspend (E) -> Unit): Bus.Subscription {
override fun subscribe(block: (E) -> Unit): Bus.Subscription {
connection
.createChannel()
.also { channel ->

6
src/main/kotlin/eventDemo/libs/command/CommandHandler.kt
View File

@@ -44,7 +44,7 @@ class CommandHandler<B : Bus<E>, E : Event<ID>, ID : AggregateId, C : Command>(
*
* It restricts to run only once the [command].
*/
suspend fun handle(
fun handle(
aggregateId: ID,
command: C,
callback: CommandCallback<C>,
@@ -99,10 +99,10 @@ private class EventCommandMap<C : Command, E : Event<*>>(
val event: E,
val date: Instant,
) {
suspend operator fun invoke(error: CommandException? = null) {
operator fun invoke(error: CommandException? = null) {
callback(command, error)
}
}
}
typealias CommandCallback<C> = suspend (command: C, error: CommandException?) -> Unit
typealias CommandCallback<C> = (command: C, error: CommandException?) -> Unit

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

@@ -14,7 +14,7 @@ class CommandRunnerController<C : Command>(
) {
private val executedCommand: ConcurrentHashMap<CommandId, Pair<Boolean, Instant>> = ConcurrentHashMap()
suspend fun runOnlyOnce(
fun runOnlyOnce(
command: C,
action: CommandBlock<C>,
) {

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

@@ -34,7 +34,7 @@ class CommandStreamChannel<C : Command>(
}
}
private suspend fun runAndLogStatus(
private fun runAndLogStatus(
command: C,
action: CommandBlock<C>,
) {
@@ -47,4 +47,4 @@ class CommandStreamChannel<C : Command>(
}
}
typealias CommandBlock<C> = suspend (C) -> Unit
typealias CommandBlock<C> = (C) -> Unit

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

@@ -4,7 +4,7 @@ package eventDemo.libs.event
* A stream to publish and read the played card event.
*/
interface EventHandler<E : Event<ID>, ID : AggregateId> {
suspend fun handle(
fun handle(
aggregateId: ID,
buildEvent: (version: Int) -> E,
): E

10
src/main/kotlin/eventDemo/libs/event/EventHandlerImpl.kt
View File

@@ -19,7 +19,7 @@ class EventHandlerImpl<E : Event<ID>, ID : AggregateId>(
/**
* Build Event then send it to the event store and bus.
*/
override suspend fun handle(
override fun handle(
aggregateId: ID,
buildEvent: (version: Int) -> E,
): E =
@@ -34,12 +34,8 @@ class EventHandlerImpl<E : Event<ID>, ID : AggregateId>(
.also {
withLoggingContext("event" to it.toString()) {
eventStore.publish(it)
}
}
}.also { event ->
withLoggingContext("event" to event.toString()) {
// Publish to the bus
eventBus.publish(event)
eventBus.publish(it)
}
}
}
}

34
src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepository.kt Normal file
View File

@@ -0,0 +1,34 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
interface ProjectionRepository<E : Event<ID>, P : Projection<ID>, ID : AggregateId> {
/**
* Update projection with the event.
*/
fun apply(event: E): P
/**
* Update projection with the event, and save it.
*/
fun applyAndSave(event: E): P
/**
* Save the projection.
*/
fun save(projection: P)
/**
* Build the list of all [Projections][Projection]
*/
fun getList(
limit: Int = 100,
offset: Int = 0,
): List<P>
/**
* Build the last version of the [Projection] from the cache.
*/
fun get(aggregateId: ID): P
}

64
src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryAbs.kt Normal file
View File

@@ -0,0 +1,64 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
import io.github.oshai.kotlinlogging.KotlinLogging
import io.github.oshai.kotlinlogging.withLoggingContext
import io.ktor.util.collections.ConcurrentMap
/**
* Repository abstraction to declare common process
*/
abstract class ProjectionRepositoryAbs<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
private val applyToProjection: P.(event: E) -> P,
) : ProjectionRepository<E, P, ID> {
private val logger = KotlinLogging.logger {}
/**
* Update projection with the event.
*
* 1. get the last projection
* 2. apply the new event to the projection
*/
override fun apply(event: E): P =
get(event.aggregateId).applyToProjectionSecure(event)
/**
* Update projection with the event, and save it.
*
* 1. get the last projection
* 2. apply the new event to projection
* 3. save it
*/
override fun applyAndSave(event: E): P =
apply(event)
.also {
withLoggingContext("projection" to it.toString(), "event" to event.toString()) {
save(it)
}
}
/**
* Wrap the [applyToProjection] lambda to avoid duplicate apply of the same event.
*/
protected val applyToProjectionSecure: P.(event: E) -> P = { event ->
withLoggingContext("event" to event.toString(), "projection" to this.toString()) {
if (canBeApply(event)) {
applyToProjection(event)
} else if (event.version <= lastEventVersion) {
"Event is already in the Projection, skip apply.".let {
logger.warn { it }
error(it)
}
} else {
"The version of the event must follow directly after the version of the projection.".let {
logger.error { it }
error(it)
}
}
}
}
private fun P.canBeApply(event: E): Boolean =
event.version == lastEventVersion + 1
}

46
src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInMemory.kt Normal file
View File

@@ -0,0 +1,46 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
import java.util.concurrent.ConcurrentHashMap
class ProjectionRepositoryInMemory<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
private val initialStateBuilder: (aggregateId: ID) -> P,
applyToProjection: P.(event: E) -> P,
) : ProjectionRepositoryAbs<E, P, ID>(applyToProjection),
ProjectionRepository<E, P, ID> {
private val projections: ConcurrentHashMap<ID, P> = ConcurrentHashMap()
/**
* Build the list of all [Projections][Projection]
*/
override fun getList(
limit: Int,
offset: Int,
): List<P> =
projections
.values
.drop(offset)
.take(limit)
/**
* Get the [Projection].
*/
override fun get(aggregateId: ID): P =
projections[aggregateId]
?: initialStateBuilder(aggregateId)
/**
* Save the projection.
*/
override fun save(projection: P) {
projections.compute(projection.aggregateId) { id: ID, proj: P? ->
val currentProjection = proj ?: initialStateBuilder(projection.aggregateId)
if (currentProjection.lastEventVersion < projection.lastEventVersion) {
projection
} else {
currentProjection
}
}
}
}

80
src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInRedis.kt Normal file
View File

@@ -0,0 +1,80 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
import io.github.oshai.kotlinlogging.KotlinLogging
import redis.clients.jedis.UnifiedJedis
import redis.clients.jedis.params.ScanParams
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
import kotlin.reflect.KClass
class ProjectionRepositoryInRedis<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
private val jedis: UnifiedJedis,
private val initialStateBuilder: (aggregateId: ID) -> P,
private val projectionClass: KClass<P>,
private val projectionToJson: (P) -> String,
private val jsonToProjection: (String) -> P,
applyToProjection: P.(event: E) -> P,
) : ProjectionRepositoryAbs<E, P, ID>(applyToProjection),
ProjectionRepository<E, P, ID> {
val logger = KotlinLogging.logger { }
private val lock = ReentrantLock()
/**
* Get the list of all [Projections][Projection]
*/
override fun getList(
limit: Int,
offset: Int,
): List<P> =
jedis
.hscan(
projectionClass.redisHKey,
offset.toString(),
ScanParams()
.match("*")
.count(limit),
).result
.mapNotNull {
jsonToProjection(it.value)
}
/**
* Get the [Projection].
*/
override fun get(aggregateId: ID): P =
jedis
.hget(
projectionClass.redisHKey,
aggregateId.id.toString(),
).let {
if (it == null || it == "nil") {
initialStateBuilder(aggregateId)
} else {
jsonToProjection(it)
}
}
override fun save(projection: P) {
lock.withLock {
if (get(projection.aggregateId).lastEventVersion < projection.lastEventVersion) {
jedis.hset(
projection.redisHKey,
projection.aggregateId.id.toString(),
projectionToJson(projection),
)
logger.info { "Projection saved" }
} else {
logger.error { "Projection save SKIP (an early version exists)" }
error("Projection save SKIP (an early version exists)")
}
}
}
}
private val <P : Projection<*>> KClass<P>.redisHKey: String get() =
"projection:$simpleName"
private val <P : Projection<*>> P.redisHKey: String get() =
this::class.redisHKey

35
src/main/kotlin/eventDemo/libs/event/projection/ProjectionSnapshotRepository.kt
View File

@@ -1,35 +0,0 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
interface ProjectionSnapshotRepository<E : Event<ID>, P : Projection<ID>, ID : AggregateId> {
/**
* Create a snapshot for the event
*/
suspend fun applyAndPutToCache(event: E): P
fun count(aggregateId: ID): Int
fun countAll(): Int
/**
* Build the list of all [Projections][Projection]
*/
fun getList(
limit: Int = 100,
offset: Int = 0,
): List<P>
/**
* Build the last version of the [Projection] from the cache.
*/
fun getLast(aggregateId: ID): P
/**
* Build the [Projection] to the specific [event][Event].
*
* It does not contain the [events][Event] it after this one.
*/
fun getUntil(event: E): P
}

226
src/main/kotlin/eventDemo/libs/event/projection/ProjectionSnapshotRepositoryInMemory.kt
View File

@@ -1,226 +0,0 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
import eventDemo.libs.event.EventStore
import eventDemo.libs.event.EventStream
import io.github.oshai.kotlinlogging.KotlinLogging
import io.github.oshai.kotlinlogging.withLoggingContext
import kotlinx.datetime.Clock
import kotlinx.datetime.Instant
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue
import kotlin.reflect.KClass
class ProjectionSnapshotRepositoryInMemory<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
val name: KClass<*> = ProjectionSnapshotRepositoryInMemory::class,
private val eventStore: EventStore<E, ID>,
private val initialStateBuilder: (aggregateId: ID) -> P,
private val snapshotCacheConfig: SnapshotConfig = SnapshotConfig(),
private val applyToProjection: P.(event: E) -> P,
) : ProjectionSnapshotRepository<E, P, ID> {
private val projectionsSnapshot: ConcurrentHashMap<ID, ConcurrentLinkedQueue<Pair<P, Instant>>> = ConcurrentHashMap()
private val logger = KotlinLogging.logger(name.qualifiedName.toString())
/**
* Create a snapshot for the event
*
* 1. get the last snapshot with a version lower than that of the event
* 2. get the events with a greater version of the snapshot
* 3. apply the event to the snapshot
* 4. apply the new event to the projection
* 5. save it
* 6. remove old one
*/
override suspend fun applyAndPutToCache(event: E): P =
getUntil(event)
.also {
withLoggingContext("projection" to it.toString()) {
save(it)
removeOldSnapshot(it.aggregateId)
}
}
override fun count(aggregateId: ID): Int =
projectionsSnapshot[aggregateId]?.count() ?: 0
override fun countAll(): Int =
projectionsSnapshot.mappingCount().toInt()
/**
* Build the list of all [Projections][Projection]
*/
override fun getList(
limit: Int,
offset: Int,
): List<P> =
projectionsSnapshot
.map { (id, b) ->
getLast(id)
}.drop(offset)
.take(limit)
/**
* Build the last version of the [Projection] from the cache.
*
* 1. get the last snapshot
* 2. get the missing event to the snapshot
* 3. apply the missing events to the snapshot
*/
override fun getLast(aggregateId: ID): P {
val lastSnapshot = getLastSnapshot(aggregateId)?.first
val missingEventOfSnapshot = getEventAfterTheSnapshot(aggregateId, lastSnapshot)
return lastSnapshot.applyEvents(aggregateId, missingEventOfSnapshot)
}
/**
* Build the [Projection] to the specific [event][Event].
*
* It does not contain the [events][Event] it after this one.
*
* 1. get the last snapshot before the event
* 2. get the events with a greater version of the snapshot but lower of passed event
* 3. apply the events to the snapshot
*/
override fun getUntil(event: E): P {
val lastSnapshot = getLastSnapshotBeforeOrEqualEvent(event)?.first
if (lastSnapshot?.lastEventVersion == event.version) {
return lastSnapshot
}
val missingEventOfSnapshot =
eventStore
.getStream(event.aggregateId)
// take the last snapshot version +1 to event version
.readVersionBetween(lastSnapshot, event)
return if (lastSnapshot?.lastEventVersion == event.version) {
lastSnapshot
} else {
lastSnapshot.applyEvents(event.aggregateId, missingEventOfSnapshot)
}
}
/**
* Remove the oldest [snapshot][P] of the [queue][projectionsSnapshot].
*
* The rules are pass in the controller.
*/
private fun removeOldSnapshot(aggregateId: ID) {
projectionsSnapshot[aggregateId]?.let { queue ->
if (snapshotCacheConfig.enabled) {
queue
.excludeFirstAndLast()
.excludeTheHeadBySize()
.excludeNewerByDate()
.excludeByModulo()
.forEach { queue.remove(it) }
}
}
}
/**
* Return a new list without the first and last snapshot.
*
* Exclude from deletion the first and the last.
*/
private fun FilteredList<P>.excludeFirstAndLast(): FilteredList<P> =
sortedBy { it.first.lastEventVersion }
.drop(1)
.dropLast(1)
/**
* Return a new list of event filtered by the version modulo.
*
* Exclude from deletion 1 element out of 10 (if modulo 10 in [config][snapshotCacheConfig]).
*/
private fun FilteredList<P>.excludeByModulo(): FilteredList<P> =
filter { (it.first.lastEventVersion % snapshotCacheConfig.modulo) != 1 }
/**
* Return a new list of event filtered by the maximum size.
*
* Exclude from removal all [snapshot][projectionsSnapshot] that in the head of the queue.
*/
private fun FilteredList<P>.excludeTheHeadBySize(): FilteredList<P> {
// filter if size exceeds the limit
return sortedBy { it.first.lastEventVersion }
.dropLast(snapshotCacheConfig.maxSnapshotCacheSize)
}
/**
* Return a new list of event filtered by the maximum date.
*
* Exclude from removal all [snapshot][projectionsSnapshot] that newer of the date (in [config][SnapshotConfig]).
*/
private fun FilteredList<P>.excludeNewerByDate(): FilteredList<P> {
val now = Clock.System.now()
val deadLine = now - snapshotCacheConfig.maxSnapshotCacheTtl
return filter { deadLine < it.second }
}
/**
* Save the snapshot.
*/
private fun save(projection: P) {
projectionsSnapshot
.computeIfAbsent(projection.aggregateId) { ConcurrentLinkedQueue() }
.add(Pair(projection, Clock.System.now()))
.also { logger.info { "Projection saved" } }
}
/**
* Get the last snapshot when the version is lower of then event version
*/
private fun getLastSnapshotBeforeOrEqualEvent(event: E) =
projectionsSnapshot[event.aggregateId]
?.sortedByDescending { it.first.lastEventVersion }
?.find { it.first.lastEventVersion <= event.version }
/**
* Get the last snapshot (with the higher version).
*/
private fun getLastSnapshot(aggregateId: ID) =
projectionsSnapshot[aggregateId]
?.maxByOrNull { it.first.lastEventVersion }
/**
* Get the events from the [event stream][EventStream] when the version is higher of the snapshot.
*
* If the snapshot is null, it takes all events from the event [event stream][EventStream]
*/
private fun getEventAfterTheSnapshot(
aggregateId: ID,
snapshot: P?,
) =
eventStore
.getStream(aggregateId)
.readGreaterOfVersion(snapshot?.lastEventVersion ?: 0)
/**
* Apply events to the projection.
*/
private fun P?.applyEvents(
aggregateId: ID,
eventsToApply: Set<E>,
): P =
eventsToApply.fold(this ?: initialStateBuilder(aggregateId), applyToProjectionSecure)
/**
* Wrap the [applyToProjection] lambda to avoid duplicate apply of the same event.
*/
private val applyToProjectionSecure: P.(event: E) -> P = { event ->
withLoggingContext("event" to event.toString(), "projection" to this.toString()) {
if (event.version == lastEventVersion + 1) {
applyToProjection(event)
} else if (event.version <= lastEventVersion) {
KotlinLogging.logger { }.warn { "Event is already in the Projection, skip apply." }
this
} else {
error("The version of the event must follow directly after the version of the projection.")
}
}
}
}
private typealias FilteredList<P> = Collection<Pair<P, Instant>>

240
src/main/kotlin/eventDemo/libs/event/projection/ProjectionSnapshotRepositoryInRedis.kt
View File

@@ -1,240 +0,0 @@
package eventDemo.libs.event.projection
import eventDemo.libs.event.AggregateId
import eventDemo.libs.event.Event
import eventDemo.libs.event.EventStore
import eventDemo.libs.toRanges
import io.github.oshai.kotlinlogging.KotlinLogging
import io.github.oshai.kotlinlogging.withLoggingContext
import redis.clients.jedis.UnifiedJedis
import redis.clients.jedis.params.ScanParams
import redis.clients.jedis.params.SortingParams
import kotlin.reflect.KClass
class ProjectionSnapshotRepositoryInRedis<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
private val eventStore: EventStore<E, ID>,
private val jedis: UnifiedJedis,
private val initialStateBuilder: (aggregateId: ID) -> P,
private val snapshotCacheConfig: SnapshotConfig = SnapshotConfig(),
private val projectionClass: KClass<P>,
private val projectionToJson: (P) -> String,
private val jsonToProjection: (String) -> P,
private val applyToProjection: P.(event: E) -> P,
) : ProjectionSnapshotRepository<E, P, ID> {
val logger = KotlinLogging.logger { }
/**
* Create a snapshot for the event
*
* 1. get the last snapshot with a version lower than that of the event
* 2. get the events with a greater version of the snapshot
* 3. apply the event to the snapshot
* 4. apply the new event to the projection
* 5. save it
* 6. remove old one
*/
override suspend fun applyAndPutToCache(event: E): P =
getUntil(event)
.also {
withLoggingContext(mapOf("projection" to it.toString(), "event" to event.toString())) {
save(it)
removeOldSnapshot(it.aggregateId, event.version)
}
}
override fun count(aggregateId: ID): Int =
jedis.zcount(projectionClass.redisKey(aggregateId), Double.MIN_VALUE, Double.MAX_VALUE).toInt()
override fun countAll(): Int =
jedis.zcount(projectionClass.redisKey, Double.MIN_VALUE, Double.MAX_VALUE).toInt()
/**
* Get the list of all [Projections][Projection]
*/
override fun getList(
limit: Int,
offset: Int,
): List<P> =
jedis
.scan(
offset.toString(),
ScanParams()
.match(projectionClass.redisKeySearchList)
.count(limit),
).result
.mapNotNull { key ->
getLastByKey(key)
?.let(jsonToProjection)
}
/**
* Get the last version of the [Projection] from the cache.
*
* 1. get the last snapshot
* 2. get the missing event to the snapshot
* 3. apply the missing events to the snapshot
*/
override fun getLast(aggregateId: ID): P =
getLastByKey(projectionClass.redisKey(aggregateId))
?.let(jsonToProjection)
?: initialStateBuilder(aggregateId)
private fun getLastByKey(key: String): String? =
jedis
.sort(
key,
SortingParams()
.desc()
.by("score")
.limit(0, 1),
).firstOrNull()
/**
* Build the [Projection] to the specific [event][Event].
*
* It does not contain the [events][Event] it after this one.
*
* 1. get the last snapshot before the event
* 2. get the events with a greater version of the snapshot but lower of passed event
* 3. apply the events to the snapshot
*/
override fun getUntil(event: E): P {
val lastSnapshot =
jedis
.zrangeByScore(
projectionClass.redisKey(event.aggregateId),
1.0,
event.version.toDouble(),
0,
1,
).firstOrNull()
?.let(jsonToProjection)
if (lastSnapshot?.lastEventVersion == event.version) {
return lastSnapshot
}
if (lastSnapshot != null && lastSnapshot.lastEventVersion > event.version) {
logger.error { "Cannot be apply event on more recent snapshot" }
error("Cannot be apply event on more recent snapshot")
}
val missingEventOfSnapshot =
eventStore
.getStream(event.aggregateId)
// take the last snapshot version +1 to event version
.readVersionBetween(lastSnapshot, event)
return if (lastSnapshot?.lastEventVersion == event.version) {
lastSnapshot
} else {
lastSnapshot.applyEvents(event.aggregateId, missingEventOfSnapshot)
}
}
private fun save(projection: P) {
val added = jedis.zadd(projection.redisKey, projection.lastEventVersion.toDouble(), projectionToJson(projection))
if (added < 1) {
logger.error { "Projection NOT saved (already exists)" }
} else {
logger.info { "Projection saved" }
if (snapshotCacheConfig.maxSnapshotCacheTtl.isFinite()) {
jedis.expire(projection.redisKey, snapshotCacheConfig.maxSnapshotCacheTtl.inWholeSeconds)
}
}
}
/**
* Apply events to the projection.
*/
private fun P?.applyEvents(
aggregateId: ID,
eventsToApply: Set<E>,
): P =
eventsToApply.fold(this ?: initialStateBuilder(aggregateId), applyToProjectionSecure)
/**
* Wrap the [applyToProjection] lambda to avoid duplicate apply of the same event.
*/
private val applyToProjectionSecure: P.(event: E) -> P = { event ->
withLoggingContext("event" to event.toString(), "projection" to this.toString()) {
if (event.version == lastEventVersion + 1) {
applyToProjection(event)
} else if (event.version <= lastEventVersion) {
KotlinLogging.logger { }.warn { "Event is already in the Projection, skip apply." }
this
} else {
error("The version of the event must follow directly after the version of the projection.")
}
}
}
fun removeOldSnapshot(
aggregateId: AggregateId,
lastVersion: Int,
) {
if (snapshotCacheConfig.enabled) {
removeByModulo(aggregateId, lastVersion)
removeTheHeadBySize(aggregateId, lastVersion)
}
}
private fun removeByModulo(
aggregateId: AggregateId,
lastVersion: Int,
) {
(lastVersion - (snapshotCacheConfig.maxSnapshotCacheSize * snapshotCacheConfig.modulo))
.let { if (it < 2) 2 else it }
.let { IntRange(it, lastVersion - 1) }
.filter { (it % snapshotCacheConfig.modulo) != 1 }
.toRanges()
.map {
jedis
.zremrangeByScore(
projectionClass.redisKey(aggregateId),
it.first.toDouble(),
it.last.toDouble(),
).also { removedCount ->
if (removedCount > 0) {
logger.debug {
"$removedCount snapshot removed Modulo(${snapshotCacheConfig.modulo}) (${it.first} to ${it.last}) [lastVersion=$lastVersion]"
}
}
}
}
}
private fun removeTheHeadBySize(
aggregateId: AggregateId,
lastVersion: Int,
) {
(lastVersion - (snapshotCacheConfig.maxSnapshotCacheSize * snapshotCacheConfig.modulo))
.toDouble()
.let {
jedis
.zremrangeByScore(
projectionClass.redisKey(aggregateId),
2.0,
it,
).also { removedCount ->
if (removedCount > 0) {
logger.info {
"$removedCount snapshot removed Size(${snapshotCacheConfig.maxSnapshotCacheSize}) (1.0 to $it) [lastVersion=$lastVersion]"
}
}
}
}
}
}
val <P : Projection<*>> KClass<P>.redisKeySearchList: String get() {
return "projection:$simpleName:*"
}
val <P : Projection<*>> P.redisKey: String get() {
return "projection:${this::class.simpleName}:${aggregateId.id}"
}
fun <P : Projection<*>, A : AggregateId> KClass<P>.redisKey(aggregateId: A): String =
"projection:$simpleName:${aggregateId.id}"
val <P : Projection<*>> KClass<P>.redisKey: String get() =
"projection:$simpleName"

26
src/main/kotlin/eventDemo/libs/event/projection/SnapshotConfig.kt
View File

@@ -1,26 +0,0 @@
package eventDemo.libs.event.projection
import kotlin.time.Duration
import kotlin.time.Duration.Companion.minutes
data class SnapshotConfig(
/**
* Keep snapshot when is on the head of the queue cache
*/
val maxSnapshotCacheSize: Int = 20,
/**
* Keep snapshot when is newer of
*
* snapshot.date > now + maxSnapshotCacheTtl
*/
val maxSnapshotCacheTtl: Duration = 10.minutes,
/**
* Keep snapshot when version is this modulo
*
* snapshot.lastVersion % modulo == 1
*/
val modulo: Int = 10,
val enabled: Boolean = true,
)
val DISABLED_CONFIG = SnapshotConfig(Int.MAX_VALUE, Duration.INFINITE, Int.MAX_VALUE, enabled = false)

17
src/test/kotlin/eventDemo/Helpers.kt
View File

@@ -6,11 +6,17 @@ import eventDemo.business.entity.Deck
import eventDemo.configuration.business.configureGameListener
import eventDemo.configuration.injection.appKoinModule
import eventDemo.configuration.ktor.configuration
import io.github.oshai.kotlinlogging.KotlinLogging
import io.kotest.engine.runBlocking
import io.ktor.server.config.ApplicationConfig
import io.ktor.server.testing.ApplicationTestBuilder
import io.ktor.server.testing.testApplication
import io.ktor.utils.io.KtorDsl
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import org.koin.core.Koin
import org.koin.core.context.startKoin
import org.koin.core.module.KoinApplicationDslMarker
import org.koin.dsl.koinApplication
import org.koin.ktor.ext.getKoin
@@ -39,6 +45,7 @@ fun testApplicationWithConfig(
configBuilder: Koin.() -> Unit = {},
block: suspend ApplicationTestBuilder.() -> Unit,
) {
val logger = KotlinLogging.logger {}
testApplication {
val conf = ApplicationConfig("application.conf")
environment {
@@ -46,11 +53,19 @@ fun testApplicationWithConfig(
}
application {
logger.info { "Config App" }
val koin = getKoin()
koin.cleanDataTest()
runCatching {
logger.info { "Starting A" }
configBuilder(koin)
logger.info { "A finish" }
}
block()
}
logger.info { "Starting B" }
this@testApplication.block()
logger.info { "B finish" }
}
}

11
src/test/kotlin/eventDemo/adapter/interfaceLayer/query/GameListRouteTest.kt
View File

@@ -8,6 +8,7 @@ import eventDemo.business.event.event.NewPlayerEvent
import eventDemo.business.event.event.PlayerReadyEvent
import eventDemo.business.event.projection.GameList
import eventDemo.testApplicationWithConfig
import io.github.oshai.kotlinlogging.KotlinLogging
import io.kotest.assertions.nondeterministic.eventually
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.collections.shouldContain
@@ -19,18 +20,20 @@ import io.ktor.client.request.get
import io.ktor.client.statement.bodyAsText
import io.ktor.http.ContentType
import io.ktor.http.HttpStatusCode
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlin.test.assertEquals
import kotlin.test.assertTrue
import kotlin.time.Duration.Companion.seconds
val logger = KotlinLogging.logger {}
class GameListRouteTest :
FunSpec({
test("/games with no game started") {
testApplicationWithConfig {
val player1 = Player(name = "Nikola")
logger.info { "Starting player1" }
httpClient()
.get("/games") {
withAuth(player1)
@@ -48,16 +51,14 @@ class GameListRouteTest :
val player1 = Player(name = "Nikola")
testApplicationWithConfig(
{
runBlocking {
get<GameEventHandler>()
.handle(gameId) {
NewPlayerEvent(gameId, player1, it)
}
}
},
) {
// Wait until the projection is created
eventually(1.seconds) {
eventually(3.seconds) {
httpClient()
.get("/games") {
withAuth(player1)
@@ -81,7 +82,6 @@ class GameListRouteTest :
val player2 = Player(name = "Einstein")
testApplicationWithConfig({
val eventHandler = get<GameEventHandler>()
runBlocking {
eventHandler.handle(gameId) { NewPlayerEvent(gameId, player1, it) }
eventHandler.handle(gameId) { NewPlayerEvent(gameId, player2, it) }
eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player1, it) }
@@ -94,7 +94,6 @@ class GameListRouteTest :
shuffleIsDisabled = true,
)
}
}
}) {
eventually(1.seconds) {
httpClient()

25
src/test/kotlin/eventDemo/adapter/interfaceLayer/query/GameSimulationTest.kt
View File

@@ -1,7 +1,6 @@
package eventDemo.adapter.interfaceLayer.query
import eventDemo.Tag
import eventDemo.adapter.infrastructureLayer.event.projection.GameStateRepositoryInMemory
import eventDemo.business.command.GameCommandHandler
import eventDemo.business.command.command.GameCommand
import eventDemo.business.command.command.IWantToJoinTheGameCommand
@@ -10,9 +9,9 @@ import eventDemo.business.command.command.IamReadyToPlayCommand
import eventDemo.business.entity.Card
import eventDemo.business.entity.GameId
import eventDemo.business.entity.Player
import eventDemo.business.event.GameEventStore
import eventDemo.business.event.event.disableShuffleDeck
import eventDemo.business.event.projection.GameState
import eventDemo.business.event.projection.GameStateRepository
import eventDemo.business.event.projection.projectionListener.PlayerNotificationListener
import eventDemo.business.notification.CommandSuccessNotification
import eventDemo.business.notification.ItsTheTurnOfNotification
@@ -61,9 +60,9 @@ class GameSimulationTest :
var player1HasJoin = false
testKoinApplicationWithConfig {
val commandHandler by inject<GameCommandHandler>()
val eventStore by inject<GameEventStore>()
val playerNotificationListener by inject<PlayerNotificationListener>()
val commandHandler = get<GameCommandHandler>()
val playerNotificationListener = get<PlayerNotificationListener>()
val gameStateRepository = get<GameStateRepository>()
// Run command handler
// In the normal process, these handlers is invoque players connect to the websocket
@@ -76,8 +75,8 @@ class GameSimulationTest :
}
}
// Consume etch notification of players, and put theses in list.
// is used later to control when other players can be executing the next action
// Consume etch notification of players, and put theses in a list.
// Is used later to control when other players can execute the next action
val player1Notifications = mutableListOf<Notification>()
val player2Notifications = mutableListOf<Notification>()
run {
@@ -94,7 +93,7 @@ class GameSimulationTest :
}
}
// The player 1 actions
// Player 1 actions
val player1Job =
launch {
playerNotificationListener.startListening(player1, gameId) {
@@ -132,11 +131,11 @@ class GameSimulationTest :
}
}
// The player 2 actions
// Player 2 actions
val player2Job =
launch {
// wait the player 1 has join the game
until(1.seconds) { player1HasJoin }
// wait player 1 has joined the game
until(3.seconds) { player1HasJoin }
playerNotificationListener.startListening(player2, gameId) {
channelNotification2.trySendBlocking(it)
@@ -176,7 +175,7 @@ class GameSimulationTest :
joinAll(player1Job, player2Job)
// Build the last state from the event store
val state = GameStateRepositoryInMemory(eventStore = eventStore).getLast(gameId)
val state = gameStateRepository.get(gameId)
// Check if the state is correct
state.aggregateId shouldBeEqual gameId
@@ -192,6 +191,6 @@ class GameSimulationTest :
})
private suspend inline fun <reified T : Notification> MutableList<Notification>.waitNotification(crossinline block: T.() -> Boolean): T =
eventually(1.seconds) {
eventually(3.seconds) {
filterIsInstance<T>().first { block(it) }
}

32
src/test/kotlin/eventDemo/adapter/interfaceLayer/query/GameStateRouteTest.kt
View File

@@ -11,7 +11,9 @@ import eventDemo.business.event.event.disableShuffleDeck
import eventDemo.business.event.projection.GameState
import eventDemo.business.event.projection.GameStateRepository
import eventDemo.testApplicationWithConfig
import io.github.oshai.kotlinlogging.KotlinLogging
import io.kotest.assertions.nondeterministic.eventually
import io.kotest.assertions.nondeterministic.until
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.collections.shouldHaveSize
import io.kotest.matchers.equals.shouldBeEqual
@@ -53,7 +55,6 @@ class GameStateRouteTest :
val gameId = GameId()
val player1 = Player(name = "Nikola")
val player2 = Player(name = "Einstein")
var lastPlayedCard: Card? = null
testApplicationWithConfig({
disableShuffleDeck()
val eventHandler = get<GameEventHandler>()
@@ -64,9 +65,8 @@ class GameStateRouteTest :
eventHandler.handle(gameId) { NewPlayerEvent(gameId, player2, it) }
eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player1, it) }
eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player2, it) }
lastPlayedCard = eventually { stateRepo.getLast(gameId).playableCards(player1).first() }
assertNotNull(lastPlayedCard)
.let { assertIs<Card.NumericCard>(lastPlayedCard) }
val lastPlayedCard = eventually(3.seconds) { stateRepo.get(gameId).playableCards(player1).first() }
assertIs<Card.NumericCard>(lastPlayedCard)
.let {
it.number shouldBeEqual 0
it.color shouldBeEqual Card.Color.Red
@@ -74,14 +74,19 @@ class GameStateRouteTest :
eventHandler.handle(gameId) {
CardIsPlayedEvent(
gameId,
assertNotNull(lastPlayedCard),
lastPlayedCard,
player1,
it,
)
}
until(3.seconds) {
stateRepo
.get(gameId)
.deck.discard
.last() == lastPlayedCard
}
}
}) {
eventually(1.seconds) {
httpClient()
.get("/games/$gameId/state") {
withAuth(player1)
@@ -101,7 +106,6 @@ class GameStateRouteTest :
}
}
}
}
test("""The route "/games/{id}/card/last" should return the last card played of the game""") {
val gameId = GameId()
@@ -118,9 +122,8 @@ class GameStateRouteTest :
eventHandler.handle(gameId) { NewPlayerEvent(gameId, player2, it) }
eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player1, it) }
eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player2, it) }
lastPlayedCard = eventually { stateRepo.getLast(gameId).playableCards(player1).first() }
assertNotNull(lastPlayedCard)
.let { assertIs<Card.NumericCard>(lastPlayedCard) }
lastPlayedCard = eventually(3.seconds) { stateRepo.get(gameId).playableCards(player1).first() }
assertIs<Card.NumericCard>(lastPlayedCard)
.let {
it.number shouldBeEqual 0
it.color shouldBeEqual Card.Color.Red
@@ -133,9 +136,15 @@ class GameStateRouteTest :
it,
)
}
until(3.seconds) {
stateRepo
.get(gameId)
.deck.discard
.last() == lastPlayedCard
}
}
}) {
eventually(1.seconds) {
httpClient()
.get("/games/$gameId/card/last") {
withAuth(player1)
@@ -146,5 +155,4 @@ class GameStateRouteTest :
}
}
}
}
})

3
src/test/kotlin/eventDemo/business/event/GameEventHandlerTest.kt
View File

@@ -7,6 +7,8 @@ import eventDemo.business.entity.Player
import eventDemo.business.event.event.GameEvent
import eventDemo.business.event.event.NewPlayerEvent
import eventDemo.libs.event.VersionBuilderLocal
import eventDemo.testApplicationWithConfig
import io.kotest.assertions.nondeterministic.until
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.collections.shouldHaveSize
import io.kotest.matchers.equals.shouldBeEqual
@@ -16,6 +18,7 @@ import io.mockk.spyk
import io.mockk.verify
import kotlin.test.assertIs
import kotlin.test.assertNotNull
import kotlin.time.Duration.Companion.seconds
class GameEventHandlerTest :
FunSpec({

63
src/test/kotlin/eventDemo/business/event/projection/GameStateRepositoryTest.kt
View File

@@ -2,7 +2,6 @@ package eventDemo.business.event.projection
import ch.qos.logback.classic.Level
import com.rabbitmq.client.impl.ForgivingExceptionHandler
import com.zaxxer.hikari.pool.ProxyConnection
import eventDemo.Tag
import eventDemo.business.command.GameCommandHandler
import eventDemo.business.entity.GameId
@@ -17,7 +16,6 @@ import io.kotest.common.KotestInternal
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.collections.shouldHaveSize
import io.kotest.matchers.equals.shouldBeEqual
import io.kotest.matchers.shouldBe
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.joinAll
@@ -41,13 +39,10 @@ class GameStateRepositoryTest :
val eventHandler = get<GameEventHandler>()
eventHandler
.handle(aggregateId) { NewPlayerEvent(aggregateId = aggregateId, player = player1, version = it) }
.also { event ->
.also {
// Wait until the projection is created
eventually(1.seconds) {
assertNotNull(repo.getUntil(event)).also {
assertNotNull(it.players) shouldBeEqual setOf(player1)
}
assertNotNull(repo.getLast(aggregateId)).also {
assertNotNull(repo.get(aggregateId)).also {
assertNotNull(it.players) shouldBeEqual setOf(player1)
}
}
@@ -68,9 +63,9 @@ class GameStateRepositoryTest :
var state: GameState? = null
projectionBus.subscribe {
repo.getLast(aggregateId).also {
state = it
}
repo
.get(aggregateId)
.also { state = it }
}
eventHandler
@@ -86,7 +81,7 @@ class GameStateRepositoryTest :
.handle(aggregateId) { NewPlayerEvent(aggregateId = aggregateId, player = player2, version = it) }
.also {
eventually(1.seconds) {
assertNotNull(repo.getLast(aggregateId)).also {
assertNotNull(repo.get(aggregateId)).also {
assertNotNull(it.players) shouldBeEqual setOf(player1, player2)
}
}
@@ -95,44 +90,7 @@ class GameStateRepositoryTest :
}
}
test("getUntil should build the state until the event") {
withLogLevel(
GameCommandHandler::class.java.name to Level.ERROR,
ForgivingExceptionHandler::class.java.name to Level.OFF,
ProxyConnection::class.java.name to Level.OFF,
Logger.ROOT_LOGGER_NAME to Level.INFO,
) {
repeat(10) {
val aggregateId = GameId()
testKoinApplicationWithConfig {
val repo = get<GameStateRepository>()
val eventHandler = get<GameEventHandler>()
val event1 =
eventHandler
.handle(aggregateId) { NewPlayerEvent(aggregateId = aggregateId, player = player1, version = it) }
.also { event1 ->
assertNotNull(repo.getUntil(event1)).also {
assertNotNull(it.players) shouldBeEqual setOf(player1)
}
}
eventHandler
.handle(aggregateId) { NewPlayerEvent(aggregateId = aggregateId, player = player2, version = it) }
.also { event2 ->
assertNotNull(repo.getUntil(event2)).also {
assertNotNull(it.players) shouldBeEqual setOf(player1, player2)
}
assertNotNull(repo.getUntil(event1)).also {
assertNotNull(it.players) shouldBeEqual setOf(player1)
}
}
}
}
}
}
test("getUntil should be concurrently secure").config(tags = setOf(Tag.Concurrence)) {
test("get should be concurrently secure").config(tags = setOf(Tag.Concurrence)) {
withLogLevel(
Logger.ROOT_LOGGER_NAME to Level.ERROR,
ForgivingExceptionHandler::class.java.name to Level.OFF,
@@ -167,17 +125,12 @@ class GameStateRepositoryTest :
includeFirst = false
},
) {
repo.getLast(aggregateId).run {
repo.get(aggregateId).run {
lastEventVersion shouldBeEqual 200
players shouldHaveSize 200
}
repo.count(aggregateId) shouldBe 21
}
}
}
}
xtest("get should be concurrently secure") {
tags(Tag.Concurrence)
}
})

38
src/test/kotlin/eventDemo/libs/event/EventHandlerTest.kt Normal file
View File

@@ -0,0 +1,38 @@
package eventDemo.libs.event
import eventDemo.libs.bus.Bus
import eventDemo.libs.bus.BusInMemory
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.equals.shouldBeEqual
class EventHandlerTest :
FunSpec({
test("EventHandler::handle should returns the built event") {
val eventBus: Bus<EventXTest> = BusInMemory()
val eventStore: EventStore<EventXTest, IdTest> = EventStoreInMemory()
val versionBuilder: VersionBuilder = VersionBuilderLocal()
val aggregateId: IdTest = IdTest()
val handler =
EventHandlerImpl(
eventBus,
eventStore,
versionBuilder,
)
// When
val event =
handler.handle(aggregateId) {
EventXTest(aggregateId = aggregateId, version = it, num = 1)
}
// Then
event.aggregateId shouldBeEqual aggregateId
event.version shouldBeEqual 1
}
xtest("EventHandler::handle should publish the event into the store")
xtest("EventHandler::handle should publish the event into the bus")
xtest("EventHandler::handle should publish the event into the bus in incremental order")
})

118
src/test/kotlin/eventDemo/business/event/projection/ProjectionSnapshotRepositoryTest.kt → src/test/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryTest.kt
View File

@@ -1,4 +1,4 @@
package eventDemo.business.event.projection
package eventDemo.libs.event.projection
import eventDemo.cleanProjections
import eventDemo.configuration.serializer.UUIDSerializer
@@ -7,12 +7,6 @@ import eventDemo.libs.event.Event
import eventDemo.libs.event.EventStore
import eventDemo.libs.event.EventStoreInMemory
import eventDemo.libs.event.VersionBuilderLocal
import eventDemo.libs.event.projection.DISABLED_CONFIG
import eventDemo.libs.event.projection.Projection
import eventDemo.libs.event.projection.ProjectionSnapshotRepository
import eventDemo.libs.event.projection.ProjectionSnapshotRepositoryInMemory
import eventDemo.libs.event.projection.ProjectionSnapshotRepositoryInRedis
import eventDemo.libs.event.projection.SnapshotConfig
import io.kotest.assertions.nondeterministic.continually
import io.kotest.core.spec.style.FunSpec
import io.kotest.datatest.withData
@@ -22,6 +16,7 @@ import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.datetime.Clock
import kotlinx.datetime.Instant
import kotlinx.serialization.Serializable
@@ -34,14 +29,14 @@ import kotlin.test.assertNotNull
import kotlin.time.Duration.Companion.seconds
@OptIn(DelicateCoroutinesApi::class)
class ProjectionSnapshotRepositoryTest :
class ProjectionRepositoryTest :
FunSpec({
data class TestData(
val store: EventStore<TestEvents, IdTest>,
val snapshotRepo: ProjectionSnapshotRepository<TestEvents, ProjectionTest, IdTest>,
val repository: ProjectionRepository<TestEvents, ProjectionTest, IdTest>,
) : WithDataTestName {
override fun dataTestName(): String =
"${snapshotRepo::class.simpleName} with ${store::class.simpleName}"
"${repository::class.simpleName} with ${store::class.simpleName}"
}
val eventStores =
@@ -50,48 +45,40 @@ class ProjectionSnapshotRepositoryTest :
)
val projectionRepo =
listOf(
::getSnapshotRepoInMemoryTest,
::getSnapshotRepoInRedisTest,
::getRepoInMemoryTest,
::getRepoInRedisTest,
)
val list =
eventStores.flatMap { store ->
projectionRepo.map { repo ->
store().let { store -> TestData(store, repo(store, DISABLED_CONFIG)) }
TestData(store(), repo())
}
}
context("when call applyAndPutToCache, the getUntil method must be use the built projection cache") {
context("when call applyAndSave, the projection should be built and save to the repository") {
withData(list) { (eventStore, repo) ->
val aggregateId = IdTest()
val eventOther = Event2Test(value2 = "valOther", version = 1, aggregateId = IdTest())
eventStore.publish(eventOther)
repo.applyAndPutToCache(eventOther)
assertNotNull(repo.getUntil(eventOther)).also {
// eventStore.publish(eventOther)
val p = repo.applyAndSave(eventOther)
println(p)
assertNotNull(repo.get(eventOther.aggregateId)).also {
assertNotNull(it.value) shouldBeEqual "valOther"
}
val event1 = Event1Test(value1 = "val1", version = 1, aggregateId = aggregateId)
eventStore.publish(event1)
repo.applyAndPutToCache(event1)
assertNotNull(repo.getLast(event1.aggregateId)).also {
assertNotNull(it.value) shouldBeEqual "val1"
}
assertNotNull(repo.getUntil(event1)).also {
// eventStore.publish(event1)
repo.applyAndSave(event1)
assertNotNull(repo.get(event1.aggregateId)).also {
assertNotNull(it.value) shouldBeEqual "val1"
}
val event2 = Event2Test(value2 = "val2", version = 2, aggregateId = aggregateId)
eventStore.publish(event2)
repo.applyAndPutToCache(event2)
assertNotNull(repo.getLast(event2.aggregateId)).also {
assertNotNull(it.value) shouldBeEqual "val1val2"
}
assertNotNull(repo.getUntil(event1)).also {
assertNotNull(it.value) shouldBeEqual "val1"
}
assertNotNull(repo.getUntil(event2)).also {
// eventStore.publish(event2)
repo.applyAndSave(event2)
assertNotNull(repo.get(event2.aggregateId)).also {
assertNotNull(it.value) shouldBeEqual "val1val2"
}
}
@@ -104,18 +91,18 @@ class ProjectionSnapshotRepositoryTest :
val eventOther = Event2Test(value2 = "valOther", version = 1, aggregateId = otherAggregateId)
eventStore.publish(eventOther)
repo.applyAndPutToCache(eventOther)
assertNotNull(repo.getUntil(eventOther)).also {
repo.applyAndSave(eventOther)
assertNotNull(repo.get(eventOther.aggregateId)).also {
assertNotNull(it.value) shouldBeEqual "valOther"
}
val event1 = Event1Test(value1 = "val1", version = 1, aggregateId = aggregateId)
eventStore.publish(event1)
repo.applyAndPutToCache(event1)
repo.applyAndSave(event1)
val event2 = Event2Test(value2 = "val2", version = 2, aggregateId = aggregateId)
eventStore.publish(event2)
repo.applyAndPutToCache(event2)
repo.applyAndSave(event2)
repo.getList().apply {
any { it.aggregateId == otherAggregateId } shouldBeEqual true
@@ -128,7 +115,7 @@ class ProjectionSnapshotRepositoryTest :
}
}
context("ProjectionSnapshotRepository should be thread safe") {
context("ProjectionRepository should be thread safe") {
continually(1.seconds) {
withData(list) { (eventStore, repo) ->
val aggregateId = IdTest()
@@ -137,43 +124,24 @@ class ProjectionSnapshotRepositoryTest :
(0..9)
.map {
GlobalScope.launch {
(1..10).forEach {
val eventX =
repeat(10) {
lock.withLock {
EventXTest(num = 1, version = versionBuilder.buildNextVersion(aggregateId), aggregateId = aggregateId)
.also { eventStore.publish(it) }
runBlocking {
EventXTest(
num = 1,
version = versionBuilder.buildNextVersion(aggregateId),
aggregateId = aggregateId,
).also { repo.applyAndSave(it) }
}
}
repo.applyAndPutToCache(eventX)
}
}
}.joinAll()
assertNotNull(repo.getLast(aggregateId)).num shouldBeEqual 100
assertNotNull(repo.count(aggregateId)) shouldBeEqual 100
assertNotNull(repo.get(aggregateId)).lastEventVersion shouldBeEqual 100
assertNotNull(repo.get(aggregateId)).num shouldBeEqual 100
}
}
}
context("removeOldSnapshot") {
withData(list) { (eventStore, repo) ->
val versionBuilder = VersionBuilderLocal()
val aggregateId = IdTest()
suspend fun buildEndSendEventX() {
EventXTest(num = 1, version = versionBuilder.buildNextVersion(aggregateId), aggregateId = aggregateId)
.also { eventStore.publish(it) }
.also { repo.applyAndPutToCache(it) }
}
buildEndSendEventX()
repo.getLast(aggregateId).num shouldBeEqual 1
buildEndSendEventX()
repo.getLast(aggregateId).num shouldBeEqual 2
buildEndSendEventX()
repo.getLast(aggregateId).num shouldBeEqual 3
buildEndSendEventX()
repo.getLast(aggregateId).num shouldBeEqual 4
}
}
})
@JvmInline
@@ -217,28 +185,18 @@ private data class EventXTest(
val num: Int,
) : TestEvents
private fun getSnapshotRepoInMemoryTest(
eventStore: EventStore<TestEvents, IdTest>,
snapshotConfig: SnapshotConfig,
): ProjectionSnapshotRepository<TestEvents, ProjectionTest, IdTest> =
ProjectionSnapshotRepositoryInMemory(
eventStore = eventStore,
private fun getRepoInMemoryTest(): ProjectionRepository<TestEvents, ProjectionTest, IdTest> =
ProjectionRepositoryInMemory(
initialStateBuilder = { aggregateId: IdTest -> ProjectionTest(aggregateId) },
snapshotCacheConfig = snapshotConfig,
applyToProjection = apply,
)
private fun getSnapshotRepoInRedisTest(
eventStore: EventStore<TestEvents, IdTest>,
snapshotConfig: SnapshotConfig,
): ProjectionSnapshotRepository<TestEvents, ProjectionTest, IdTest> {
private fun getRepoInRedisTest(): ProjectionRepository<TestEvents, ProjectionTest, IdTest> {
val jedis = JedisPooled("redis://localhost:6379")
jedis.cleanProjections()
return ProjectionSnapshotRepositoryInRedis(
eventStore = eventStore,
return ProjectionRepositoryInRedis(
jedis = jedis,
initialStateBuilder = { aggregateId: IdTest -> ProjectionTest(aggregateId) },
snapshotCacheConfig = snapshotConfig,
projectionClass = ProjectionTest::class,
projectionToJson = { Json.encodeToString(it) },
jsonToProjection = { Json.decodeFromString(it) },