From 6eff4bdf40815598a2d987e08d89df6b97663967 Mon Sep 17 00:00:00 2001 From: Him188 Date: Tue, 29 Jun 2021 22:42:47 +0800 Subject: [PATCH] Add `EVENT_LAUNCH_UNDISPATCHED` to allow to launch coroutines for event listeners in a UNDISPATCHED start mode --- .../internal/event/InternalEventListeners.kt | 22 +++++- .../event/EventLaunchUndispatchedTest.kt | 74 +++++++++++++++++++ 2 files changed, 93 insertions(+), 3 deletions(-) create mode 100644 mirai-core-api/src/jvmTest/kotlin/event/EventLaunchUndispatchedTest.kt diff --git a/mirai-core-api/src/commonMain/kotlin/internal/event/InternalEventListeners.kt b/mirai-core-api/src/commonMain/kotlin/internal/event/InternalEventListeners.kt index e6183ce5b..3be958e10 100644 --- a/mirai-core-api/src/commonMain/kotlin/internal/event/InternalEventListeners.kt +++ b/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 internal constructor( subscriberContext: CoroutineContext, @JvmField val handler: suspend (E) -> ListeningStatus, override val concurrencyKind: ConcurrencyKind, - override val priority: EventPriority + override val priority: EventPriority, ) : Listener, 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 internal constructor( internal class ListenerRegistry( val listener: Listener, - val type: KClass + val type: KClass, ) @@ -120,12 +122,26 @@ internal suspend fun 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 process( container: ConcurrentLinkedQueue, registry: ListenerRegistry, diff --git a/mirai-core-api/src/jvmTest/kotlin/event/EventLaunchUndispatchedTest.kt b/mirai-core-api/src/jvmTest/kotlin/event/EventLaunchUndispatchedTest.kt new file mode 100644 index 000000000..e936eb6a1 --- /dev/null +++ b/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() + + val thread = Thread.currentThread() + + val job = SupervisorJob() + globalEventChannel().parentJob(job).exceptionHandler { } // printing exception to stdout is very slow + .run { + subscribeAlways(priority = EventPriority.MONITOR) { + assertSame(thread, Thread.currentThread()) + invoked.add(1) + awaitCancellation() + } + repeat(10000) { i -> + subscribeAlways(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() + } +} \ No newline at end of file