Convert bukkit async scheduler to coroutine

Rothes · Rothes · commit 64e56eb15903 · 2025-11-19T03:49:50.000+08:00
diff --git a/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/ChatAntiSpamModule.kt b/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/ChatAntiSpamModule.kt
@@ -12,14 +12,13 @@ import io.github.rothes.esu.bukkit.user
 import io.github.rothes.esu.bukkit.user.ConsoleUser
 import io.github.rothes.esu.bukkit.user.PlayerUser
 import io.github.rothes.esu.bukkit.util.ComponentBukkitUtils.user
-import io.github.rothes.esu.bukkit.util.scheduler.ScheduledTask
-import io.github.rothes.esu.bukkit.util.scheduler.Scheduler
 import io.github.rothes.esu.bukkit.util.version.adapter.PlayerAdapter.Companion.displayName_
 import io.github.rothes.esu.core.configuration.ConfigurationPart
 import io.github.rothes.esu.core.configuration.data.MessageData
 import io.github.rothes.esu.core.configuration.data.MessageData.Companion.message
 import io.github.rothes.esu.core.configuration.meta.Comment
 import io.github.rothes.esu.core.configuration.serializer.MapSerializer.DefaultedEnumMap
+import io.github.rothes.esu.core.coroutine.AsyncScope
 import io.github.rothes.esu.core.module.configuration.BaseModuleConfiguration
 import io.github.rothes.esu.core.user.User
 import io.github.rothes.esu.core.util.ComponentUtils.component
@@ -28,6 +27,10 @@ import io.github.rothes.esu.core.util.ComponentUtils.parsed
 import io.github.rothes.esu.core.util.ComponentUtils.unparsed
 import io.github.rothes.esu.lib.adventure.text.minimessage.tag.resolver.TagResolver
 import io.github.rothes.esu.lib.configurate.objectmapping.meta.PostProcess
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
 import org.bukkit.Bukkit
 import org.incendo.cloud.component.DefaultValue
 import java.time.Duration
@@ -40,11 +43,16 @@ import kotlin.time.toJavaDuration
 
 object ChatAntiSpamModule: BukkitModule<ChatAntiSpamModule.ModuleConfig, ChatAntiSpamModule.ModuleLocale>() {
 
-    private var purgeTask: ScheduledTask? = null
+    private var purgeTask: Job? = null
 
     override fun onEnable() {
         CasDataManager
-        purgeTask = Scheduler.asyncTicks(20, 5 * 60 * 20) { CasDataManager.purgeCache(true) }
+        purgeTask = AsyncScope.launch {
+            while (isActive) {
+                CasDataManager.purgeCache(true)
+                delay(5.minutes)
+            }
+        }
         CasListeners.enable()
         Bukkit.getOnlinePlayers().map { it.user }.forEach {
             if (it.hasPerm("notify"))
diff --git a/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/CommandAntiSpamModule.kt b/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/CommandAntiSpamModule.kt
@@ -4,40 +4,47 @@ import com.google.common.collect.HashBasedTable
 import io.github.rothes.esu.bukkit.user
 import io.github.rothes.esu.bukkit.util.extension.ListenerExt.register
 import io.github.rothes.esu.bukkit.util.extension.ListenerExt.unregister
-import io.github.rothes.esu.bukkit.util.scheduler.ScheduledTask
-import io.github.rothes.esu.bukkit.util.scheduler.Scheduler
 import io.github.rothes.esu.core.configuration.ConfigurationPart
 import io.github.rothes.esu.core.configuration.meta.Comment
+import io.github.rothes.esu.core.coroutine.AsyncScope
 import io.github.rothes.esu.core.module.configuration.BaseModuleConfiguration
 import io.github.rothes.esu.core.user.User
 import io.github.rothes.esu.core.util.extension.DurationExt.compareTo
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
 import org.bukkit.event.EventHandler
 import org.bukkit.event.EventPriority
 import org.bukkit.event.Listener
 import org.bukkit.event.player.PlayerCommandPreprocessEvent
 import java.time.Duration
+import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
 
 object CommandAntiSpamModule: BukkitModule<CommandAntiSpamModule.ModuleConfig, CommandAntiSpamModule.ModuleLocale>() {
     
-    private var cacheTask: ScheduledTask? = null
+    private var cacheTask: Job? = null
 
     override fun onEnable() {
         Listeners.register()
-        cacheTask = Scheduler.asyncTicks(5 * 60 * 20L, 5 * 60 * 20L) {
-            val now = System.currentTimeMillis()
-            Listeners.hits.cellSet().toList().forEach { cell ->
-                val cmd = cell.columnKey
-                val queue = cell.value
-                val conf = config.commands.find { it.commands.any { regex -> regex.containsMatchIn(cmd) } }
-                if (conf == null) {
-                    queue.clear()
-                } else {
-                    queue.removeIf { now - it > conf.expireInterval }
-                }
-                if (queue.isEmpty()) {
-                    Listeners.hits.remove(cell.rowKey, cell.columnKey)
+        cacheTask = AsyncScope.launch {
+            while (isActive) {
+                delay(5.minutes)
+                val now = System.currentTimeMillis()
+                Listeners.hits.cellSet().toList().forEach { cell ->
+                    val cmd = cell.columnKey
+                    val queue = cell.value
+                    val conf = config.commands.find { it.commands.any { regex -> regex.containsMatchIn(cmd) } }
+                    if (conf == null) {
+                        queue.clear()
+                    } else {
+                        queue.removeIf { now - it > conf.expireInterval }
+                    }
+                    if (queue.isEmpty()) {
+                        Listeners.hits.remove(cell.rowKey, cell.columnKey)
+                    }
                 }
             }
         }
diff --git a/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/exploitfix/VaultUnlocking.kt b/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/exploitfix/VaultUnlocking.kt
@@ -7,15 +7,18 @@ import io.github.rothes.esu.bukkit.util.CoordinateUtils.blockKey_
 import io.github.rothes.esu.bukkit.util.ServerCompatibility
 import io.github.rothes.esu.bukkit.util.extension.ListenerExt.register
 import io.github.rothes.esu.bukkit.util.extension.ListenerExt.unregister
-import io.github.rothes.esu.bukkit.util.scheduler.ScheduledTask
-import io.github.rothes.esu.bukkit.util.scheduler.Scheduler
+import io.github.rothes.esu.core.coroutine.AsyncScope
 import io.github.rothes.esu.core.util.CollectionUtils.removeWhile
 import io.github.rothes.esu.core.util.ComponentUtils.duration
 import io.github.rothes.esu.core.util.version.Version
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap
 import it.unimi.dsi.fastutil.longs.LongArrayList
 import it.unimi.dsi.fastutil.longs.LongList
 import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
 import org.bukkit.Material
 import org.bukkit.World
 import org.bukkit.block.Block
@@ -35,7 +38,7 @@ internal object VaultUnlocking: Listener {
 
     private var enabled = AtomicBoolean(false)
     private val opened = Object2ObjectOpenHashMap<World, Long2ObjectOpenHashMap<LongList>>()
-    private var purgeTask: ScheduledTask? = null
+    private var purgeTask: Job? = null
 
     private fun getBlockTimes(block: Block) =
         opened.getOrPut(block.world) { Long2ObjectOpenHashMap() }.getOrPut(block.blockKey_) { LongArrayList() }
@@ -48,21 +51,24 @@ internal object VaultUnlocking: Listener {
             }
             if (enabled.compareAndSet(expectedValue = false, newValue = true)) {
                 this.register()
-                purgeTask = Scheduler.async(30.minutes, 30.minutes) {
-                    val now = System.currentTimeMillis()
-                    val config = ExploitFixesModule.config.vaultUnlocking
-                    val expired = now - config.unlockExpiry.toMillis()
-                    val iterator = opened.iterator()
-                    for ((_, map) in iterator) {
-                        val iterator2 = map.iterator()
-                        for ((_, list) in iterator2) {
-                            list.removeWhile { it < expired }
-                            if (list.isEmpty()) {
-                                iterator2.remove()
+                purgeTask = AsyncScope.launch {
+                    while (isActive) {
+                        delay(30.minutes)
+                        val now = System.currentTimeMillis()
+                        val config = ExploitFixesModule.config.vaultUnlocking
+                        val expired = now - config.unlockExpiry.toMillis()
+                        val iterator = opened.iterator()
+                        for ((_, map) in iterator) {
+                            val iterator2 = map.iterator()
+                            for ((_, list) in iterator2) {
+                                list.removeWhile { it < expired }
+                                if (list.isEmpty()) {
+                                    iterator2.remove()
+                                }
+                            }
+                            if (map.isEmpty()) {
+                                iterator.remove()
                             }
-                        }
-                        if (map.isEmpty()) {
-                            iterator.remove()
                         }
                     }
                 }
diff --git a/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/networkthrottle/HighLatencyAdjust.kt b/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/networkthrottle/HighLatencyAdjust.kt
@@ -13,22 +13,26 @@ import io.github.rothes.esu.bukkit.util.ServerCompatibility
 import io.github.rothes.esu.bukkit.util.extension.ListenerExt.register
 import io.github.rothes.esu.bukkit.util.extension.ListenerExt.unregister
 import io.github.rothes.esu.bukkit.util.extension.checkPacketEvents
-import io.github.rothes.esu.bukkit.util.scheduler.ScheduledTask
-import io.github.rothes.esu.bukkit.util.scheduler.Scheduler
 import io.github.rothes.esu.core.configuration.ConfigurationPart
 import io.github.rothes.esu.core.configuration.data.MessageData
 import io.github.rothes.esu.core.configuration.data.MessageData.Companion.message
 import io.github.rothes.esu.core.configuration.meta.Comment
+import io.github.rothes.esu.core.coroutine.AsyncScope
 import io.github.rothes.esu.core.module.CommonFeature
 import io.github.rothes.esu.core.module.Feature
 import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
 import org.bukkit.Bukkit
 import org.bukkit.entity.Player
 import org.bukkit.event.EventHandler
 import org.bukkit.event.Listener
 import org.bukkit.event.player.PlayerQuitEvent
 import java.time.Duration
 import kotlin.math.min
+import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
 
 object HighLatencyAdjust: CommonFeature<HighLatencyAdjust.FeatureConfig, HighLatencyAdjust.FeatureLang>(), Listener {
@@ -37,7 +41,7 @@ object HighLatencyAdjust: CommonFeature<HighLatencyAdjust.FeatureConfig, HighLat
 
     val adjusted = hashMapOf<Player, Int>()
     val startTime = Object2LongOpenHashMap<Player>().also { it.defaultReturnValue(NO_TIME) }
-    var task: ScheduledTask? = null
+    var task: Job? = null
 
     @EventHandler
     fun onQuit(e: PlayerQuitEvent) {
@@ -63,32 +67,35 @@ object HighLatencyAdjust: CommonFeature<HighLatencyAdjust.FeatureConfig, HighLat
         }
         data.originalViewDistance.clear()
 
-        task = Scheduler.asyncTicks(0, 15 * 20) {
-            val now = System.currentTimeMillis()
-            for (player in Bukkit.getOnlinePlayers()) {
-                if (player.ping >= config.latencyThreshold) {
-                    val last = startTime.getLong(player)
-                    if (last == NO_TIME) {
-                        startTime[player] = now
-                        continue
-                    } else if ((now - last) < config.duration.toMillis()) {
-                        continue
-                    }
-                    startTime.removeLong(player)
-
-                    if (!adjusted.containsKey(player)) {
-                        adjusted[player] = player.clientViewDistance
-                        player.sendViewDistance = min(player.clientViewDistance, player.viewDistance) - 1
-                    } else {
-                        if (player.sendViewDistance <= config.minViewDistance) {
+        task = AsyncScope.launch {
+            while (isActive) {
+                val now = System.currentTimeMillis()
+                for (player in Bukkit.getOnlinePlayers()) {
+                    if (player.ping >= config.latencyThreshold) {
+                        val last = startTime.getLong(player)
+                        if (last == NO_TIME) {
+                            startTime[player] = now
+                            continue
+                        } else if ((now - last) < config.duration.toMillis()) {
                             continue
                         }
-                        player.sendViewDistance--
+                        startTime.removeLong(player)
+
+                        if (!adjusted.containsKey(player)) {
+                            adjusted[player] = player.clientViewDistance
+                            player.sendViewDistance = min(player.clientViewDistance, player.viewDistance) - 1
+                        } else {
+                            if (player.sendViewDistance <= config.minViewDistance) {
+                                continue
+                            }
+                            player.sendViewDistance--
+                        }
+                        player.user.message(lang, { adjustedWarning })
+                    } else {
+                        startTime.removeLong(player)
                     }
-                    player.user.message(lang, { adjustedWarning })
-                } else {
-                    startTime.removeLong(player)
                 }
+                delay(15.seconds)
             }
         }
         PacketEvents.getAPI().eventManager.registerListener(PacketListeners)
diff --git a/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/news/NewsDataManager.kt b/bukkit/src/main/kotlin/io/github/rothes/esu/bukkit/module/news/NewsDataManager.kt
@@ -1,21 +1,21 @@
 package io.github.rothes.esu.bukkit.module.news
 
 import io.github.rothes.esu.bukkit.module.NewsModule
-import io.github.rothes.esu.bukkit.util.scheduler.ScheduledTask
-import io.github.rothes.esu.bukkit.util.scheduler.Scheduler
 import io.github.rothes.esu.core.storage.StorageManager
 import io.github.rothes.esu.core.storage.StorageManager.database
 import io.github.rothes.esu.core.storage.StorageManager.upgrader
 import io.github.rothes.esu.core.user.User
 import io.github.rothes.esu.core.util.DataSerializer.deserialize
 import io.github.rothes.esu.core.util.DataSerializer.serialize
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.*
 import kotlinx.datetime.LocalDateTime
 import org.jetbrains.exposed.v1.core.*
 import org.jetbrains.exposed.v1.datetime.datetime
 import org.jetbrains.exposed.v1.jdbc.*
 import org.jetbrains.exposed.v1.jdbc.transactions.transaction
 import org.jetbrains.exposed.v1.json.json
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
 
 object NewsDataManager {
 
@@ -36,7 +36,7 @@ object NewsDataManager {
         override val primaryKey: PrimaryKey = PrimaryKey(user, channel)
     }
 
-    private var task: ScheduledTask? = null
+    private var task: Job? = null
     var news: List<NewsItem> = emptyList()
         private set
 
@@ -57,15 +57,12 @@ object NewsDataManager {
 
     fun start() {
         task?.cancel()
-        fetchNews()
-        fun schedule() {
-            // We want a random offset
-            task = Scheduler.asyncTicks(20 * 60L + (-200 .. 200).random()) {
+        task = CoroutineScope(Dispatchers.IO).launch {
+            while (isActive) {
                 fetchNews()
-                schedule()
+                delay(1.minutes + (1 .. 10).random().seconds) // Random offset to stagger the queries
             }
         }
-        schedule()
     }
 
     fun shutdown() {
diff --git a/core/src/main/kotlin/io/github/rothes/esu/core/coroutine/Scopes.kt b/core/src/main/kotlin/io/github/rothes/esu/core/coroutine/Scopes.kt
@@ -0,0 +1,11 @@
+package io.github.rothes.esu.core.coroutine
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlin.coroutines.CoroutineContext
+
+object AsyncScope : CoroutineScope {
+
+    override val coroutineContext: CoroutineContext = Dispatchers.Default
+
+}
