Refactor event listener management in Task and EventFlow

Removed redundant listener management methods and replaced them with a streamlined `unsubscribe` approach. Cleaned up code to improve maintainability and consistency across Task state transitions, ensuring proper cleanup of listeners tied to task instances.

Author: Avanatiker

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

@@ -87,6 +87,11 @@ object EventFlow {
      */
     val concurrentListeners = Subscriber()
 
+    fun Any.unsubscribe() {
+        syncListeners.unsubscribe(this)
+        concurrentListeners.unsubscribe(this)
+    }
+
     init {
         // parallel event execution on dedicated threads
         runConcurrent {

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

@@ -54,6 +54,18 @@ class Subscriber : ConcurrentHashMap<KClass<out Event>, ConcurrentSkipListSet<Li
     inline fun <reified T : Event> unsubscribe(listener: Listener<T>) =
         getOrElse(T::class) { defaultListenerSet }.remove(listener)
 
+    /**
+     * Unsubscribes all listeners associated with the current instance (the caller object).
+     * This method iterates over all values in the `Subscriber`'s map and removes listeners
+     * whose `owner` property matches the caller object.
+     *
+     * Use this method when you want to clean up listeners that were registered with the
+     * current instance, preventing further event notifications.
+     */
+    fun unsubscribe(owner: Any) {
+        values.forEach { it.removeAll { listener -> listener.owner == owner } }
+    }
+
     /** Allows a [Subscriber] to stop receiving all [Event]s of another [Subscriber] */
     infix fun unsubscribe(subscriber: Subscriber) {
         subscriber.forEach { (eventType, listeners) ->

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

@@ -130,51 +130,6 @@ class SafeListener<T : Event>(
             return listener
         }
 
-        /**
-         * Registers a new [SafeListener] for a generic [Event] type [T] within the context of a [Task].
-         * The [function] is executed on the same thread where the [Event] was dispatched.
-         * The [function] will only be executed when the context satisfies certain safety conditions.
-         * These conditions are met when none of the following [SafeContext] properties are null:
-         * - [SafeContext.world]
-         * - [SafeContext.player]
-         * - [SafeContext.interaction]
-         * - [SafeContext.connection]
-         *
-         * Usage:
-         * ```kotlin
-         * myTask.listen<MyEvent> { event ->
-         *     player.sendMessage("Event received: $event")
-         * }
-         *
-         * myTask.listen<MyEvent>(priority = 1) { event ->
-         *     player.sendMessage("Event received before the previous listener: $event")
-         * }
-         * ```
-         *
-         * @param T The type of the event to listen for.
-         * This should be a subclass of Event.
-         * @param priority The priority of the listener.
-         * Listeners with higher priority will be executed first.
-         * The Default value is 0.
-         * @param alwaysListen If true, the listener will be executed even if it is muted. The Default value is false.
-         * @param function The function to be executed when the event is posted.
-         * This function should take a SafeContext and an event of type T as parameters.
-         * @return The newly created and registered [SafeListener].
-         */
-        inline fun <reified T : Event> Task<*>.listen(
-            priority: Int = 0,
-            alwaysListen: Boolean = false,
-            noinline function: SafeContext.(T) -> Unit = {},
-        ): SafeListener<T> {
-            val listener = SafeListener<T>(priority, this, alwaysListen) { event ->
-                function(event) // ToDo: run function always on game thread
-            }
-
-            syncListeners.subscribe<T>(listener)
-
-            return listener
-        }
-
         /**
          * This function registers a new [SafeListener] for a generic [Event] type [T].
          * The [transform] is executed on the same thread where the [Event] was dispatched.

common/src/main/kotlin/com/lambda/task/Task.kt

@@ -19,8 +19,8 @@ package com.lambda.task
 
 import com.lambda.Lambda.LOG
 import com.lambda.context.SafeContext
-import com.lambda.event.EventFlow
-import com.lambda.event.Subscriber
+import com.lambda.event.EventFlow.unsubscribe
+import com.lambda.event.Muteable
 import com.lambda.event.events.TickEvent
 import com.lambda.event.listener.SafeListener.Companion.listen
 import com.lambda.module.modules.client.TaskFlowModule
@@ -33,10 +33,11 @@ typealias TaskGenerator<R> = SafeContext.(R) -> Task<*>
 typealias TaskGeneratorOrNull<R> = SafeContext.(R) -> Task<*>?
 typealias TaskGeneratorUnit<R> = SafeContext.(R) -> Unit
 
-abstract class Task<Result> : Nameable {
+abstract class Task<Result> : Nameable, Muteable {
     private var parent: Task<*>? = null
     private val subTasks = mutableListOf<Task<*>>()
-    private var state = State.RUNNING
+    private var state = State.INIT
+    override val isMuted: Boolean get() = state == State.PAUSED || state == State.INIT
     var age = 0
     private val depth: Int get() = parent?.depth?.plus(1) ?: 0
     val isCompleted get() = state == State.COMPLETED
@@ -48,10 +49,8 @@ abstract class Task<Result> : Nameable {
     private var nextTaskOrNull: TaskGeneratorOrNull<Result>? = null
     private var onFinish: TaskGeneratorUnit<Result>? = null
 
-    val syncListeners = Subscriber()
-    private val concurrentListeners = Subscriber()
-
     enum class State {
+        INIT,
         RUNNING,
         PAUSED,
         CANCELLED,
@@ -118,14 +117,14 @@ abstract class Task<Result> : Nameable {
             LOG.info("$name deactivating parent ${owner.name}")
             if (owner !is TaskFlow) owner.deactivate()
         }
+        state = State.RUNNING
         runSafe { runCatching { onStart() }.onFailure { failure(it) } }
-        startListening()
         return this
     }
 
     @Ta5kBuilder
     fun success(result: Result) {
-        stopListening()
+        unsubscribe()
         state = State.COMPLETED
         runSafe {
             executeNextTask(result)
@@ -141,15 +140,13 @@ abstract class Task<Result> : Nameable {
     fun activate() {
         if (state != State.PAUSED) return
         state = State.RUNNING
-        startListening()
     }
 
     @Ta5kBuilder
     fun deactivate() {
         if (state != State.RUNNING) return
         if (unpausable) return
         state = State.PAUSED
-        stopListening()
     }
 
     private fun SafeContext.executeNextTask(result: Result) {
@@ -174,7 +171,7 @@ abstract class Task<Result> : Nameable {
         if (this is TaskFlow) return
         if (state == State.COMPLETED || state == State.CANCELLED) return
         state = State.CANCELLED
-        stopListening()
+        unsubscribe()
     }
 
     @Ta5kBuilder
@@ -198,7 +195,7 @@ abstract class Task<Result> : Nameable {
         stacktrace: MutableList<Task<*>> = mutableListOf()
     ) {
         state = State.FAILED
-        stopListening()
+        unsubscribe()
         stacktrace.add(this)
         parent?.failure(e, stacktrace) ?: run {
             val message = buildString {
@@ -214,16 +211,6 @@ abstract class Task<Result> : Nameable {
         }
     }
 
-    private fun startListening() {
-        EventFlow.syncListeners.subscribe(syncListeners)
-        EventFlow.concurrentListeners.subscribe(concurrentListeners)
-    }
-
-    private fun stopListening() {
-        EventFlow.syncListeners.unsubscribe(syncListeners)
-        EventFlow.concurrentListeners.unsubscribe(concurrentListeners)
-    }
-
     /**
      * Specifies the next task to execute after the current task completes successfully.
      *