almost done, just need to make thread safe, test, and document

5 files changed, 110 insertions, 46 deletions

diff --git a/src/main/kotlin/me/bush/illnamethislater/CancelledState.kt b/src/main/kotlin/me/bush/illnamethislater/CancelledState.kt
index ed2263a..87846c7 100644
--- a/src/main/kotlin/me/bush/illnamethislater/CancelledState.kt
+++ b/src/main/kotlin/me/bush/illnamethislater/CancelledState.kt
@@ -1,7 +1,9 @@
 package me.bush.illnamethislater
 
 import sun.misc.Unsafe
+import java.lang.reflect.Modifier
 import kotlin.reflect.KClass
+import kotlin.reflect.KMutableProperty
 import kotlin.reflect.KMutableProperty1
 import kotlin.reflect.full.declaredMembers
 import kotlin.reflect.full.isSubclassOf
@@ -22,8 +24,6 @@ internal fun interface CancelledState {
      */
     fun isCancelled(event: Any): Boolean
 
-    // Maybe move this to eventbus or util? todo
-    // Make CancelledState a class? todo
     companion object {
         private val UNSAFE = Unsafe::class.declaredMembers.single { it.name == "theUnsafe" }.handleCall() as Unsafe
         private val CANCELLED_NAMES = arrayOf("canceled", "cancelled")
@@ -33,19 +33,21 @@ internal fun interface CancelledState {
         /**
          * Creates a [CancelledState] object for events of class [type].
          */
-        // TODO: 3/31/2022 static/singleton fields
         fun of(type: KClass<*>, config: Config): CancelledState {
-            // Default impl for our event class
+            // Default impl for our event class.
             if (type.isSubclassOf(Event::class)) return CancelledState { (it as Event).cancelled }
-            // If compat is disabled
+            // If compat is disabled.
             if (!config.thirdPartyCompatibility) return NOT_CANCELLABLE
-            // Find a field named "cancelled" or "canceled"
+            // Find a field named "cancelled" or "canceled" that is a boolean, and has a backing field.
             type.allMembers.filter { it.name in CANCELLED_NAMES && it.returnType == typeOf<Boolean>() }
-                .filterIsInstance<KMutableProperty1<*, *>>().toList().let {
+                .filterIsInstance<KMutableProperty<*>>().filter { it.javaField != null }.toList().let {
                     if (it.isEmpty()) return NOT_CANCELLABLE
                     if (it.size != 1) config.logger.warn("Multiple possible cancel fields found for event type $type")
-                    OFFSETS[type] = UNSAFE.objectFieldOffset(it[0].javaField)
-                    // This is not using reflection, and it is the same speed as direct access.
+                    it[0].javaField!!.let { field ->
+                        if (Modifier.isStatic(field.modifiers)) OFFSETS[type] = UNSAFE.staticFieldOffset(field)
+                        else OFFSETS[type] = UNSAFE.objectFieldOffset(field)
+                    }
+                    // This is the same speed as direct access, plus one JNI call and hashmap access.
                     // If you are familiar with C, this is essentially the same idea as pointers.
                     return CancelledState { event -> UNSAFE.getBoolean(event, OFFSETS[type]!!) }
                 }
diff --git a/src/main/kotlin/me/bush/illnamethislater/Config.kt b/src/main/kotlin/me/bush/illnamethislater/Config.kt
index eb9ccf5..ef946c5 100644
--- a/src/main/kotlin/me/bush/illnamethislater/Config.kt
+++ b/src/main/kotlin/me/bush/illnamethislater/Config.kt
@@ -1,7 +1,9 @@
 package me.bush.illnamethislater
 
+import kotlinx.coroutines.Dispatchers
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
+import kotlin.coroutines.CoroutineContext
 
 
 /**
@@ -15,15 +17,33 @@ import org.apache.logging.log4j.Logger
 data class Config(
 
     /**
-     * The logger this [EventBus] will use to log errors, or [EventBus.debugInfo]
+     * The logger this [EventBus] will use to log errors, or log [EventBus.debugInfo]
      */
     val logger: Logger = LogManager.getLogger("Eventbus"),
 
     /**
+     * The [CoroutineContext] to use when posting events to parallel listeners. The default
+     * value will work just fine, but you can specify a custom context if desired.
+     *
+     * [What is a Coroutine Context?](https://kotlinlang.org/docs/coroutine-context-and-dispatchers.html)
+     *
+     * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
+     */
+    val parallelContext: CoroutineContext = Dispatchers.Default,
+
+    /**
      * Whether this [EventBus] should try to find a "cancelled" field in events being listened for that
      * are not a subclass of [Event]. This is experimental, and should be set to `false` if problems arise.
      *
      * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
      */
-    val thirdPartyCompatibility: Boolean = true
+    val thirdPartyCompatibility: Boolean = true,
+
+    /**
+     * Whether parallel listeners should be called before or after sequential listeners. Parallel listeners
+     * will always finish before sequential listeners are called, or before [EventBus.post] returns.
+     *
+     * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
+     */
+    val parallelFirst: Boolean = true
 )
diff --git a/src/main/kotlin/me/bush/illnamethislater/EventBus.kt b/src/main/kotlin/me/bush/illnamethislater/EventBus.kt
index 90ec395..6b9b0bf 100644
--- a/src/main/kotlin/me/bush/illnamethislater/EventBus.kt
+++ b/src/main/kotlin/me/bush/illnamethislater/EventBus.kt
@@ -2,17 +2,15 @@ package me.bush.illnamethislater
 
 import kotlin.reflect.KClass
 
-// TODO: 3/30/2022 Refactor some stuff
-
 /**
- * [A simple event dispatcher](http://github.com/therealbush/eventbus-kotlin)
+ * A simple event dispatcher
+ *
+ * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
  *
  * @author bush
  * @since 1.0.0
  */
-class EventBus(
-    private val config: Config = Config()
-) {
+class EventBus(private val config: Config = Config()) {
     private val listeners = hashMapOf<KClass<*>, ListenerGroup>()
     private val subscribers = mutableSetOf<Any>()
 
@@ -29,7 +27,7 @@ class EventBus(
                     it.subscriber = subscriber
                 })
             }
-            subscribers += subscriber
+            subscribers.add(subscriber)
             true
         }.getOrElse {
             config.logger.error("Unable to register listeners for subscriber $subscriber", it)
@@ -38,19 +36,6 @@ class EventBus(
     }
 
     /**
-     * Registers a listener (which may not belong to any subscriber) to this [EventBus]. If no object
-     * is given, a key will be returned which can be used in [unsubscribe] to remove the listener.
-     *
-     * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
-     */
-    fun register(listener: Listener): Listener {
-        listeners.computeIfAbsent(listener.type) {
-            ListenerGroup(it, config)
-        }.add(listener)
-        return listener
-    }
-
-    /**
      * doc
      *
      * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
@@ -58,18 +43,30 @@ class EventBus(
     fun unsubscribe(subscriber: Any): Boolean {
         return subscribers.remove(subscriber).also { contains ->
             if (contains) listeners.entries.removeIf {
-                it.value.removeFrom(subscriber)
+                it.value.unsubscribe(subscriber)
                 it.value.sequential.isEmpty() && it.value.parallel.isEmpty()
             }
         }
     }
 
     /**
+     * Registers a listener (which may not belong to any subscriber) to this [EventBus]. If no object
+     * is given, a key will be returned which can be used in [unsubscribe] to remove the listener.
+     *
+     * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
+     */
+    fun register(listener: Listener) = listener.also {
+        listeners.computeIfAbsent(it.type) { type -> ListenerGroup(type, config) }.register(it)
+    }
+
+    /**
      * doc
      *
      * [Information and examples](https://github.com/therealbush/eventbus-kotlin#tododothething)
      */
-    fun unregister(listener: Listener) = listeners[listener.type]?.remove(listener) ?: false
+    fun unregister(listener: Listener) = listener.also {
+        listeners[it.type]?.unregister(it)
+    }
 
     /**
      * Posts an event. doc
diff --git a/src/main/kotlin/me/bush/illnamethislater/Listener.kt b/src/main/kotlin/me/bush/illnamethislater/Listener.kt
index 2d96d54..7105baf 100644
--- a/src/main/kotlin/me/bush/illnamethislater/Listener.kt
+++ b/src/main/kotlin/me/bush/illnamethislater/Listener.kt
@@ -14,15 +14,15 @@ import kotlin.reflect.KClass
 class Listener @PublishedApi internal constructor(
     listener: (Nothing) -> Unit,
     internal val type: KClass<*>,
-    internal var priority: Int = 0,
-    internal var parallel: Boolean = false,
-    internal var receiveCancelled: Boolean = false
+    internal val priority: Int = 0,
+    internal val parallel: Boolean = false,
+    internal val receiveCancelled: Boolean = false
 ) {
     @Suppress("UNCHECKED_CAST")
     // Generics have no benefit here,
     // it is easier just to force cast.
     internal val listener = listener as (Any) -> Unit
-    internal lateinit var subscriber: Any
+    internal var subscriber: Any? = null
 }
 
 /**
diff --git a/src/main/kotlin/me/bush/illnamethislater/ListenerGroup.kt b/src/main/kotlin/me/bush/illnamethislater/ListenerGroup.kt
index 56b5421..9c8714b 100644
--- a/src/main/kotlin/me/bush/illnamethislater/ListenerGroup.kt
+++ b/src/main/kotlin/me/bush/illnamethislater/ListenerGroup.kt
@@ -1,9 +1,13 @@
 package me.bush.illnamethislater
 
+import kotlinx.coroutines.*
 import java.util.concurrent.CopyOnWriteArrayList
+import kotlin.coroutines.CoroutineContext
 import kotlin.reflect.KClass
 
 /**
+ * A class for storing and handling listeners.
+ *
  * @author bush
  * @since 1.0.0
  */
@@ -11,29 +15,70 @@ internal class ListenerGroup(
     private val type: KClass<*>,
     private val config: Config
 ) {
-    val cancelledState = CancelledState.of(type, config)
+    private val cancelledState = CancelledState.of(type, config)
     val sequential = CopyOnWriteArrayList<Listener>()
     val parallel = CopyOnWriteArrayList<Listener>()
 
-    fun add(listener: Listener) {
+    /**
+     * Adds [listener] to this [ListenerGroup], and sorts its list.
+     */
+    fun register(listener: Listener) {
         with(if (listener.parallel) parallel else sequential) {
             add(listener)
-            sortBy { it.priority }
+            sortedByDescending { it.priority }
         }
     }
 
-    fun remove(listener: Listener) = with(if (listener.parallel) parallel else sequential) {
-        remove(listener)
+    /**
+     * Removes [listener] from this [ListenerGroup].
+     */
+    fun unregister(listener: Listener) {
+        if (listener.parallel) parallel.remove(listener)
+        else sequential.remove(listener)
     }
 
-    fun removeFrom(subscriber: Any) {
-        parallel.removeIf(Listener::subscriber::equals)
-        sequential.removeIf(Listener::subscriber::equals)
+    /**
+     * Removes every listener whose subscriber is [subscriber].
+     */
+    fun unsubscribe(subscriber: Any) {
+        parallel.removeIf { it.subscriber == subscriber }
+        sequential.removeIf { it.subscriber == subscriber }
     }
 
+    /**
+     * Posts an event to every listener. Returns true of the event was cancelled.
+     */
     fun post(event: Any): Boolean {
-        return false
+        if (config.parallelFirst) postParallel(event)
+        sequential.forEach {
+            if (it.receiveCancelled || !cancelledState.isCancelled(event)) {
+                it.listener(event)
+            }
+        }
+        if (!config.parallelFirst) postParallel(event)
+        return cancelledState.isCancelled(event)
+    }
+
+    /**
+     * Posts an event to all parallel listeners. Cancel state of the event is checked once before
+     * posting the event as opposed to before calling each listener, to avoid inconsistencies.
+     */
+    private fun postParallel(event: Any) {
+        if (parallel.isEmpty()) return
+        // We check this once, because listener order is not consistent
+        val cancelled = cancelledState.isCancelled(event)
+        // Credit to KB for the idea
+        runBlocking(config.parallelContext) {
+            parallel.forEach {
+                if (it.receiveCancelled || !cancelled) launch {
+                    it.listener(event)
+                }
+            }
+        }
     }
 
+    /**
+     * Logs information about this [ListenerGroup].
+     */
     fun debugInfo() = config.logger.info("${type.simpleName}: ${sequential.size}, ${parallel.size}")
 }