Add Comments
This commit is contained in:
@@ -31,6 +31,9 @@ class GameRoute(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* API route to send a request to play card.
|
||||||
|
*/
|
||||||
fun Routing.playNewCard() {
|
fun Routing.playNewCard() {
|
||||||
val commandStream by inject<GameCommandStream>()
|
val commandStream by inject<GameCommandStream>()
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,9 @@ import eventDemo.shared.entity.Game
|
|||||||
import kotlinx.serialization.SerialName
|
import kotlinx.serialization.SerialName
|
||||||
import kotlinx.serialization.Serializable
|
import kotlinx.serialization.Serializable
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A command to perform an action to play a new card
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("PlayCard")
|
@SerialName("PlayCard")
|
||||||
data class PlayCardCommand(
|
data class PlayCardCommand(
|
||||||
|
|||||||
@@ -9,13 +9,17 @@ import kotlinx.coroutines.launch
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Listen [PlayCardCommand] on [GameCommandStream], check the validity and execute the action.
|
* Listen [PlayCardCommand] on [GameCommandStream], check the validity and execute the action.
|
||||||
|
*
|
||||||
* This action produces a new [CardIsPlayedEvent]
|
* This action produces a new [CardIsPlayedEvent]
|
||||||
*/
|
*/
|
||||||
class PlayCardCommandHandler(
|
class PlayCardCommandHandler(
|
||||||
private val commandStream: GameCommandStream,
|
private val commandStream: GameCommandStream,
|
||||||
private val eventStream: GameEventStream,
|
private val eventStream: GameEventStream,
|
||||||
) {
|
) {
|
||||||
operator fun invoke() {
|
/**
|
||||||
|
* Init the handler
|
||||||
|
*/
|
||||||
|
fun init() {
|
||||||
CoroutineScope(Dispatchers.IO).launch {
|
CoroutineScope(Dispatchers.IO).launch {
|
||||||
commandStream.process {
|
commandStream.process {
|
||||||
// TODO check the command can be executed
|
// TODO check the command can be executed
|
||||||
|
|||||||
@@ -27,6 +27,9 @@ class Game(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* API route to read the last card played.
|
||||||
|
*/
|
||||||
fun Routing.readLastPlayedCard() {
|
fun Routing.readLastPlayedCard() {
|
||||||
val eventStream by inject<GameEventStream>()
|
val eventStream by inject<GameEventStream>()
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,9 @@ import eventDemo.plugins.CommandIdSerializer
|
|||||||
import kotlinx.serialization.Serializable
|
import kotlinx.serialization.Serializable
|
||||||
import java.util.UUID
|
import java.util.UUID
|
||||||
|
|
||||||
|
/**
|
||||||
|
* An ID for the [Command]
|
||||||
|
*/
|
||||||
@JvmInline
|
@JvmInline
|
||||||
@Serializable(with = CommandIdSerializer::class)
|
@Serializable(with = CommandIdSerializer::class)
|
||||||
value class CommandId(
|
value class CommandId(
|
||||||
@@ -14,6 +17,11 @@ value class CommandId(
|
|||||||
override fun toString(): String = id.toString()
|
override fun toString(): String = id.toString()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Interface to represent a Command.
|
||||||
|
*
|
||||||
|
* A command is a request for an action.
|
||||||
|
*/
|
||||||
interface Command {
|
interface Command {
|
||||||
val id: CommandId
|
val id: CommandId
|
||||||
val name: String
|
val name: String
|
||||||
|
|||||||
@@ -2,6 +2,11 @@ package eventDemo.libs.command
|
|||||||
|
|
||||||
import kotlin.reflect.KClass
|
import kotlin.reflect.KClass
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Represent a Command stream.
|
||||||
|
*
|
||||||
|
* The stream contains a list of all actions yet to be executed.
|
||||||
|
*/
|
||||||
interface CommandStream<C : Command> {
|
interface CommandStream<C : Command> {
|
||||||
/**
|
/**
|
||||||
* Send a new [Command] to the queue.
|
* Send a new [Command] to the queue.
|
||||||
@@ -22,7 +27,7 @@ interface CommandStream<C : Command> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A class to implement succes/faild action.
|
* A class to implement success/failed action.
|
||||||
*/
|
*/
|
||||||
interface ComputeStatus {
|
interface ComputeStatus {
|
||||||
fun ack()
|
fun ack()
|
||||||
@@ -30,7 +35,10 @@ interface CommandStream<C : Command> {
|
|||||||
fun nack()
|
fun nack()
|
||||||
}
|
}
|
||||||
|
|
||||||
suspend fun process(block: CommandBlock<C>)
|
/**
|
||||||
|
* Apply an action to all command income in the stream.
|
||||||
|
*/
|
||||||
|
suspend fun process(action: CommandBlock<C>)
|
||||||
}
|
}
|
||||||
|
|
||||||
suspend inline fun <reified C : Command> CommandStream<C>.send(vararg command: C) = send(C::class, *command)
|
suspend inline fun <reified C : Command> CommandStream<C>.send(vararg command: C) = send(C::class, *command)
|
||||||
|
|||||||
@@ -31,18 +31,18 @@ abstract class CommandStreamInMemory<C : Command> : CommandStream<C> {
|
|||||||
queue.send(command)
|
queue.send(command)
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun process(block: CommandBlock<C>) {
|
override suspend fun process(action: CommandBlock<C>) {
|
||||||
queue.consumeEach { command ->
|
queue.consumeEach { command ->
|
||||||
compute(command, block)
|
compute(command, action)
|
||||||
}
|
}
|
||||||
for (command in queue) {
|
for (command in queue) {
|
||||||
compute(command, block)
|
compute(command, action)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun compute(
|
private fun compute(
|
||||||
command: C,
|
command: C,
|
||||||
block: CommandBlock<C>,
|
action: CommandBlock<C>,
|
||||||
) {
|
) {
|
||||||
val status = object : CommandStream.ComputeStatus {
|
val status = object : CommandStream.ComputeStatus {
|
||||||
var isSet: Boolean = false
|
var isSet: Boolean = false
|
||||||
@@ -58,7 +58,7 @@ abstract class CommandStreamInMemory<C : Command> : CommandStream<C> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (runCatching { status.block(command) }.isFailure) {
|
if (runCatching { status.action(command) }.isFailure) {
|
||||||
markAsFailed(command)
|
markAsFailed(command)
|
||||||
} else if (!status.isSet) {
|
} else if (!status.isSet) {
|
||||||
status.ack()
|
status.ack()
|
||||||
@@ -75,7 +75,7 @@ abstract class CommandStreamInMemory<C : Command> : CommandStream<C> {
|
|||||||
private fun <C : Command> markAsFailed(command: C) {
|
private fun <C : Command> markAsFailed(command: C) {
|
||||||
failedCommand.add(command)
|
failedCommand.add(command)
|
||||||
logger.atWarn {
|
logger.atWarn {
|
||||||
message = "Compute command FAILD and it put on the top of the stack : $command"
|
message = "Compute command FAILED and it put it ot the top of the stack : $command"
|
||||||
payload = mapOf("command" to command)
|
payload = mapOf("command" to command)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,10 +2,18 @@ package eventDemo.libs.event
|
|||||||
|
|
||||||
import java.util.UUID
|
import java.util.UUID
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Represent an ID for one aggregate, and it used in events
|
||||||
|
* @see Event
|
||||||
|
*/
|
||||||
interface AggregateId {
|
interface AggregateId {
|
||||||
val id: UUID
|
val id: UUID
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The basic interface for an Event
|
||||||
|
* @see EventStream
|
||||||
|
*/
|
||||||
interface Event<ID : AggregateId> {
|
interface Event<ID : AggregateId> {
|
||||||
val id: ID
|
val id: ID
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,17 +3,25 @@ package eventDemo.libs.event
|
|||||||
import kotlinx.coroutines.flow.Flow
|
import kotlinx.coroutines.flow.Flow
|
||||||
import kotlin.reflect.KClass
|
import kotlin.reflect.KClass
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Interface representing an event stream for publishing and reading domain events
|
||||||
|
*/
|
||||||
interface EventStream<E : Event<ID>, ID : AggregateId> {
|
interface EventStream<E : Event<ID>, ID : AggregateId> {
|
||||||
|
/** Publishes a single event to the event stream */
|
||||||
fun publish(event: E)
|
fun publish(event: E)
|
||||||
|
|
||||||
|
/** Publishes multiple events to the event stream */
|
||||||
fun publish(vararg events: E)
|
fun publish(vararg events: E)
|
||||||
|
|
||||||
|
/** Reads the last event associated with a given aggregate ID */
|
||||||
fun readLast(aggregateId: ID): E?
|
fun readLast(aggregateId: ID): E?
|
||||||
|
|
||||||
|
/** Reads the last event of a specific type associated with a given aggregate ID */
|
||||||
fun <R : E> readLastOf(
|
fun <R : E> readLastOf(
|
||||||
aggregateId: ID,
|
aggregateId: ID,
|
||||||
eventType: KClass<out R>,
|
eventType: KClass<out R>,
|
||||||
): E?
|
): E?
|
||||||
|
|
||||||
|
/** Reads all events associated with a given aggregate ID as a Flow (asynchronous stream) */
|
||||||
fun readAll(aggregateId: ID): Flow<E>
|
fun readAll(aggregateId: ID): Flow<E>
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,11 @@ import kotlinx.coroutines.flow.Flow
|
|||||||
import kotlinx.coroutines.flow.flow
|
import kotlinx.coroutines.flow.flow
|
||||||
import kotlin.reflect.KClass
|
import kotlin.reflect.KClass
|
||||||
|
|
||||||
|
/**
|
||||||
|
* An In-Memory implementation of an event stream.
|
||||||
|
*
|
||||||
|
* All methods are implemented.
|
||||||
|
*/
|
||||||
abstract class EventStreamInMemory<E : Event<ID>, ID : AggregateId>(
|
abstract class EventStreamInMemory<E : Event<ID>, ID : AggregateId>(
|
||||||
private val eventType: Class<E>,
|
private val eventType: Class<E>,
|
||||||
) : EventStream<E, ID> {
|
) : EventStream<E, ID> {
|
||||||
|
|||||||
@@ -4,6 +4,11 @@ import eventDemo.app.actions.playNewCard.PlayCardCommandHandler
|
|||||||
import io.ktor.server.application.Application
|
import io.ktor.server.application.Application
|
||||||
import org.koin.java.KoinJavaComponent.getKoin
|
import org.koin.java.KoinJavaComponent.getKoin
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Configure the command handler for the PlayCard.
|
||||||
|
*/
|
||||||
fun Application.configureCommandHandler() {
|
fun Application.configureCommandHandler() {
|
||||||
getKoin().get<PlayCardCommandHandler>()()
|
getKoin()
|
||||||
|
.get<PlayCardCommandHandler>()
|
||||||
|
.init()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ fun Application.configureKoin() {
|
|||||||
|
|
||||||
val appModule =
|
val appModule =
|
||||||
module {
|
module {
|
||||||
singleOf<GameEventStream>(::GameEventStream)
|
singleOf(::GameEventStream)
|
||||||
singleOf<GameCommandStream>(::GameCommandStream)
|
singleOf(::GameCommandStream)
|
||||||
singleOf(::PlayCardCommandHandler)
|
singleOf(::PlayCardCommandHandler)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,9 +2,13 @@ package eventDemo.shared
|
|||||||
|
|
||||||
import eventDemo.libs.event.AggregateId
|
import eventDemo.libs.event.AggregateId
|
||||||
import eventDemo.plugins.GameIdSerializer
|
import eventDemo.plugins.GameIdSerializer
|
||||||
|
import eventDemo.shared.entity.Game
|
||||||
import kotlinx.serialization.Serializable
|
import kotlinx.serialization.Serializable
|
||||||
import java.util.UUID
|
import java.util.UUID
|
||||||
|
|
||||||
|
/**
|
||||||
|
* An [AggregateId] for the [Game].
|
||||||
|
*/
|
||||||
@JvmInline
|
@JvmInline
|
||||||
@Serializable(with = GameIdSerializer::class)
|
@Serializable(with = GameIdSerializer::class)
|
||||||
value class GameId(
|
value class GameId(
|
||||||
|
|||||||
@@ -3,4 +3,7 @@ package eventDemo.shared.command
|
|||||||
import eventDemo.app.actions.playNewCard.PlayCardCommand
|
import eventDemo.app.actions.playNewCard.PlayCardCommand
|
||||||
import eventDemo.libs.command.CommandStreamInMemory
|
import eventDemo.libs.command.CommandStreamInMemory
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A stream to publish and read the played card command.
|
||||||
|
*/
|
||||||
class GameCommandStream : CommandStreamInMemory<PlayCardCommand>()
|
class GameCommandStream : CommandStreamInMemory<PlayCardCommand>()
|
||||||
|
|||||||
@@ -3,8 +3,14 @@ package eventDemo.shared.entity
|
|||||||
import kotlinx.serialization.SerialName
|
import kotlinx.serialization.SerialName
|
||||||
import kotlinx.serialization.Serializable
|
import kotlinx.serialization.Serializable
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A Play card
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
sealed interface Card {
|
sealed interface Card {
|
||||||
|
/**
|
||||||
|
* The color of a card
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
enum class Color {
|
enum class Color {
|
||||||
Blue,
|
Blue,
|
||||||
@@ -13,6 +19,9 @@ sealed interface Card {
|
|||||||
Green,
|
Green,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A play card with color and number
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("Simple")
|
@SerialName("Simple")
|
||||||
data class NumericCard(
|
data class NumericCard(
|
||||||
@@ -22,30 +31,45 @@ sealed interface Card {
|
|||||||
|
|
||||||
sealed interface Special : Card
|
sealed interface Special : Card
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A revert card to revert the order of the turn.
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("Reverse")
|
@SerialName("Reverse")
|
||||||
data class ReverseCard(
|
data class ReverseCard(
|
||||||
val color: Color,
|
val color: Color,
|
||||||
) : Special
|
) : Special
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A pass card to pass the turn of the next player.
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("Pass")
|
@SerialName("Pass")
|
||||||
data class PassCard(
|
data class PassCard(
|
||||||
val color: Color,
|
val color: Color,
|
||||||
) : Special
|
) : Special
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A play card to force the next player to take 2 card and pass the turn.
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("Plus2")
|
@SerialName("Plus2")
|
||||||
data class Plus2Card(
|
data class Plus2Card(
|
||||||
val color: Color,
|
val color: Color,
|
||||||
) : Special
|
) : Special
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A play card to force the next player to take 4 card and pass the turn.
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("Plus4")
|
@SerialName("Plus4")
|
||||||
data class Plus4Card(
|
data class Plus4Card(
|
||||||
val nextColor: Color,
|
val nextColor: Color,
|
||||||
) : Special
|
) : Special
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A play card to change the color.
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
@SerialName("ChangeColor")
|
@SerialName("ChangeColor")
|
||||||
data class ChangeColorCard(
|
data class ChangeColorCard(
|
||||||
|
|||||||
@@ -3,6 +3,9 @@ package eventDemo.shared.entity
|
|||||||
import eventDemo.shared.GameId
|
import eventDemo.shared.GameId
|
||||||
import kotlinx.serialization.Serializable
|
import kotlinx.serialization.Serializable
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Represent a Game
|
||||||
|
*/
|
||||||
@Serializable
|
@Serializable
|
||||||
data class Game(
|
data class Game(
|
||||||
val id: GameId,
|
val id: GameId,
|
||||||
|
|||||||
@@ -3,11 +3,18 @@ package eventDemo.shared.event
|
|||||||
import eventDemo.libs.event.Event
|
import eventDemo.libs.event.Event
|
||||||
import eventDemo.shared.GameId
|
import eventDemo.shared.GameId
|
||||||
import eventDemo.shared.entity.Card
|
import eventDemo.shared.entity.Card
|
||||||
|
import eventDemo.shared.entity.Game
|
||||||
|
|
||||||
|
/**
|
||||||
|
* An [Event] of a [Game].
|
||||||
|
*/
|
||||||
sealed interface GameEvent : Event<GameId> {
|
sealed interface GameEvent : Event<GameId> {
|
||||||
override val id: GameId
|
override val id: GameId
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* An [Event] to represent a played card.
|
||||||
|
*/
|
||||||
data class CardIsPlayedEvent(
|
data class CardIsPlayedEvent(
|
||||||
override val id: GameId,
|
override val id: GameId,
|
||||||
val card: Card,
|
val card: Card,
|
||||||
|
|||||||
@@ -3,4 +3,7 @@ package eventDemo.shared.event
|
|||||||
import eventDemo.libs.event.EventStreamInMemory
|
import eventDemo.libs.event.EventStreamInMemory
|
||||||
import eventDemo.shared.GameId
|
import eventDemo.shared.GameId
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A stream to publish and read the played card event.
|
||||||
|
*/
|
||||||
class GameEventStream : EventStreamInMemory<GameEvent, GameId>(GameEvent::class.java)
|
class GameEventStream : EventStreamInMemory<GameEvent, GameId>(GameEvent::class.java)
|
||||||
|
|||||||
Reference in New Issue
Block a user