Proper cancellation handling and minor fixes

Author: Avanatiker

common/src/main/kotlin/com/lambda/Lambda.kt

@@ -1,5 +1,6 @@
 package com.lambda
 
+import com.lambda.event.EventFlow
 import com.lambda.event.listener.SafeListener.Companion.listener
 import com.lambda.event.events.PacketEvent
 import com.lambda.event.listener.Listener.Companion.unsubscribe
@@ -24,17 +25,40 @@ object Lambda {
     init {
         listener<PacketEvent.Send.Pre> {
             if (it.packet is LookAndOnGround) {
-                it.cancel()
+//                it.cancel()
                 LOG.info("SAFE: Canceled: ${it.packet::class.simpleName}")
             }
         }
 
+        listener<PacketEvent.Send.Pre> {
+            if (it.packet is LookAndOnGround) {
+//                it.cancel()
+                LOG.info("2 SAFE: Canceled: ${it.packet::class.simpleName}")
+            }
+        }
+
+        unsafeListener<PacketEvent.Send.Pre> {
+            if (it.packet is LookAndOnGround) {
+//                it.cancel()
+                LOG.info("UNSAFE: Canceled: ${it.packet::class.simpleName}")
+            }
+        }
+
         concurrentListener<PacketEvent.Send.Pre> {
             if (it.packet is LookAndOnGround) {
                 LOG.info("CONCURRENT: ${it.packet::class.simpleName}")
             }
         }
 
+        unsafeConcurrentListener<PacketEvent.Send.Pre> {
+            if (it.packet is LookAndOnGround) {
+                LOG.info("UNSAFE CONCURRENT: ${it.packet::class.simpleName}")
+            }
+        }
+
+        LOG.info("Registered sync listeners: ${EventFlow.syncListeners}")
+        LOG.info("Registered concurrent listeners: ${EventFlow.concurrentListeners}")
+
         runConcurrent {
             sleep(60000)
             LOG.info("Unsubscribing")

common/src/main/kotlin/com/lambda/event/CallbackEvent.kt

@@ -0,0 +1,3 @@
+package com.lambda.event
+
+interface CallbackEvent : Event

common/src/main/kotlin/com/lambda/event/EventFlow.kt

@@ -1,13 +1,12 @@
 package com.lambda.event
 
-import com.lambda.Lambda.LOG
-import com.lambda.Lambda.mc
 import com.lambda.event.listener.Listener
 import com.lambda.runConcurrent
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.filterNot
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentSkipListSet
 import kotlin.reflect.KClass
@@ -38,6 +37,7 @@ object EventFlow {
             concurrentFlow
                 // early filter to avoid an unnecessary collection
                 .filter { concurrentListeners.containsKey(it::class) }
+                .filterNot { it is ICancellable && it.isCanceled() }
                 .collect { event ->
                     event.executeListenerConcurrently()
                 }
@@ -51,7 +51,8 @@ object EventFlow {
      * Each asynchronous listener will execute its listener function on a new thread.
      *
      * After notifying asynchronous listeners, it executes the listener functions of all synchronous listeners.
-     * An instant callback can only be achieved by synchronous listening objects as the concurrently listener will be executed "later".
+     * An instant callback can only be achieved by synchronous listening objects
+     * as the concurrent listener will be executed "later".
      *
      * @param event The event to be posted to the event flow.
      */
@@ -63,12 +64,14 @@ object EventFlow {
 
     private fun Event.executeListenerSynchronous() {
         syncListeners[this::class]?.forEach { listener ->
+            if (this is ICancellable && this.isCanceled()) return
             listener.execute(this@executeListenerSynchronous)
         }
     }
 
     private fun Event.executeListenerConcurrently() {
         concurrentListeners[this::class]?.forEach { listener ->
+            if (this is ICancellable && this.isCanceled()) return
             runConcurrent {
                 listener.execute(this@executeListenerConcurrently)
             }

common/src/main/kotlin/com/lambda/event/ICancellable.kt

@@ -2,7 +2,7 @@ package com.lambda.event
 
 import java.util.concurrent.atomic.AtomicBoolean
 
-interface ICancellable : MutableEvent {
+interface ICancellable : CallbackEvent {
     val cancelSignal: AtomicBoolean
 
     fun cancel() {

common/src/main/kotlin/com/lambda/event/MutableEvent.kt

@@ -1,5 +1,6 @@
 package com.lambda.event.events
 
-import com.lambda.event.MutableEvent
+import com.lambda.event.Cancellable
+import com.lambda.event.ICancellable
 
-class KeyPressEvent(val key: Int): MutableEvent
+class KeyPressEvent(val key: Int): ICancellable by Cancellable()

common/src/main/kotlin/com/lambda/event/events/KeyPressEvent.kt

@@ -6,7 +6,7 @@ import com.lambda.event.EventFlow
 import com.lambda.runSafe
 import java.util.concurrent.ConcurrentSkipListSet
 
-data class SafeListener(
+class SafeListener(
     override val priority: Int = 0,
     val function: SafeContext.(Event) -> Unit
 ) : Listener() {

common/src/main/kotlin/com/lambda/event/listener/SafeListener.kt

@@ -4,7 +4,7 @@ import com.lambda.event.Event
 import com.lambda.event.EventFlow
 import java.util.concurrent.ConcurrentSkipListSet
 
-data class UnsafeListener(
+class UnsafeListener(
     override val priority: Int,
     val function: (Event) -> Unit
 ) : Listener() {
@@ -25,7 +25,7 @@ data class UnsafeListener(
          * The [function] is executed on the same thread where the [Event] was dispatched.
          * The execution of the [function] is independent of the safety conditions of the context.
          * Use this function when you need to listen to an [Event] in a context that is not in-game.
-         * For only in-game related contexts, use the [listener] function instead.
+         * For only in-game related contexts, use the [SafeListener.listener] function instead.
          *
          * Usage:
          * ```kotlin
@@ -55,6 +55,7 @@ data class UnsafeListener(
          * Registers a new [UnsafeListener] for a generic [Event] type [T].
          * The [function] is executed on a new thread running asynchronously to the game thread.
          * This function should only be used when the [function] performs read actions on the game data.
+         * For only in-game related contexts, use the [SafeListener.concurrentListener] function instead.
          *
          * Caution: Using this function to write to the game data can lead to race conditions. Therefore, it is recommended
          * to use this function only for read operations to avoid potential concurrency issues.