Add EVENT_LAUNCH_UNDISPATCHED to allow to launch coroutines for event listeners in a UNDISPATCHED start mode

mirai-core-api/src/commonMain/kotlin/internal/event/InternalEventListeners.kt

@@ -15,6 +15,8 @@ import kotlinx.coroutines.sync.withLock
 import net.mamoe.mirai.event.*
 import net.mamoe.mirai.event.events.BotEvent
 import net.mamoe.mirai.utils.MiraiLogger
+import net.mamoe.mirai.utils.lateinitMutableProperty
+import net.mamoe.mirai.utils.systemProp
 import java.util.*
 import java.util.concurrent.ConcurrentLinkedQueue
 import kotlin.coroutines.CoroutineContext
@@ -29,7 +31,7 @@ internal class Handler<in E : Event> internal constructor(
     subscriberContext: CoroutineContext,
     @JvmField val handler: suspend (E) -> ListeningStatus,
     override val concurrencyKind: ConcurrencyKind,
-    override val priority: EventPriority
+    override val priority: EventPriority,
 ) : Listener<E>, CompletableJob by SupervisorJob(parentJob) { // avoid being cancelled on handling event
 
     private val subscriberContext: CoroutineContext = subscriberContext + this // override Job.
@@ -70,7 +72,7 @@ internal class Handler<in E : Event> internal constructor(
 
 internal class ListenerRegistry(
     val listener: Listener<Event>,
-    val type: KClass<out Event>
+    val type: KClass<out Event>,
 )
 
 
@@ -120,12 +122,26 @@ internal suspend fun <E : AbstractEvent> callAndRemoveIfRequired(event: E) {
         else -> supervisorScope {
             for (registry in GlobalEventListeners[EventPriority.MONITOR]) {
                 if (!registry.type.isInstance(event)) continue
-                launch { process(container, registry, registry.listener, event) }
+                launch(start = if (EVENT_LAUNCH_UNDISPATCHED) CoroutineStart.UNDISPATCHED else CoroutineStart.DEFAULT) {
+                    process(container, registry, registry.listener, event)
+                }
             }
         }
     }
 }
 
+/**
+ * If `true`, all event listeners runs directly in the broadcaster's thread until first suspension.
+ *
+ * If there is not suspension point in the listener, the coroutine executing [Event.broadcast] will not suspend,
+ * so the thread before and after execution will be the same and no other code is being executed if there is only one thread.
+ *
+ * This is useful for tests to not to depend on `delay`
+ */
+internal var EVENT_LAUNCH_UNDISPATCHED: Boolean by lateinitMutableProperty {
+    systemProp("mirai.event.launch.undispatched", false)
+}
+
 private suspend fun <E : AbstractEvent> process(
     container: ConcurrentLinkedQueue<ListenerRegistry>,
     registry: ListenerRegistry,

mirai-core-api/src/jvmTest/kotlin/event/EventLaunchUndispatchedTest.kt

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ *  此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ *  Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ *  https://github.com/mamoe/mirai/blob/master/LICENSE
+ */
+
+@file:JvmBlockingBridge
+
+package net.mamoe.mirai.event
+
+import kotlinx.coroutines.*
+import net.mamoe.kjbb.JvmBlockingBridge
+import net.mamoe.mirai.internal.event.EVENT_LAUNCH_UNDISPATCHED
+import org.junit.jupiter.api.Test
+import java.util.concurrent.ConcurrentLinkedQueue
+import kotlin.test.assertEquals
+import kotlin.test.assertFails
+import kotlin.test.assertSame
+
+@Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
+internal class EventLaunchUndispatchedTest : AbstractEventTest() {
+    internal class TestEvent : AbstractEvent()
+
+    val originalValue = EVENT_LAUNCH_UNDISPATCHED
+
+    @Test
+    suspend fun `event runs undispatched`() {
+        EVENT_LAUNCH_UNDISPATCHED = true
+        doTest()
+        EVENT_LAUNCH_UNDISPATCHED = originalValue
+    }
+
+    @Test
+    suspend fun `event runs undispatched fail`() {
+        EVENT_LAUNCH_UNDISPATCHED = false
+        assertFails { doTest() }
+        EVENT_LAUNCH_UNDISPATCHED = originalValue
+    }
+
+    private suspend fun doTest() = coroutineScope {
+        val invoked = ConcurrentLinkedQueue<Int>()
+
+        val thread = Thread.currentThread()
+
+        val job = SupervisorJob()
+        globalEventChannel().parentJob(job).exceptionHandler { } // printing exception to stdout is very slow
+            .run {
+                subscribeAlways<TestEvent>(priority = EventPriority.MONITOR) {
+                    assertSame(thread, Thread.currentThread())
+                    invoked.add(1)
+                    awaitCancellation()
+                }
+                repeat(10000) { i ->
+                    subscribeAlways<TestEvent>(priority = EventPriority.MONITOR) {
+                        assertSame(thread, Thread.currentThread())
+                        invoked.add(i + 2)
+                        awaitCancellation()
+                    }
+                }
+            }
+
+        launch(start = CoroutineStart.UNDISPATCHED) { TestEvent().broadcast() }
+        // `launch` returns on first suspension point of `broadcast`
+        // if EVENT_LAUNCH_UNDISPATCHED is `true`, all listeners run to `awaitCancellation` when `broadcast` is suspended
+        // otherwise, they are put into tasks queue to be scheduled. 10000 tasks wont complete very quickly, so the following `invoked.size` check works.
+        assertSame(thread, Thread.currentThread())
+        assertEquals(invoked.toList(), invoked.sorted())
+        assertEquals(10000 + 1, invoked.size)
+        job.cancel()
+    }
+}