Fix bugs and networking quality of life improvements (#523)

* Clear suspension on walk and item on item interactions fixes #522

* Fix object collisions overlapping zone boundaries closes #517

* Split GameObjectCollision into two classes closes #474

* Remove try catch in SaveQueue #509 and coroutine context #508

* Remove GameLoop CoroutineScope usage #508

* Add NoCancellable to SaveQueue closes #510

* Replace Events coroutine context with scopes #508

* Replace ActionQueue coroutine context with scopes #508

* Replace SharedFlow with Channel closes #519

engine/src/main/kotlin/world/gregs/voidps/engine/EngineModules.kt

@@ -20,9 +20,7 @@ import world.gregs.voidps.engine.entity.item.drop.DropTables
 import world.gregs.voidps.engine.entity.item.floor.FloorItemTracking
 import world.gregs.voidps.engine.entity.item.floor.FloorItems
 import world.gregs.voidps.engine.entity.obj.GameObjects
-import world.gregs.voidps.engine.map.collision.CollisionStrategyProvider
-import world.gregs.voidps.engine.map.collision.Collisions
-import world.gregs.voidps.engine.map.collision.GameObjectCollision
+import world.gregs.voidps.engine.map.collision.*
 import world.gregs.voidps.engine.map.zone.DynamicZones
 import world.gregs.voidps.network.client.ConnectionQueue
 import world.gregs.voidps.type.Tile
@@ -34,7 +32,7 @@ val engineModule = module {
     // Entities
     single { NPCs(get(), get(), get()) }
     single { Players() }
-    single { GameObjects(get(), get(), get(), getProperty<String>("loadUnusedObjects") == "true").apply { get<ZoneBatchUpdates>().register(this) } }
+    single { GameObjects(get(), get(), get(), get(), getProperty<String>("loadUnusedObjects") == "true").apply { get<ZoneBatchUpdates>().register(this) } }
     single { FloorItems(get(), get()).apply { get<ZoneBatchUpdates>().register(this) } }
     single { FloorItemTracking(get(), get(), get()) }
     single { Hunting(get(), get(), get(), get(), get(), get()) }
@@ -77,7 +75,8 @@ val engineModule = module {
     single {
         ConnectionQueue(getIntProperty("connectionPerTickCap", 1))
     }
-    single(createdAtStart = true) { GameObjectCollision(get()) }
+    single(createdAtStart = true) { GameObjectCollisionAdd(get()) }
+    single(createdAtStart = true) { GameObjectCollisionRemove(get()) }
     // Collision
     single { Collisions() }
     single { CollisionStrategyProvider() }

engine/src/main/kotlin/world/gregs/voidps/engine/GameLoop.kt

@@ -3,34 +3,29 @@ package world.gregs.voidps.engine
 import com.github.michaelbull.logging.InlineLogger
 import kotlinx.coroutines.*
 import java.util.concurrent.TimeUnit
-import kotlin.coroutines.CoroutineContext
 
 class GameLoop(
     private val stages: List<Runnable>,
-    override val coroutineContext: CoroutineContext = Contexts.Game
-) : CoroutineScope {
-
+    private val delay: Long = ENGINE_DELAY
+) {
     private val logger = InlineLogger()
-    private lateinit var job: Job
 
-    fun start() {
+    fun start(scope: CoroutineScope) = scope.launch {
+        var start: Long
+        var took: Long
         try {
-            var start: Long
-            var took: Long
-            job = launch {
-                while (isActive) {
-                    start = System.nanoTime()
-                    tick()
-                    took = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)
-                    if (took > MILLI_THRESHOLD) {
-                        logger.info { "Tick $tick took ${took}ms" }
-                    }
-                    delay(ENGINE_DELAY - took)
-                    tick++
+            while (isActive) {
+                start = System.nanoTime()
+                tick()
+                took = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)
+                if (took > MILLI_THRESHOLD) {
+                    logger.info { "Tick $tick took ${took}ms" }
                 }
+                delay(delay - took)
+                tick++
             }
-        } catch (t: Throwable) {
-            logger.error(t) { "Error in game loop!" }
+        } catch (e: Exception) {
+            logger.error(e) { "Error in game loop!" }
         }
     }
 
@@ -49,10 +44,6 @@ class GameLoop(
         }
     }
 
-    fun stop() {
-        job.cancel()
-    }
-
     companion object {
         var tick: Int = 0
         private const val ENGINE_DELAY = 600L

engine/src/main/kotlin/world/gregs/voidps/engine/client/PlayerAccountLoader.kt

@@ -2,7 +2,7 @@ package world.gregs.voidps.engine.client
 
 import com.github.michaelbull.logging.InlineLogger
 import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.withContext
 import world.gregs.voidps.engine.data.AccountManager
 import world.gregs.voidps.engine.data.AccountStorage
@@ -40,7 +40,7 @@ class PlayerAccountLoader(
     /**
      * @return flow of instructions for the player to be controlled with
      */
-    override suspend fun load(client: Client, username: String, passwordHash: String, displayMode: Int): MutableSharedFlow<Instruction>? {
+    override suspend fun load(client: Client, username: String, passwordHash: String, displayMode: Int): SendChannel<Instruction>? {
         try {
             val saving = saveQueue.saving(username)
             if (saving) {

engine/src/main/kotlin/world/gregs/voidps/engine/client/instruction/InstructionTask.kt

@@ -1,7 +1,14 @@
 package world.gregs.voidps.engine.client.instruction
 
 import com.github.michaelbull.logging.InlineLogger
-import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.produceIn
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import world.gregs.voidps.engine.data.definition.InterfaceDefinitions
 import world.gregs.voidps.engine.data.definition.ItemDefinitions
 import world.gregs.voidps.engine.data.definition.NPCDefinitions
@@ -36,11 +43,10 @@ class InstructionTask(
         handler
     )
 
-    @OptIn(ExperimentalCoroutinesApi::class)
     override fun run() {
         for (player in players) {
-            val instructions = player.instructions
-            for (instruction in instructions.replayCache) {
+            for (i in 0 until MAX_INSTRUCTIONS) {
+                val instruction = player.instructions.tryReceive().getOrNull() ?: break
                 if (player["debug", false]) {
                     logger.debug { "${player.accountName} ${player.tile} - $instruction" }
                 }
@@ -50,7 +56,36 @@ class InstructionTask(
                     logger.error(e) { "Error in instruction $instruction" }
                 }
             }
-            instructions.resetReplayCache()
+        }
+    }
+
+    companion object {
+        const val MAX_INSTRUCTIONS = 20
+
+        fun Flow<Int>.test() {
+
+            flow<List<Int>> {
+                coroutineScope {
+                    val upstreamChannel = buffer(10).produceIn(this)
+                    upstreamChannel.tryReceive().getOrNull()
+                }
+            }
+        }
+
+        @JvmStatic
+        fun main(args: Array<String>): Unit = runBlocking {
+
+            val channel = Channel<Int>(5)
+
+            for (i in 0 until 4) {
+                channel.send(i)
+            }
+            launch {
+                for (i in 0 until 5) {
+                    val it = channel.tryReceive().getOrNull() ?: break
+                    println(it)
+                }
+            }
         }
     }
 }

engine/src/main/kotlin/world/gregs/voidps/engine/client/instruction/handle/InterfaceOnInterfaceOptionHandler.kt

@@ -22,6 +22,8 @@ class InterfaceOnInterfaceOptionHandler(
         val (toId, toComponent, toItem, toInventory) = handler.getInterfaceItem(player, toInterfaceId, toComponentId, toItemId, toSlot) ?: return
 
         player.closeInterfaces()
+        player.queue.clearWeak()
+        player.suspension = null
         player.emit(
             ItemOnItem(
                 fromItem,

engine/src/main/kotlin/world/gregs/voidps/engine/client/instruction/handle/WalkHandler.kt

@@ -16,6 +16,8 @@ class WalkHandler : InstructionHandler<Walk>() {
         }
         player.closeInterfaces()
         player.clearWatch()
+        player.queue.clearWeak()
+        player.suspension = null
         player.walkTo(player.tile.copy(instruction.x, instruction.y))
     }
 

engine/src/main/kotlin/world/gregs/voidps/engine/data/SaveQueue.kt

@@ -1,45 +1,55 @@
 package world.gregs.voidps.engine.data
 
 import com.github.michaelbull.logging.InlineLogger
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.*
 import world.gregs.voidps.engine.client.ui.chat.plural
 import world.gregs.voidps.engine.entity.character.player.Player
+import java.lang.Runnable
 import java.util.concurrent.ConcurrentHashMap
-import kotlin.coroutines.CoroutineContext
 import kotlin.system.measureTimeMillis
 
 class SaveQueue(
     private val storage: AccountStorage,
     private val fallback: AccountStorage = storage,
-    override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : Runnable, CoroutineScope {
+    private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
+) : Runnable {
     private val pending = ConcurrentHashMap<String, PlayerSave>()
     private val logger = InlineLogger()
 
+    private val handler = CoroutineExceptionHandler { _, exception ->
+        logger.error(exception) { "Error saving players!" }
+        scope.fallback(pending.values.toList())
+    }
+    private val fallbackHandler = CoroutineExceptionHandler { _, exception ->
+        logger.error(exception) { "Fallback save failed!" }
+    }
+
     override fun run() {
         if (pending.isEmpty()) {
             return
         }
-        val accounts = pending.values.toList()
-        launch {
-            try {
-                val took = measureTimeMillis {
-                    storage.save(accounts)
-                    for (account in accounts) {
-                        pending.remove(account.name)
-                    }
-                }
-                logger.info { "Saved ${accounts.size} ${"account".plural(accounts.size)} in ${took}ms" }
-            } catch (e: Exception) {
-                logger.error(e) { "Error saving players!" }
-                fallback.save(accounts)
+        scope.save(pending.values.toList())
+    }
+
+    private fun CoroutineScope.save(accounts: List<PlayerSave>) = launch(handler) {
+        val took = measureTimeMillis {
+            withContext(NonCancellable) {
+                storage.save(accounts)
                 for (account in accounts) {
                     pending.remove(account.name)
                 }
             }
         }
+        logger.info { "Saved ${accounts.size} ${"account".plural(accounts.size)} in ${took}ms" }
+    }
+
+    private fun CoroutineScope.fallback(accounts: List<PlayerSave>) = launch(fallbackHandler) {
+        withContext(NonCancellable) {
+            fallback.save(accounts)
+            for (account in accounts) {
+                pending.remove(account.name)
+            }
+        }
     }
 
     fun save(player: Player) {

engine/src/main/kotlin/world/gregs/voidps/engine/data/definition/AreaDefinitions.kt

@@ -17,7 +17,6 @@ class AreaDefinitions(
     private var areas: Map<Int, Set<AreaDefinition>> = Int2ObjectOpenHashMap()
 ) {
 
-
     fun getOrNull(name: String): AreaDefinition? {
         return named[name]
     }

engine/src/main/kotlin/world/gregs/voidps/engine/entity/character/player/Player.kt

@@ -1,7 +1,8 @@
 package world.gregs.voidps.engine.entity.character.player
 
-import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.channels.Channel
 import org.rsmod.game.pathfinder.collision.CollisionStrategy
+import world.gregs.voidps.engine.client.instruction.InstructionTask
 import world.gregs.voidps.engine.client.ui.InterfaceOptions
 import world.gregs.voidps.engine.client.ui.Interfaces
 import world.gregs.voidps.engine.client.update.view.Viewport
@@ -53,7 +54,7 @@ class Player(
         }
 
     override lateinit var visuals: PlayerVisuals
-    val instructions = MutableSharedFlow<Instruction>(replay = 20)
+    val instructions = Channel<Instruction>(capacity = InstructionTask.MAX_INSTRUCTIONS)
     lateinit var options: PlayerOptions
     lateinit var interfaces: Interfaces
     lateinit var interfaceOptions: InterfaceOptions

engine/src/main/kotlin/world/gregs/voidps/engine/entity/obj/GameObjects.kt

@@ -9,7 +9,8 @@ import world.gregs.voidps.engine.entity.Despawn
 import world.gregs.voidps.engine.entity.Spawn
 import world.gregs.voidps.engine.entity.character.player.Player
 import world.gregs.voidps.engine.get
-import world.gregs.voidps.engine.map.collision.GameObjectCollision
+import world.gregs.voidps.engine.map.collision.GameObjectCollisionAdd
+import world.gregs.voidps.engine.map.collision.GameObjectCollisionRemove
 import world.gregs.voidps.network.login.protocol.encode.send
 import world.gregs.voidps.network.login.protocol.encode.zone.ObjectAddition
 import world.gregs.voidps.network.login.protocol.encode.zone.ObjectRemoval
@@ -25,7 +26,8 @@ import world.gregs.voidps.type.Zone
  * @param storeUnused store non-interactive and objects without configs for debugging and content dev (uses ~240MB more ram).
  */
 class GameObjects(
-    private val collisions: GameObjectCollision,
+    private val collisionAdd: GameObjectCollisionAdd,
+    private val collisionRemove: GameObjectCollisionRemove,
     private val batches: ZoneBatchUpdates,
     private val definitions: ObjectDefinitions,
     private val storeUnused: Boolean = false
@@ -62,7 +64,7 @@ class GameObjects(
             map.remove(obj, REPLACED)
             batches.add(obj.tile.zone, ObjectAddition(obj.tile.id, obj.intId, obj.shape, obj.rotation))
             if (collision) {
-                collisions.modify(obj, add = true)
+                collisionAdd.modify(obj)
             }
             size++
         } else {
@@ -83,7 +85,7 @@ class GameObjects(
             replacements[obj.index] = obj.value(replaced = true)
             batches.add(obj.tile.zone, ObjectAddition(obj.tile.id, obj.intId, obj.shape, obj.rotation))
             if (collision) {
-                collisions.modify(obj, add = true)
+                collisionAdd.modify(obj)
             }
             size++
             obj.emit(Spawn)
@@ -94,7 +96,7 @@ class GameObjects(
         val gameObject = GameObject(id(objectValue), obj.x, obj.y, obj.level, shape(objectValue), rotation(objectValue))
         batches.add(obj.tile.zone, ObjectRemoval(obj.tile.id, gameObject.shape, gameObject.rotation))
         if (collision) {
-            collisions.modify(gameObject, add = false)
+            collisionRemove.modify(gameObject)
         }
         size--
         return gameObject
@@ -104,7 +106,7 @@ class GameObjects(
      * Sets the original placement of a game object
      */
     fun set(id: Int, x: Int, y: Int, level: Int, shape: Int, rotation: Int, definition: ObjectDefinition) {
-        collisions.modify(definition, x, y, level, shape, rotation, add = true)
+        collisionAdd.modify(definition, x, y, level, shape, rotation)
         if (interactive(definition)) {
             map[x, y, level, ObjectLayer.layer(shape)] = value(false, id, shape, rotation)
             size++
@@ -143,7 +145,7 @@ class GameObjects(
             replacements.remove(obj.index)
             batches.add(obj.tile.zone, ObjectRemoval(obj.tile.id, obj.shape, obj.rotation))
             if (collision) {
-                collisions.modify(obj, add = false)
+                collisionRemove.modify(obj)
             }
             size--
             obj.emit(Despawn)
@@ -153,7 +155,7 @@ class GameObjects(
                 val originalObj = GameObject(id(original), obj.x, obj.y, obj.level, shape(original), rotation(original))
                 batches.add(obj.tile.zone, ObjectAddition(obj.tile.id, originalObj.intId, originalObj.shape, originalObj.rotation))
                 if (collision) {
-                    collisions.modify(originalObj, add = true)
+                    collisionAdd.modify(originalObj)
                 }
                 size++
             }
@@ -162,7 +164,7 @@ class GameObjects(
             map.add(obj, REPLACED)
             batches.add(obj.tile.zone, ObjectRemoval(obj.tile.id, obj.shape, obj.rotation))
             if (collision) {
-                collisions.modify(obj, add = false)
+                collisionRemove.modify(obj)
             }
             size--
         }