feat(events): migrate game-creation and bot flows to Redis Streams NCS-89 #62

Merged
Janis merged 3 commits from feat/NCS-89-redis-eventbus-streams into main 2026-06-09 10:31:32 +02:00
3 changed files with 28 additions and 22 deletions
Showing only changes of commit 5f35318ebd - Show all commits
@@ -8,10 +8,11 @@ import de.nowchess.api.player.PlayerType
import de.nowchess.api.event.{EventEnvelope, EventType} import de.nowchess.api.event.{EventEnvelope, EventType}
import io.quarkus.redis.datasource.RedisDataSource import io.quarkus.redis.datasource.RedisDataSource
import io.quarkus.redis.datasource.stream.{StreamMessage, XAddArgs, XGroupCreateArgs, XReadGroupArgs} import io.quarkus.redis.datasource.stream.{StreamMessage, XAddArgs, XGroupCreateArgs, XReadGroupArgs}
import io.quarkus.runtime.Startup import io.quarkus.runtime.StartupEvent
import jakarta.annotation.PostConstruct
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import jakarta.enterprise.event.Observes
import jakarta.inject.Inject import jakarta.inject.Inject
import org.eclipse.microprofile.config.inject.ConfigProperty
import org.eclipse.microprofile.context.ManagedExecutor import org.eclipse.microprofile.context.ManagedExecutor
import org.jboss.logging.Logger import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
@@ -21,7 +22,6 @@ import java.time.Duration
import java.util.UUID import java.util.UUID
import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, TimeUnit} import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, TimeUnit}
@Startup
@ApplicationScoped @ApplicationScoped
class GameCreationStreamClient: class GameCreationStreamClient:
@@ -30,6 +30,8 @@ class GameCreationStreamClient:
@Inject var redisConfig: RedisConfig = uninitialized @Inject var redisConfig: RedisConfig = uninitialized
@Inject var objectMapper: ObjectMapper = uninitialized @Inject var objectMapper: ObjectMapper = uninitialized
@Inject var executor: ManagedExecutor = uninitialized @Inject var executor: ManagedExecutor = uninitialized
@ConfigProperty(name = "nowchess.game-creation-stream.enabled", defaultValue = "true")
private var streamEnabled: Boolean = true
// scalafix:on DisableSyntax.var // scalafix:on DisableSyntax.var
private val log = Logger.getLogger(classOf[GameCreationStreamClient]) private val log = Logger.getLogger(classOf[GameCreationStreamClient])
@@ -44,8 +46,8 @@ class GameCreationStreamClient:
private def requestStream: String = s"${redisConfig.prefix}:game-creation" private def requestStream: String = s"${redisConfig.prefix}:game-creation"
private def responseStream: String = s"${redisConfig.prefix}:game-creation-response" private def responseStream: String = s"${redisConfig.prefix}:game-creation-response"
@PostConstruct def start(@Observes _ev: StartupEvent): Unit =
def start(): Unit = if streamEnabled then
createGroupIfAbsent() createGroupIfAbsent()
executor.submit( executor.submit(
new Runnable: new Runnable:
@@ -34,3 +34,5 @@ nowchess:
secret: test-secret secret: test-secret
auth: auth:
enabled: false enabled: false
game-creation-stream:
enabled: false
@@ -7,10 +7,11 @@ import de.nowchess.chess.config.RedisConfig
import de.nowchess.chess.service.GameCreationService import de.nowchess.chess.service.GameCreationService
import io.quarkus.redis.datasource.RedisDataSource import io.quarkus.redis.datasource.RedisDataSource
import io.quarkus.redis.datasource.stream.{StreamMessage, XAddArgs, XGroupCreateArgs, XReadGroupArgs} import io.quarkus.redis.datasource.stream.{StreamMessage, XAddArgs, XGroupCreateArgs, XReadGroupArgs}
import io.quarkus.runtime.Startup import io.quarkus.runtime.StartupEvent
import jakarta.annotation.PostConstruct
import jakarta.enterprise.context.ApplicationScoped import jakarta.enterprise.context.ApplicationScoped
import jakarta.enterprise.event.Observes
import jakarta.inject.Inject import jakarta.inject.Inject
import org.eclipse.microprofile.config.inject.ConfigProperty
import org.eclipse.microprofile.context.ManagedExecutor import org.eclipse.microprofile.context.ManagedExecutor
import org.jboss.logging.Logger import org.jboss.logging.Logger
import scala.compiletime.uninitialized import scala.compiletime.uninitialized
@@ -19,7 +20,6 @@ import scala.util.{Failure, Success, Try}
import java.time.Duration import java.time.Duration
import java.util.UUID import java.util.UUID
@Startup
@ApplicationScoped @ApplicationScoped
class GameCreationStreamListener: class GameCreationStreamListener:
@@ -29,6 +29,8 @@ class GameCreationStreamListener:
@Inject var creationService: GameCreationService = uninitialized @Inject var creationService: GameCreationService = uninitialized
@Inject var executor: ManagedExecutor = uninitialized @Inject var executor: ManagedExecutor = uninitialized
@Inject var redisConfig: RedisConfig = uninitialized @Inject var redisConfig: RedisConfig = uninitialized
@ConfigProperty(name = "nowchess.game-creation-stream.enabled", defaultValue = "true")
private var streamEnabled: Boolean = true
// scalafix:on DisableSyntax.var // scalafix:on DisableSyntax.var
private val log = Logger.getLogger(classOf[GameCreationStreamListener]) private val log = Logger.getLogger(classOf[GameCreationStreamListener])
@@ -41,8 +43,8 @@ class GameCreationStreamListener:
private def responseStream: String = s"${redisConfig.prefix}:game-creation-response" private def responseStream: String = s"${redisConfig.prefix}:game-creation-response"
private def dlqStream: String = s"${redisConfig.prefix}:dlq" private def dlqStream: String = s"${redisConfig.prefix}:dlq"
@PostConstruct def start(@Observes _ev: StartupEvent): Unit =
def start(): Unit = if streamEnabled then
createGroupIfAbsent() createGroupIfAbsent()
executor.submit( executor.submit(
new Runnable: new Runnable: