Merge pull request #16 from dalbodeule/develop

CoroutinesEventBus add.

build.gradle.kts

@@ -27,11 +27,6 @@ repositories {
 }
 
 dependencies {
-    // https://mvnrepository.com/artifact/net.dv8tion/JDA
-    implementation("net.dv8tion:JDA:5.0.1") {
-        exclude(module = "opus-java")
-    }
-
     // https://mvnrepository.com/artifact/ch.qos.logback/logback-classic
     implementation("ch.qos.logback:logback-classic:1.5.6")
 
@@ -46,6 +41,9 @@ dependencies {
     // https://mvnrepository.com/artifact/io.github.cdimascio/dotenv-kotlin
     implementation("io.github.cdimascio:dotenv-kotlin:6.4.1")
 
+    // https://mvnrepository.com/artifact/io.insert-koin/koin-core
+    implementation("io.insert-koin:koin-core:4.0.0-RC1")
+
     kotlin("stdlib")
 
     listOf(project(":common"), project(":chatbot"), project(":webserver")).forEach {

chatbot/build.gradle.kts

@@ -11,7 +11,7 @@ repositories {
 
 dependencies {
     // https://mvnrepository.com/artifact/net.dv8tion/JDA
-    implementation("net.dv8tion:JDA:5.0.1") {
+    api("net.dv8tion:JDA:5.0.1") {
         exclude(module = "opus-java")
     }
 
@@ -35,6 +35,9 @@ dependencies {
     // https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp
     implementation("com.squareup.okhttp3:okhttp:4.12.0")
 
+    // https://mvnrepository.com/artifact/io.insert-koin/koin-core
+    implementation("io.insert-koin:koin-core:4.0.0-RC1")
+
     testImplementation(kotlin("test"))
 
     listOf(project(":common")).forEach {

chatbot/src/main/kotlin/space/mori/chzzk_bot/chatbot/chzzk/ChzzkHandler.kt

@@ -1,5 +1,9 @@
 package space.mori.chzzk_bot.chatbot.chzzk
 
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import space.mori.chzzk_bot.chatbot.chzzk.Connector.chzzk
@@ -138,11 +142,14 @@ class UserHandler(
             logger.info("ChzzkChat connecting... ${channel.channelName} - ${channel.channelId}")
             listener.connectBlocking()
 
-            Discord.sendDiscord(user, status)
             streamStartTime = LocalDateTime.now()
 
             listener.sendChat("${user.username} 님의 방송이 감지되었습니다.")
 
+            CoroutineScope(Dispatchers.Default).launch {
+                delay(5000L)
+                Discord.sendDiscord(user, status)
+            }
         } else {
             logger.info("${user.username} is offline.")
             streamStartTime = null

chatbot/src/main/kotlin/space/mori/chzzk_bot/chatbot/chzzk/MessageHandler.kt

@@ -1,6 +1,10 @@
 package space.mori.chzzk_bot.chatbot.chzzk
 
-import space.mori.chzzk_bot.common.events.EventDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import org.koin.java.KoinJavaComponent.inject
+import space.mori.chzzk_bot.common.events.CoroutinesEventBus
 import space.mori.chzzk_bot.common.events.TimerEvent
 import space.mori.chzzk_bot.common.events.TimerType
 import space.mori.chzzk_bot.common.models.User
@@ -30,6 +34,8 @@ class MessageHandler(
     private val logger = handler.logger
     private val listener = handler.listener
 
+    private val dispatcher: CoroutinesEventBus by inject(CoroutinesEventBus::class.java)
+
     init {
         reloadCommand()
     }
@@ -118,39 +124,53 @@ class MessageHandler(
             return
         }
 
-        val parts = msg.content.split("/", limit = 2)
+        val parts = msg.content.split(" ", limit = 2)
         if (parts.size < 2) {
             listener.sendChat("타이머 명령어 형식을 잘 찾아봐주세요!")
             return
         }
 
-        val command = parts[1]
+        when (val command = parts[1]) {
-        val dispatcher = EventDispatcher
-        when(command) {
             "업타임" -> {
+                logger.debug("${user.token} / 업타임")
                 val currentTime = LocalDateTime.now()
                 val streamOnTime = handler.streamStartTime
 
-                val hours = ChronoUnit.HOURS.between(currentTime, streamOnTime)
+                val hours = ChronoUnit.HOURS.between(streamOnTime, currentTime)
-                val minutes = ChronoUnit.MINUTES.between(currentTime.plusHours(hours), streamOnTime)
+                val minutes = ChronoUnit.MINUTES.between(streamOnTime?.plusHours(hours), currentTime)
-                val seconds = ChronoUnit.MINUTES.between(currentTime.plusHours(hours).plusMinutes(minutes), streamOnTime)
+                val seconds = ChronoUnit.SECONDS.between(streamOnTime?.plusHours(hours)?.plusMinutes(minutes), currentTime)
 
-                dispatcher.dispatch(TimerEvent(
+                CoroutineScope(Dispatchers.Default).launch {
+                    dispatcher.post(
+                        TimerEvent(
                             user.token,
                             TimerType.TIMER,
                             String.format("%02d:%02d:%02d", hours, minutes, seconds)
-                ))
+                        )
+                    )
+                }
+            }
+            "삭제" -> {
+                logger.debug("${user.token} / 삭제")
+                CoroutineScope(Dispatchers.Default).launch {
+                    dispatcher.post(TimerEvent(user.token, TimerType.REMOVE, ""))
+                }
             }
-            "삭제" -> dispatcher.dispatch(TimerEvent(user.token, TimerType.REMOVE, ""))
             else -> {
+                logger.debug("${user.token} / 그외")
                 try {
                     val time = command.toInt()
                     val currentTime = LocalDateTime.now()
-                    val timestamp = ChronoUnit.MINUTES.addTo(currentTime, time.toLong())
+                    val timestamp = currentTime.plus(time.toLong(), ChronoUnit.MINUTES)
 
-                    dispatcher.dispatch(TimerEvent(user.token, TimerType.TIMER, timestamp.toString()))
+                    CoroutineScope(Dispatchers.Default).launch {
-                } catch(_: Exception) {
+                        dispatcher.post(TimerEvent(user.token, TimerType.TIMER, timestamp.toString()))
+                    }
+                } catch (e: NumberFormatException) {
                     listener.sendChat("!타이머/숫자 형식으로 적어주세요! 단위: 분")
+                } catch (e: Exception) {
+                    listener.sendChat("타이머 설정 중 오류가 발생했습니다.")
+                    logger.error("Error processing timer command: ${e.message}", e)
                 }
             }
         }
@@ -190,7 +210,7 @@ class MessageHandler(
         }
 
         // Replace followPattern
-        result = followPattern.replace(result) { matchResult ->
+        result = followPattern.replace(result) { _ ->
             try {
                 val followingDate = getFollowDate(listener.chatId, msg.userId)
                     .content.streamingProperty.following?.followDate

common/src/main/kotlin/space/mori/chzzk_bot/common/events/EventBase.kt

@@ -1,19 +1,32 @@
 package space.mori.chzzk_bot.common.events
 
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.launch
+import kotlin.reflect.KClass
+
 interface Event
 
-interface EventHandler<E: Event> {
+interface EventBus {
-    fun handle(event: E)
+    suspend fun <T: Event> post(event: T)
+    fun <T: Event> subscribe(eventClass: KClass<T>, listener: (T) -> Unit)
 }
 
-object EventDispatcher {
+class CoroutinesEventBus: EventBus {
-    private val handlers = mutableMapOf<Class<out Event>, MutableList<EventHandler<out Event>>>()
+    private val _events = MutableSharedFlow<Event>()
+    val events: SharedFlow<Event> get() = _events
 
-    fun <E : Event> register(eventClass: Class<E>, handler: EventHandler<E>) {
+    override suspend fun<T: Event> post(event: T) = _events.emit(event)
-        handlers.computeIfAbsent(eventClass) { mutableListOf() }.add(handler)
+
+    override fun <T: Event> subscribe(eventClass: KClass<T>, listener: (T) -> Unit) {
+        CoroutineScope(Dispatchers.Default).launch {
+            events.filterIsInstance(eventClass)
+                .collect {
+                    listener(it)
+                }
         }
-
-    fun <E : Event> dispatch(event: E) {
-        handlers[event::class.java]?.forEach { (it as EventHandler<E>).handle(event) }
     }
 }

common/src/main/kotlin/space/mori/chzzk_bot/common/events/TimerEvents.kt

@@ -8,4 +8,6 @@ class TimerEvent(
     val uid: String,
     val type: TimerType,
     val time: String?
-): Event
+): Event {
+    var TAG = javaClass.simpleName
+}

src/main/kotlin/space/mori/chzzk_bot/Main.kt

@@ -3,12 +3,15 @@ package space.mori.chzzk_bot
 import io.github.cdimascio.dotenv.dotenv
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
+import org.koin.core.context.GlobalContext.startKoin
+import org.koin.dsl.module
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import space.mori.chzzk_bot.chatbot.chzzk.ChzzkHandler
 import space.mori.chzzk_bot.chatbot.discord.Discord
 import space.mori.chzzk_bot.chatbot.chzzk.Connector as ChzzkConnector
 import space.mori.chzzk_bot.common.Connector
+import space.mori.chzzk_bot.common.events.CoroutinesEventBus
 import space.mori.chzzk_bot.webserver.start
 import space.mori.chzzk_bot.webserver.stop
 import java.util.concurrent.TimeUnit
@@ -26,6 +29,13 @@ val chzzkConnector = ChzzkConnector
 val chzzkHandler = ChzzkHandler
 
 fun main(args: Array<String>) {
+    val dispatcher = module {
+        single { CoroutinesEventBus() }
+    }
+    startKoin {
+        modules(dispatcher)
+    }
+
     discord.enable()
     chzzkHandler.enable()
     chzzkHandler.runStreamInfo()

webserver/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
     // https://mvnrepository.com/artifact/org.jetbrains.kotlin/kotlin-reflect
     implementation("org.jetbrains.kotlin:kotlin-reflect:2.0.0")
 
+    // https://mvnrepository.com/artifact/io.insert-koin/koin-core
+    implementation("io.insert-koin:koin-core:4.0.0-RC1")
+
     // https://mvnrepository.com/artifact/ch.qos.logback/logback-classic
     implementation("ch.qos.logback:logback-classic:1.5.6")
 

webserver/src/main/kotlin/space/mori/chzzk_bot/webserver/routes/WSTimerRoutes.kt

@@ -6,15 +6,19 @@ import io.ktor.websocket.*
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
+import kotlinx.coroutines.flow.forEach
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import kotlinx.serialization.Serializable
-import space.mori.chzzk_bot.common.events.Event
+import org.koin.java.KoinJavaComponent.inject
+import org.slf4j.LoggerFactory
+import space.mori.chzzk_bot.common.events.*
 import space.mori.chzzk_bot.common.events.TimerType
 import space.mori.chzzk_bot.common.services.UserService
-import space.mori.chzzk_bot.common.events.EventDispatcher
-import space.mori.chzzk_bot.common.events.EventHandler
 import java.util.concurrent.ConcurrentHashMap
 
+val logger = LoggerFactory.getLogger("WSTimerRoutes")
+
 fun Routing.wsTimerRoutes() {
     val sessions = ConcurrentHashMap<String, WebSocketServerSession>()
 
@@ -51,29 +55,16 @@ fun Routing.wsTimerRoutes() {
         }
     }
 
-    run {
+    val dispatcher: CoroutinesEventBus by inject(CoroutinesEventBus::class.java)
-        val dispatcher = EventDispatcher
 
-        dispatcher.register(TimerEvent::class.java, object : EventHandler<TimerEvent> {
+    dispatcher.subscribe(TimerEvent::class) {
-            override fun handle(event: TimerEvent) {
+        logger.debug("TimerEvent: {} / {}", it.uid, it.type)
-                CoroutineScope(Dispatchers.IO).launch {
+        CoroutineScope(Dispatchers.Default).launch {
-                    sessions[event.uid]?.sendSerialized(TimerResponse(event.type, event.time ?: ""))
+            sessions[it.uid]?.sendSerialized(TimerResponse(it.type, it.time ?: ""))
         }
     }
-        })
-    }
 }
 
-enum class TimerType {
-    UPTIME, TIMER
-}
-
-class TimerEvent(
-    val uid: String,
-    val type: TimerType,
-    val time: String?
-): Event
-
 @Serializable
 data class TimerResponse(
     val type: TimerType,