diff --git a/common/src/main/java/taboolib/common/TabooLib.java b/common/src/main/java/taboolib/common/TabooLib.java index c26de1ccb..dfcf963ef 100644 --- a/common/src/main/java/taboolib/common/TabooLib.java +++ b/common/src/main/java/taboolib/common/TabooLib.java @@ -64,11 +64,14 @@ public Class getClass(String name, boolean initialize, ClassLoader classLoade * 执行生命周期任务 */ public static void lifeCycle(LifeCycle lifeCycle) { - if (isStopped) { + if (isStopped && lifeCycle != LifeCycle.DISABLE) { return; } // 检查 Kotlin 环境是否就绪 if (!TabooLib.isKotlinEnvironment()) { + if (lifeCycle == LifeCycle.DISABLE) { + return; + } isStopped = true; throw new RuntimeException( t( diff --git a/common/src/test/java/taboolib/common/TabooLibDisableLifecycleTest.java b/common/src/test/java/taboolib/common/TabooLibDisableLifecycleTest.java new file mode 100644 index 000000000..678d63d1a --- /dev/null +++ b/common/src/test/java/taboolib/common/TabooLibDisableLifecycleTest.java @@ -0,0 +1,30 @@ +package taboolib.common; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class TabooLibDisableLifecycleTest { + + @AfterEach + void restoreStoppedFlag() { + TabooLib.setStopped(false); + } + + @Test + void disableLifecycleStillRunsWhenLoadingWasStopped() { + AtomicInteger calls = new AtomicInteger(); + TabooLib.registerLifeCycleTask(LifeCycle.DISABLE, 0, calls::incrementAndGet); + TabooLib.setStopped(true); + + TabooLib.lifeCycle(LifeCycle.DISABLE); + + assertEquals(1, calls.get()); + assertEquals(LifeCycle.DISABLE, TabooLib.getCurrentLifeCycle()); + assertTrue(TabooLib.isStopped()); + } +} diff --git a/platform/platform-afybroker/build.gradle.kts b/platform/platform-afybroker/build.gradle.kts index 9e6ca2455..7093498d7 100644 --- a/platform/platform-afybroker/build.gradle.kts +++ b/platform/platform-afybroker/build.gradle.kts @@ -5,4 +5,9 @@ dependencies { compileOnly(project(":common-platform-api")) compileOnly("com.github.AfyerDev.AfyBroker:afybroker-server:f6261eab2a") compileOnly("org.slf4j:slf4j-api:1.7.32") + + testImplementation(project(":common")) + testImplementation(project(":common-util")) + testImplementation(project(":common-platform-api")) + testImplementation("com.github.AfyerDev.AfyBroker:afybroker-server:f6261eab2a") } \ No newline at end of file diff --git a/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerActiveGate.java b/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerActiveGate.java new file mode 100644 index 000000000..8756eed4e --- /dev/null +++ b/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerActiveGate.java @@ -0,0 +1,36 @@ +package taboolib.platform; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; + +final class AfyBrokerActiveGate { + + private enum State { + OPEN, + ACTIVATING, + CLOSED + } + + private final AtomicReference state = new AtomicReference<>(State.OPEN); + private final CompletableFuture activationClosed = new CompletableFuture<>(); + + boolean activate(Runnable action) { + if (!state.compareAndSet(State.OPEN, State.ACTIVATING)) { + return false; + } + try { + action.run(); + return true; + } finally { + state.set(State.CLOSED); + activationClosed.complete(null); + } + } + + CompletableFuture close() { + if (state.compareAndSet(State.OPEN, State.CLOSED)) { + activationClosed.complete(null); + } + return activationClosed; + } +} diff --git a/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerPlugin.java b/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerPlugin.java index 254d3209e..89dd3aaee 100644 --- a/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerPlugin.java +++ b/platform/platform-afybroker/src/main/java/taboolib/platform/AfyBrokerPlugin.java @@ -13,7 +13,9 @@ import taboolib.common.platform.Plugin; import java.io.File; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import static taboolib.common.PrimitiveIO.t; @@ -30,6 +32,8 @@ public class AfyBrokerPlugin extends net.afyer.afybroker.server.plugin.Plugin { @Nullable private static Plugin pluginInstance; private static AfyBrokerPlugin instance; + private final AfyBrokerActiveGate activeGate = new AfyBrokerActiveGate(); + private final AtomicBoolean disabled = new AtomicBoolean(); static { PrimitiveIO.debug("AfyBroker 插件初始化完成，用时 {0} 毫秒。", TabooLib.execution(() -> { @@ -107,12 +111,20 @@ public void onEnable() { Broker.getScheduler().schedule(this, new Runnable() { @Override public void run() { - // 生命周期任务 - TabooLib.lifeCycle(LifeCycle.ACTIVE); - // 调用 Plugin 实现的 onActive() 方法 - if (pluginInstance != null) { - pluginInstance.onActive(); - } + activeGate.activate(new Runnable() { + @Override + public void run() { + if (TabooLib.isStopped()) { + return; + } + // 生命周期任务 + TabooLib.lifeCycle(LifeCycle.ACTIVE); + // 调用 Plugin 实现的 onActive() 方法 + if (pluginInstance != null) { + pluginInstance.onActive(); + } + } + }); } }, 0, TimeUnit.MILLISECONDS); } @@ -120,12 +132,67 @@ public void run() { @Override public void onDisable() { + // 第一时间关闭激活入口；若 ACTIVE 正在执行，则在其结束后再进入 DISABLE + CompletableFuture activationClosed = activeGate.close(); + if (activationClosed.isDone()) { + disable(); + return; + } + activationClosed.whenComplete((unused, failure) -> { + if (failure != null) { + reportDisableFailure(failure); + return; + } + try { + disable(); + } catch (Throwable ex) { + reportDisableFailure(ex); + } + }); + } + + private void disable() { + if (!disabled.compareAndSet(false, true)) { + return; + } + Throwable failure = null; // 在插件未关闭的前提下，执行 onDisable() 方法 if (pluginInstance != null && !TabooLib.isStopped()) { - pluginInstance.onDisable(); + try { + pluginInstance.onDisable(); + } catch (Throwable ex) { + failure = ex; + } } - // 生命周期任务 - TabooLib.lifeCycle(LifeCycle.DISABLE); + // 生命周期任务必须执行，不能被用户回调异常跳过 + try { + TabooLib.lifeCycle(LifeCycle.DISABLE); + } catch (Throwable ex) { + if (failure == null) { + failure = ex; + } else { + failure.addSuppressed(ex); + } + } + if (failure != null) { + AfyBrokerPlugin.rethrow(failure); + } + } + + private void reportDisableFailure(Throwable ex) { + try { + PrimitiveIO.error("AfyBroker 平台禁用流程执行异常：{0}", ex.getMessage() == null ? ex.getClass().getName() : ex.getMessage()); + } catch (Throwable ignored) { + } + try { + ex.printStackTrace(); + } catch (Throwable ignored) { + } + } + + @SuppressWarnings("unchecked") + private static void rethrow(Throwable throwable) throws T { + throw (T) throwable; } @NotNull diff --git a/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutor.kt b/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutor.kt index 27272d6a0..ab20a1289 100644 --- a/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutor.kt +++ b/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutor.kt @@ -4,13 +4,16 @@ import net.afyer.afybroker.server.Broker import net.afyer.afybroker.server.scheduler.ScheduledTask import taboolib.common.Inject import taboolib.common.LifeCycle +import taboolib.common.PrimitiveIO +import taboolib.common.TabooLib import taboolib.common.platform.Awake import taboolib.common.platform.Platform import taboolib.common.platform.PlatformSide import taboolib.common.platform.service.PlatformExecutor import java.io.Closeable -import java.util.concurrent.CompletableFuture +import java.util.concurrent.RejectedExecutionException import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean /** * TabooLib @@ -24,88 +27,195 @@ import java.util.concurrent.TimeUnit @PlatformSide(Platform.AFYBROKER) class AfyBrokerExecutor : PlatformExecutor { - private val tasks = ArrayList() - private var started = false + private val tasks = AfyBrokerTaskRegistry() + + init { + TabooLib.registerLifeCycleTask(LifeCycle.DISABLE, 2) { stop() } + } @Awake(LifeCycle.ENABLE) override fun start() { - started = true - // 提交列队中的任务 - tasks.forEach { - if (it.runnable.now) { - it.execute() - } else { - it.execute(it.runnable.async, it.runnable.delay, it.runnable.period) + executeAll(tasks.start()) + } + + fun stop() { + cancelAll(tasks.stop()) + } + + private fun executeAll(pendingTasks: List) { + var failure: Throwable? = null + pendingTasks.forEach { task -> + try { + execute(task) + } catch (ex: Throwable) { + if (failure == null) { + failure = ex + } else { + failure?.addSuppressed(ex) + } } } - tasks.clear() + failure?.let { throw it } + } + + private fun cancelAll(activeTasks: List) { + var failure: Throwable? = null + activeTasks.forEach { task -> + try { + task.cancel() + } catch (ex: Throwable) { + if (failure == null) { + failure = ex + } else { + failure?.addSuppressed(ex) + } + } + } + failure?.let { throw it } + } + + private fun execute(task: AfyBrokerRunningTask) { + if (task.runnable.now) { + task.execute() + } else { + task.execute(task.runnable.async, task.runnable.delay, task.runnable.period) + } } class AfyBrokerRunningTask(val runnable: PlatformExecutor.PlatformRunnable) { + private val cancellation = AfyBrokerTaskCancellation { it.cancel() } + private var onCancelled: () -> Unit = {} + private var onCompleted: () -> Unit = {} + lateinit var scheduledTask: ScheduledTask + internal fun observe(onCancelled: () -> Unit, onCompleted: () -> Unit) { + this.onCancelled = onCancelled + this.onCompleted = onCompleted + } + fun execute() { - runnable.executor(BrokerPlatformTask { }) + executeUserTask(completeAfterRun = true) } fun execute(async: Boolean, delay: Long, period: Long) { - scheduledTask = if (period < 1) { - if (async) { - Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { - Broker.getScheduler().runAsync(AfyBrokerPlugin.getInstance()) { - runnable.executor(platformTask()) - } - }, delay * 50L, TimeUnit.MILLISECONDS) + if (cancellation.isCancelled()) { + onCompleted() + return + } + try { + val scheduled = if (period < 1) { + scheduleOnce(async, delay) } else { - Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { - runnable.executor(platformTask()) - }, delay * 50L, TimeUnit.MILLISECONDS) + scheduleRepeated(async, delay, period) } + scheduledTask = scheduled + cancellation.bind(scheduled) + } catch (ex: Throwable) { + onCompleted() + throw ex + } + } + + private fun scheduleOnce(async: Boolean, delay: Long): ScheduledTask { + return if (async) { + Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { + if (!cancellation.isCancelled()) { + runAfyBrokerDispatch(::reportTaskFailure, onCompleted) { + Broker.getScheduler().runAsync(AfyBrokerPlugin.getInstance()) { + executeUserTask(completeAfterRun = true) + } + } + } else { + onCompleted() + } + }, delay * 50L, TimeUnit.MILLISECONDS) } else { - if (async) { - Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { - Broker.getScheduler().runAsync(AfyBrokerPlugin.getInstance()) { - runnable.executor(platformTask()) + Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { + executeUserTask(completeAfterRun = true) + }, delay * 50L, TimeUnit.MILLISECONDS) + } + } + + private fun scheduleRepeated(async: Boolean, delay: Long, period: Long): ScheduledTask { + return if (async) { + Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { + if (!cancellation.isCancelled()) { + runAfyBrokerDispatch(::reportTaskFailure, ::cancel) { + Broker.getScheduler().runAsync(AfyBrokerPlugin.getInstance()) { + executeUserTask(completeAfterRun = false) + } } - }, delay * 50L, period * 50L, TimeUnit.MILLISECONDS) - } else { - Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { + } + }, delay * 50L, period * 50L, TimeUnit.MILLISECONDS) + } else { + Broker.getScheduler().schedule(AfyBrokerPlugin.getInstance(), { + executeUserTask(completeAfterRun = false) + }, delay * 50L, period * 50L, TimeUnit.MILLISECONDS) + } + } + + private fun executeUserTask(completeAfterRun: Boolean) { + try { + cancellation.runIfActive { + runAfyBrokerTask(::reportTaskFailure) { runnable.executor(platformTask()) - }, delay * 50L, period * 50L, TimeUnit.MILLISECONDS) + } + } + } finally { + if (completeAfterRun) { + onCompleted() } } } fun platformTask(): PlatformExecutor.PlatformTask { - return BrokerPlatformTask { scheduledTask.cancel() } + return BrokerPlatformTask { cancel() } + } + + internal fun cancel() { + if (this::scheduledTask.isInitialized) { + cancellation.bind(scheduledTask) + } + cancellation.cancel(onCancelled) + } + + private fun reportTaskFailure(ex: Throwable) { + PrimitiveIO.error( + "AfyBroker 平台任务执行异常：{0}", + ex.message ?: ex.javaClass.name + ) + ex.printStackTrace() } } override fun submit(runnable: PlatformExecutor.PlatformRunnable): PlatformExecutor.PlatformTask { val task = AfyBrokerRunningTask(runnable) - return if (started) { - if (runnable.now) { - task.execute() - } else { - task.execute(runnable.async, runnable.delay, runnable.period) - } - task.platformTask() - } else { - tasks += task - BrokerPlatformTask { - if (!task.runnable.now) { - task.platformTask().cancel() - } - tasks -= task + task.observe( + onCancelled = { tasks.remove(task) }, + onCompleted = { tasks.remove(task) } + ) + val platformTask = task.platformTask() + when (tasks.register(task)) { + AfyBrokerTaskRegistration.PENDING -> Unit + AfyBrokerTaskRegistration.ACTIVE -> execute(task) + AfyBrokerTaskRegistration.REJECTED -> { + task.cancel() + throw RejectedExecutionException("AfyBrokerExecutor has been stopped") } } + return platformTask } class BrokerPlatformTask(val runnable: Closeable) : PlatformExecutor.PlatformTask { + private val cancelled = AtomicBoolean() + override fun cancel() { - runnable.close() + if (cancelled.compareAndSet(false, true)) { + runnable.close() + } } } -} \ No newline at end of file +} diff --git a/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutorLifecycle.kt b/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutorLifecycle.kt new file mode 100644 index 000000000..9899ba0a1 --- /dev/null +++ b/platform/platform-afybroker/src/main/kotlin/taboolib/platform/AfyBrokerExecutorLifecycle.kt @@ -0,0 +1,173 @@ +package taboolib.platform + +internal enum class AfyBrokerExecutorState { + NEW, + RUNNING, + STOPPED +} + +internal enum class AfyBrokerTaskRegistration { + PENDING, + ACTIVE, + REJECTED +} + +internal class AfyBrokerTaskRegistry { + + private val lock = Any() + private val pending = LinkedHashSet() + private val active = LinkedHashSet() + private var state = AfyBrokerExecutorState.NEW + + fun register(task: T): AfyBrokerTaskRegistration { + return synchronized(lock) { + when (state) { + AfyBrokerExecutorState.NEW -> { + pending += task + AfyBrokerTaskRegistration.PENDING + } + AfyBrokerExecutorState.RUNNING -> { + active += task + AfyBrokerTaskRegistration.ACTIVE + } + AfyBrokerExecutorState.STOPPED -> AfyBrokerTaskRegistration.REJECTED + } + } + } + + fun start(): List { + return synchronized(lock) { + if (state != AfyBrokerExecutorState.NEW) { + return@synchronized emptyList() + } + state = AfyBrokerExecutorState.RUNNING + val tasks = pending.toList() + pending.clear() + active += tasks + tasks + } + } + + fun remove(task: T): Boolean { + return synchronized(lock) { + pending.remove(task) || active.remove(task) + } + } + + fun stop(): List { + return synchronized(lock) { + if (state == AfyBrokerExecutorState.STOPPED) { + return@synchronized emptyList() + } + state = AfyBrokerExecutorState.STOPPED + val tasks = ArrayList(pending.size + active.size) + tasks += pending + tasks += active + pending.clear() + active.clear() + tasks + } + } + + fun state(): AfyBrokerExecutorState { + return synchronized(lock) { state } + } + + fun pendingCount(): Int { + return synchronized(lock) { pending.size } + } + + fun activeCount(): Int { + return synchronized(lock) { active.size } + } +} + +internal class AfyBrokerTaskCancellation(private val cancelDelegate: (T) -> Unit) { + + private val lock = Any() + + @Volatile + private var cancelled = false + private var delegate: T? = null + + fun bind(value: T) { + val cancelNow = synchronized(lock) { + val current = delegate + check(current == null || current === value) { "Scheduled task is already bound" } + if (current == null) { + delegate = value + cancelled + } else { + false + } + } + if (cancelNow) { + cancelDelegate(value) + } + } + + fun cancel(afterCancellation: () -> Unit = {}): Boolean { + val bound = synchronized(lock) { + if (cancelled) { + return false + } + cancelled = true + delegate + } + try { + if (bound != null) { + cancelDelegate(bound) + } + } finally { + afterCancellation() + } + return true + } + + fun isCancelled(): Boolean { + return cancelled + } + + fun runIfActive(action: () -> Unit): Boolean { + if (cancelled) { + return false + } + action() + return true + } +} + +internal inline fun runAfyBrokerDispatch( + reporter: (Throwable) -> Unit, + cleanup: () -> Unit, + action: () -> T, +): T { + try { + return action() + } catch (ex: Throwable) { + try { + reporter(ex) + } catch (reportingFailure: Throwable) { + ex.addSuppressed(reportingFailure) + } + try { + cleanup() + } catch (cleanupFailure: Throwable) { + ex.addSuppressed(cleanupFailure) + } + throw ex + } +} + +internal inline fun runAfyBrokerTask(reporter: (Throwable) -> Unit, action: () -> T): T { + try { + return action() + } catch (ex: Throwable) { + try { + reporter(ex) + } catch (reportingFailure: Throwable) { + ex.addSuppressed(reportingFailure) + } + throw ex + } +} diff --git a/platform/platform-afybroker/src/test/kotlin/taboolib/platform/AfyBrokerExecutorLifecycleTest.kt b/platform/platform-afybroker/src/test/kotlin/taboolib/platform/AfyBrokerExecutorLifecycleTest.kt new file mode 100644 index 000000000..a5c02dbf3 --- /dev/null +++ b/platform/platform-afybroker/src/test/kotlin/taboolib/platform/AfyBrokerExecutorLifecycleTest.kt @@ -0,0 +1,249 @@ +package taboolib.platform + +import net.afyer.afybroker.server.scheduler.ScheduledTask +import org.junit.jupiter.api.Assertions.assertDoesNotThrow +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import taboolib.common.platform.service.PlatformExecutor +import java.io.Closeable +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.atomic.AtomicReference + +class AfyBrokerExecutorLifecycleTest { + + @Test + fun `cancel before binding cancels delegate exactly once`() { + val delegate = ManualDelegate() + val cancellation = AfyBrokerTaskCancellation { it.cancel() } + + assertTrue(cancellation.cancel()) + assertFalse(cancellation.cancel()) + cancellation.bind(delegate) + cancellation.bind(delegate) + + assertEquals(1, delegate.cancelCount) + assertTrue(cancellation.isCancelled()) + } + + @Test + fun `cancellation cleanup runs even when delegate throws`() { + val failure = IllegalStateException("cancel failed") + val delegate = ManualDelegate() + val cancellation = AfyBrokerTaskCancellation { throw failure } + var cleanupCount = 0 + cancellation.bind(delegate) + + val thrown = assertThrows(IllegalStateException::class.java) { + cancellation.cancel { cleanupCount++ } + } + + assertSame(failure, thrown) + assertEquals(1, cleanupCount) + assertTrue(cancellation.isCancelled()) + assertFalse(cancellation.cancel { cleanupCount++ }) + assertEquals(1, cleanupCount) + } + + @Test + fun `platform task cancel is idempotent`() { + var cancelCount = 0 + val task = AfyBrokerExecutor.BrokerPlatformTask(Closeable { cancelCount++ }) + + task.cancel() + task.cancel() + + assertEquals(1, cancelCount) + } + + @Test + fun `pending task cancel does not access unbound scheduled task`() { + val runningTask = AfyBrokerExecutor.AfyBrokerRunningTask( + PlatformExecutor.PlatformRunnable(false, false, 0, 0) {} + ) + + assertDoesNotThrow { runningTask.platformTask().cancel() } + } + + @Test + fun `binding before cancel is safe and idempotent`() { + val delegate = ManualDelegate() + val cancellation = AfyBrokerTaskCancellation { it.cancel() } + + cancellation.bind(delegate) + assertEquals(0, delegate.cancelCount) + assertTrue(cancellation.cancel()) + assertFalse(cancellation.cancel()) + + assertEquals(1, delegate.cancelCount) + } + + @Test + fun `cancelled task gate rejects later execution`() { + val cancellation = AfyBrokerTaskCancellation { it.cancel() } + var executions = 0 + + cancellation.cancel() + + assertFalse(cancellation.runIfActive { executions++ }) + assertEquals(0, executions) + } + + @Test + fun `registry moves pending tasks to active and completes them`() { + val registry = AfyBrokerTaskRegistry() + + assertEquals(AfyBrokerTaskRegistration.PENDING, registry.register("pending")) + assertEquals(AfyBrokerExecutorState.NEW, registry.state()) + assertEquals(1, registry.pendingCount()) + + assertEquals(listOf("pending"), registry.start()) + assertEquals(AfyBrokerExecutorState.RUNNING, registry.state()) + assertEquals(0, registry.pendingCount()) + assertEquals(1, registry.activeCount()) + assertEquals(AfyBrokerTaskRegistration.ACTIVE, registry.register("active")) + assertTrue(registry.remove("pending")) + assertEquals(1, registry.activeCount()) + } + + @Test + fun `stop drains pending and active tasks then rejects submissions`() { + val registry = AfyBrokerTaskRegistry() + registry.register("pending") + registry.start() + registry.register("active") + + assertEquals(listOf("pending", "active"), registry.stop()) + assertEquals(AfyBrokerExecutorState.STOPPED, registry.state()) + assertEquals(0, registry.pendingCount()) + assertEquals(0, registry.activeCount()) + assertEquals(AfyBrokerTaskRegistration.REJECTED, registry.register("late")) + assertTrue(registry.stop().isEmpty()) + assertTrue(registry.start().isEmpty()) + } + + @Test + fun `stopped executor rejects now and scheduled submissions`() { + val executor = AfyBrokerExecutor() + executor.stop() + + assertThrows(RejectedExecutionException::class.java) { + executor.submit(PlatformExecutor.PlatformRunnable(true, false, 0, 0) {}) + } + assertThrows(RejectedExecutionException::class.java) { + executor.submit(PlatformExecutor.PlatformRunnable(false, false, 0, 0) {}) + } + } + + @Test + fun `async dispatch failure reports and cleans up without replacing failure`() { + val failure = RejectedExecutionException("dispatch rejected") + val cleanupFailure = IllegalStateException("cleanup failed") + var reported: Throwable? = null + var cleanupCount = 0 + + val thrown = assertThrows(RejectedExecutionException::class.java) { + runAfyBrokerDispatch( + reporter = { reported = it }, + cleanup = { + cleanupCount++ + throw cleanupFailure + }, + ) { + throw failure + } + } + + assertSame(failure, thrown) + assertSame(failure, reported) + assertEquals(1, cleanupCount) + assertEquals(listOf(cleanupFailure), failure.suppressed.toList()) + } + + @Test + fun `user task failure is reported and rethrown unchanged`() { + val failure = IllegalStateException("boom") + var reported: Throwable? = null + + val thrown = assertThrows(IllegalStateException::class.java) { + runAfyBrokerTask({ reported = it }) { + throw failure + } + } + + assertSame(failure, reported) + assertSame(failure, thrown) + } + + @Test + fun `reporter failure is suppressed without replacing user failure`() { + val failure = IllegalStateException("user") + val reporterFailure = IllegalArgumentException("reporter") + + val thrown = assertThrows(IllegalStateException::class.java) { + runAfyBrokerTask({ throw reporterFailure }) { + throw failure + } + } + + assertSame(failure, thrown) + assertEquals(listOf(reporterFailure), thrown.suppressed.toList()) + } + + @Test + fun `active gate prevents callback after disable`() { + val gate = AfyBrokerActiveGate() + var activeCalls = 0 + + assertTrue(gate.close().isDone) + assertFalse(gate.activate { activeCalls++ }) + assertEquals(0, activeCalls) + } + + @Test + fun `active gate defers disable continuation without blocking`() { + val gate = AfyBrokerActiveGate() + val order = ArrayList() + val closed = AtomicReference>() + + assertTrue(gate.activate { + order += "active-start" + closed.set(gate.close()) + assertFalse(closed.get().isDone) + closed.get().thenRun { order += "disable" } + order += "active-end" + }) + + assertTrue(closed.get().isDone) + assertEquals(listOf("active-start", "active-end", "disable"), order) + assertFalse(gate.activate { order += "late-active" }) + } + + @Test + fun `public executor contract remains compatible`() { + val executorClass = AfyBrokerExecutor::class.java + val runningTaskClass = AfyBrokerExecutor.AfyBrokerRunningTask::class.java + + executorClass.getDeclaredConstructor() + assertTrue(PlatformExecutor::class.java.isAssignableFrom(executorClass)) + assertEquals(ScheduledTask::class.java, runningTaskClass.getField("scheduledTask").type) + assertEquals(ScheduledTask::class.java, runningTaskClass.getMethod("getScheduledTask").returnType) + assertEquals( + PlatformExecutor.PlatformTask::class.java, + runningTaskClass.getMethod("platformTask").returnType + ) + } + + private class ManualDelegate { + + var cancelCount = 0 + private set + + fun cancel() { + cancelCount++ + } + } +} diff --git a/platform/platform-application/build.gradle.kts b/platform/platform-application/build.gradle.kts index 561762910..10b3997f8 100644 --- a/platform/platform-application/build.gradle.kts +++ b/platform/platform-application/build.gradle.kts @@ -3,6 +3,10 @@ dependencies { compileOnly(project(":common-env")) compileOnly(project(":common-util")) compileOnly(project(":common-platform-api")) + testImplementation(project(":common")) + testImplementation(project(":common-env")) + testImplementation(project(":common-util")) + testImplementation(project(":common-platform-api")) // 工具 implementation("net.minecrell:terminalconsoleappender:1.3.0") // implementation("org.apache.logging.log4j:log4j-api:2.17.2") diff --git a/platform/platform-application/src/main/java/taboolib/platform/App.java b/platform/platform-application/src/main/java/taboolib/platform/App.java index 5fd8797c8..ae9c8ba1a 100644 --- a/platform/platform-application/src/main/java/taboolib/platform/App.java +++ b/platform/platform-application/src/main/java/taboolib/platform/App.java @@ -1,6 +1,5 @@ package taboolib.platform; -import taboolib.common.LifeCycle; import taboolib.common.PrimitiveIO; import taboolib.common.TabooLib; import taboolib.common.classloader.IsolatedClassLoader; @@ -19,6 +18,9 @@ @PlatformSide(Platform.APPLICATION) public class App { + private static final AppLifeCycle LIFE_CYCLE = new AppLifeCycle(); + private static volatile boolean running; + static { // 如果是 Application 启动，则跳过重定向 env().skipSelfRelocate(true).skipKotlinRelocate(true); @@ -46,10 +48,21 @@ public static void init() { // 初始化 IsolatedClassLoader IsolatedClassLoader.init(App.class); // 生命周期任务 - TabooLib.lifeCycle(LifeCycle.CONST); - TabooLib.lifeCycle(LifeCycle.INIT); - TabooLib.lifeCycle(LifeCycle.LOAD); - TabooLib.lifeCycle(LifeCycle.ENABLE); + running = true; + try { + running = LIFE_CYCLE.run(TabooLib::lifeCycle) && !TabooLib.isStopped(); + if (TabooLib.isStopped()) { + LIFE_CYCLE.shutdown(TabooLib::lifeCycle); + } + } catch (RuntimeException | Error ex) { + running = false; + try { + LIFE_CYCLE.shutdown(TabooLib::lifeCycle); + } catch (RuntimeException | Error cleanupFailure) { + ex.addSuppressed(cleanupFailure); + } + throw ex; + } })); } @@ -57,7 +70,12 @@ public static void init() { * 结束 */ public static void shutdown() { - TabooLib.lifeCycle(LifeCycle.DISABLE); + running = false; + LIFE_CYCLE.shutdown(TabooLib::lifeCycle); + } + + static boolean isRunning() { + return running; } /** diff --git a/platform/platform-application/src/main/java/taboolib/platform/AppLifeCycle.java b/platform/platform-application/src/main/java/taboolib/platform/AppLifeCycle.java new file mode 100644 index 000000000..cd59cf79a --- /dev/null +++ b/platform/platform-application/src/main/java/taboolib/platform/AppLifeCycle.java @@ -0,0 +1,154 @@ +package taboolib.platform; + +import taboolib.common.LifeCycle; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.function.Consumer; + +final class AppLifeCycle { + + private static final List INITIALIZATION = Collections.unmodifiableList(Arrays.asList( + LifeCycle.CONST, + LifeCycle.INIT, + LifeCycle.LOAD, + LifeCycle.ENABLE, + LifeCycle.ACTIVE + )); + + private enum State { + NEW, + INITIALIZING, + ACTIVE, + STOP_REQUESTED, + DISABLING, + DISABLED + } + + private final Object lock = new Object(); + private State state = State.NEW; + private boolean transitionRunning; + + static List initialization() { + return INITIALIZATION; + } + + boolean run(Consumer action) { + synchronized (lock) { + if (state == State.INITIALIZING || state == State.ACTIVE) { + return true; + } + if (state != State.NEW) { + return false; + } + state = State.INITIALIZING; + } + for (LifeCycle lifeCycle : INITIALIZATION) { + if (!beginTransition()) { + return isRunning(); + } + Throwable failure = null; + try { + action.accept(lifeCycle); + } catch (Throwable ex) { + failure = ex; + } + boolean disable = finishTransition(); + if (disable) { + try { + runDisable(action); + } catch (Throwable ex) { + if (failure == null) { + failure = ex; + } else { + failure.addSuppressed(ex); + } + } + } + if (failure != null) { + AppLifeCycle.rethrow(failure); + } + if (disable) { + return false; + } + } + synchronized (lock) { + if (state == State.INITIALIZING) { + state = State.ACTIVE; + return true; + } + return state == State.ACTIVE; + } + } + + void shutdown(Consumer action) { + boolean disable = false; + synchronized (lock) { + switch (state) { + case NEW: + case ACTIVE: + state = State.DISABLING; + disable = true; + break; + case INITIALIZING: + state = State.STOP_REQUESTED; + if (!transitionRunning) { + state = State.DISABLING; + disable = true; + } + break; + case STOP_REQUESTED: + case DISABLING: + case DISABLED: + return; + } + } + if (disable) { + runDisable(action); + } + } + + private boolean beginTransition() { + synchronized (lock) { + if (state != State.INITIALIZING) { + return false; + } + transitionRunning = true; + return true; + } + } + + private boolean finishTransition() { + synchronized (lock) { + transitionRunning = false; + if (state == State.STOP_REQUESTED) { + state = State.DISABLING; + return true; + } + return false; + } + } + + private void runDisable(Consumer action) { + try { + action.accept(LifeCycle.DISABLE); + } finally { + synchronized (lock) { + transitionRunning = false; + state = State.DISABLED; + } + } + } + + private boolean isRunning() { + synchronized (lock) { + return state == State.INITIALIZING || state == State.ACTIVE; + } + } + + @SuppressWarnings("unchecked") + private static void rethrow(Throwable throwable) throws T { + throw (T) throwable; + } +} diff --git a/platform/platform-application/src/main/kotlin/taboolib/platform/AppCommand.kt b/platform/platform-application/src/main/kotlin/taboolib/platform/AppCommand.kt index 283f06751..f71ae62a5 100644 --- a/platform/platform-application/src/main/kotlin/taboolib/platform/AppCommand.kt +++ b/platform/platform-application/src/main/kotlin/taboolib/platform/AppCommand.kt @@ -11,6 +11,7 @@ import taboolib.common.platform.command.CommandStructure import taboolib.common.platform.command.component.CommandBase import taboolib.common.platform.function.info import taboolib.common.platform.service.PlatformCommand +import java.util.concurrent.CopyOnWriteArraySet /** * @author Score2 @@ -26,15 +27,14 @@ class AppCommand : PlatformCommand { val unknownCommandMessage: String get() = System.getProperty("taboolib.application.command.unknown.message") ?: "Unknown command." - val commands = mutableSetOf() + val commands: MutableSet = CopyOnWriteArraySet() fun register(command: Command) { commands.add(command) } fun unregister(name: String) { - commands.find { it.command.aliases.contains(name) } ?: return - unregister(name) + commands.removeIf { it.matches(name) } } fun unregister(command: Command) { @@ -46,7 +46,7 @@ class AppCommand : PlatformCommand { return } val label = if (content.contains(" ")) content.substringBefore(" ") else content - val command = commands.find { it.aliases.contains(label) } ?: return info(unknownCommandMessage) + val command = commands.find { it.matches(label) } ?: return info(unknownCommandMessage) val args = if (content.contains(" ")) content.substringAfter(" ").split(" ") else listOf() command.executor.execute(AppConsole, command.command, label, args.toTypedArray()) } @@ -57,7 +57,7 @@ class AppCommand : PlatformCommand { return suggestion() } val label = if (content.contains(" ")) content.substringBefore(" ") else content - val command = commands.find { it.aliases.contains(label) } ?: return suggestion().filter { it.startsWith(label) } + val command = commands.find { it.matches(label) } ?: return suggestion().filter { it.startsWith(label, ignoreCase = true) } return if (content.contains(" ")) { command.completer.execute(AppConsole, command.command, label, content.substringAfter(" ").split(" ").toTypedArray()) ?: listOf() } else { @@ -71,6 +71,8 @@ class AppCommand : PlatformCommand { val aliases get() = listOf(command.name, *command.aliases.toTypedArray()) + fun matches(name: String) = aliases.any { it.equals(name, ignoreCase = true) } + fun register() = register(this) fun unregister() = unregister(this) @@ -85,10 +87,10 @@ class AppCommand : PlatformCommand { } override fun unregisterCommand(command: String) { - unregister(commands.find { it.command.aliases.contains(command) } ?: return) + unregister(command) } override fun unregisterCommands() { - commands.forEach { unregister(it) } + commands.clear() } } \ No newline at end of file diff --git a/platform/platform-application/src/main/kotlin/taboolib/platform/AppConsole.kt b/platform/platform-application/src/main/kotlin/taboolib/platform/AppConsole.kt index 3000aa16a..09d2d16cd 100644 --- a/platform/platform-application/src/main/kotlin/taboolib/platform/AppConsole.kt +++ b/platform/platform-application/src/main/kotlin/taboolib/platform/AppConsole.kt @@ -16,6 +16,10 @@ import taboolib.common.platform.function.info import taboolib.common.platform.function.pluginId import taboolib.common.platform.function.pluginVersion +internal fun isApplicationRunning(running: Boolean, stopped: Boolean): Boolean { + return running && !stopped +} + /** * @author Score2 * @since 2022/06/08 13:37 @@ -75,7 +79,7 @@ object AppConsole : SimpleTerminalConsole(), ProxyCommandSender { } override fun isRunning(): Boolean { - return !TabooLib.isStopped() + return isApplicationRunning(App.isRunning(), TabooLib.isStopped()) } override fun buildReader(builder: LineReaderBuilder): LineReader { diff --git a/platform/platform-application/src/main/kotlin/taboolib/platform/AppExecutor.kt b/platform/platform-application/src/main/kotlin/taboolib/platform/AppExecutor.kt index 9ee1d5602..efe0e5ac7 100644 --- a/platform/platform-application/src/main/kotlin/taboolib/platform/AppExecutor.kt +++ b/platform/platform-application/src/main/kotlin/taboolib/platform/AppExecutor.kt @@ -1,13 +1,23 @@ package taboolib.platform import taboolib.common.Inject +import taboolib.common.LifeCycle +import taboolib.common.PrimitiveIO +import taboolib.common.TabooLib import taboolib.common.platform.Awake import taboolib.common.platform.Platform import taboolib.common.platform.PlatformSide import taboolib.common.platform.service.PlatformExecutor import java.util.concurrent.CompletableFuture import java.util.concurrent.Executors +import java.util.concurrent.Future +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.ThreadFactory import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference /** * TabooLib @@ -19,42 +29,129 @@ import java.util.concurrent.TimeUnit @Awake @Inject @PlatformSide(Platform.APPLICATION) -class AppExecutor : PlatformExecutor { +class AppExecutor private constructor( + private val executor: ScheduledExecutorService, + private val exceptionReporter: (Throwable) -> Unit, + registerStopTask: Boolean, +) : PlatformExecutor { - private val tasks = ArrayList() - private val executor = Executors.newScheduledThreadPool(16) + constructor() : this(createExecutor(), ::reportTaskException, true) + internal enum class State { + NEW, RUNNING, STOPPED + } + + private val state = AtomicReference(State.NEW) + + init { + if (registerStopTask) { + TabooLib.registerLifeCycleTask(LifeCycle.DISABLE, 2) { stop() } + } + } + + @Awake(LifeCycle.ENABLE) override fun start() { + state.compareAndSet(State.NEW, State.RUNNING) } override fun submit(runnable: PlatformExecutor.PlatformRunnable): PlatformExecutor.PlatformTask { - val future = CompletableFuture() - val task = AppPlatformTask(future) - val scheduledTask = when { - runnable.now -> { - runnable.executor(task) - null - } - runnable.period > 0 -> { - executor.scheduleAtFixedRate({ runnable.executor(task) }, runnable.delay * 50L, runnable.period * 50L, TimeUnit.MILLISECONDS) - } - runnable.delay > 0 -> { - executor.schedule({ runnable.executor(task) }, runnable.delay * 50L, TimeUnit.MILLISECONDS) - } - else -> { - executor.submit { runnable.executor(task) } + rejectIfStopped() + val task = AppPlatformTask() + if (runnable.now) { + executeUserTask(task, runnable) + return task + } + val command = Runnable { + if (!task.isCancelled) { + executeUserTask(task, runnable) } } - future.thenAccept { - scheduledTask?.cancel(false) + val future = when { + runnable.period > 0 -> executor.scheduleAtFixedRate(command, runnable.delay * 50L, runnable.period * 50L, TimeUnit.MILLISECONDS) + runnable.delay > 0 -> executor.schedule(command, runnable.delay * 50L, TimeUnit.MILLISECONDS) + else -> executor.schedule(command, 0L, TimeUnit.MILLISECONDS) } + task.attach(future) return task } - class AppPlatformTask(private val future: CompletableFuture) : PlatformExecutor.PlatformTask { + fun stop() { + if (state.getAndSet(State.STOPPED) != State.STOPPED) { + executor.shutdownNow() + } + } + + internal fun currentState(): State = state.get() + + private fun rejectIfStopped() { + if (state.get() == State.STOPPED) { + throw RejectedExecutionException("AppExecutor has been stopped") + } + } + + private fun executeUserTask(task: AppPlatformTask, runnable: PlatformExecutor.PlatformRunnable) { + runAppTask(exceptionReporter) { runnable.executor(task) } + } + + class AppPlatformTask() : PlatformExecutor.PlatformTask { + + private val cancelled = AtomicBoolean(false) + private val future = AtomicReference?>() + private var cancellationSignal: CompletableFuture? = null + + constructor(cancellationSignal: CompletableFuture) : this() { + this.cancellationSignal = cancellationSignal + } + + internal val isCancelled: Boolean + get() = cancelled.get() + + internal fun attach(scheduled: Future<*>) { + check(future.compareAndSet(null, scheduled)) { "Scheduled task is already bound" } + if (cancelled.get()) { + scheduled.cancel(false) + } + } override fun cancel() { - future.complete(null) + if (cancelled.compareAndSet(false, true)) { + cancellationSignal?.complete(null) + future.get()?.cancel(false) + } + } + } + + companion object { + + private fun createExecutor(): ScheduledExecutorService { + return Executors.newScheduledThreadPool(16, AppExecutorThreadFactory()) + } + + private fun reportTaskException(ex: Throwable) { + PrimitiveIO.error("Application 平台任务执行异常：{0}", ex.message ?: ex.javaClass.name) + ex.printStackTrace() + } + } +} + +internal class AppExecutorThreadFactory : ThreadFactory { + + private val counter = AtomicInteger() + + override fun newThread(runnable: Runnable): Thread { + return Thread(runnable, "TabooLib-Application-Executor-${counter.incrementAndGet()}") + } +} + +internal inline fun runAppTask(reporter: (Throwable) -> Unit, action: () -> T): T { + try { + return action() + } catch (ex: Throwable) { + try { + reporter(ex) + } catch (reportingFailure: Throwable) { + ex.addSuppressed(reportingFailure) } + throw ex } -} \ No newline at end of file +} diff --git a/platform/platform-application/src/main/kotlin/taboolib/platform/AppIO.kt b/platform/platform-application/src/main/kotlin/taboolib/platform/AppIO.kt index c8cb79f6b..858906a5b 100644 --- a/platform/platform-application/src/main/kotlin/taboolib/platform/AppIO.kt +++ b/platform/platform-application/src/main/kotlin/taboolib/platform/AppIO.kt @@ -81,12 +81,14 @@ class AppIO : PlatformIO { if (file.exists() && !replace) { return file } - newFile(file).writeBytes(javaClass.classLoader.getResourceAsStream(source)?.readBytes() ?: error("resource not found: $source")) + val content = javaClass.classLoader.getResourceAsStream(source)?.use { it.readBytes() } + ?: error("resource not found: $source") + newFile(file).writeBytes(content) return file } override fun getJarFile(): File { - return File(AppIO::class.java.protectionDomain.codeSource.location.toURI().path) + return File(AppIO::class.java.protectionDomain.codeSource.location.toURI()) } override fun getDataFolder(): File { diff --git a/platform/platform-application/src/test/kotlin/taboolib/platform/ApplicationPlatformTest.kt b/platform/platform-application/src/test/kotlin/taboolib/platform/ApplicationPlatformTest.kt new file mode 100644 index 000000000..4d8f7ea0a --- /dev/null +++ b/platform/platform-application/src/test/kotlin/taboolib/platform/ApplicationPlatformTest.kt @@ -0,0 +1,206 @@ +package taboolib.platform + +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import taboolib.common.LifeCycle +import taboolib.common.platform.command.CommandCompleter +import taboolib.common.platform.command.CommandExecutor +import taboolib.common.platform.command.CommandStructure +import taboolib.common.platform.command.PermissionDefault +import taboolib.common.platform.service.PlatformExecutor +import java.lang.reflect.Modifier +import java.util.concurrent.CompletableFuture +import java.util.concurrent.FutureTask +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.atomic.AtomicInteger + +class ApplicationPlatformTest { + + @AfterEach + fun cleanupCommands() { + AppCommand.commands.clear() + } + + @Test + fun `initialization lifecycle reaches active in order`() { + assertEquals( + listOf(LifeCycle.CONST, LifeCycle.INIT, LifeCycle.LOAD, LifeCycle.ENABLE, LifeCycle.ACTIVE), + AppLifeCycle.initialization() + ) + } + + @Test + fun `shutdown during enable prevents active lifecycle regression`() { + val lifeCycle = AppLifeCycle() + val calls = ArrayList() + + val running = lifeCycle.run { + calls += it + if (it == LifeCycle.ENABLE) { + lifeCycle.shutdown { calls += it } + } + } + lifeCycle.shutdown { calls += it } + + assertFalse(running) + assertEquals( + listOf(LifeCycle.CONST, LifeCycle.INIT, LifeCycle.LOAD, LifeCycle.ENABLE, LifeCycle.DISABLE), + calls + ) + } + + @Test + fun `console stops running at disable`() { + assertTrue(isApplicationRunning(true, false)) + assertFalse(isApplicationRunning(false, false)) + assertFalse(isApplicationRunning(true, true)) + assertTrue(Modifier.isVolatile(App::class.java.getDeclaredField("running").modifiers)) + } + + @Test + fun `command unregister matches primary name and aliases`() { + val service = AppCommand() + val primary = command("primary", listOf("alias")) + val other = command("other", listOf("secondary")) + primary.register() + other.register() + + service.unregisterCommand("PRIMARY") + assertEquals(setOf(other), AppCommand.commands) + + service.unregisterCommand("SECONDARY") + assertTrue(AppCommand.commands.isEmpty()) + } + + @Test + fun `command bulk unregister clears concurrent set`() { + val service = AppCommand() + command("one").register() + command("two").register() + + service.unregisterCommands() + + assertTrue(AppCommand.commands.isEmpty()) + assertEquals(java.util.Set::class.java, AppCommand.Companion::class.java.getMethod("getCommands").returnType) + } + + @Test + fun `executor has explicit lifecycle and rejects all tasks after stop`() { + val executor = AppExecutor() + try { + assertEquals(AppExecutor.State.NEW, executor.currentState()) + executor.start() + assertEquals(AppExecutor.State.RUNNING, executor.currentState()) + executor.stop() + assertEquals(AppExecutor.State.STOPPED, executor.currentState()) + + assertThrows(RejectedExecutionException::class.java) { + executor.submit(runnable(now = true) {}) + } + assertThrows(RejectedExecutionException::class.java) { + executor.submit(runnable(now = false) {}) + } + } finally { + executor.stop() + } + } + + @Test + fun `executor keeps immediate pre-start behavior and exposes task failures`() { + val executor = AppExecutor() + val executions = AtomicInteger() + try { + executor.submit(runnable(now = true) { executions.incrementAndGet() }) + assertEquals(1, executions.get()) + assertThrows(IllegalStateException::class.java) { + executor.submit(runnable(now = true) { error("observable") }) + } + } finally { + executor.stop() + } + } + + @Test + fun `executor task failure is reported and rethrown unchanged`() { + val failure = IllegalStateException("boom") + var reported: Throwable? = null + + val thrown = assertThrows(IllegalStateException::class.java) { + runAppTask({ reported = it }) { throw failure } + } + + assertTrue(reported === failure) + assertTrue(thrown === failure) + } + + @Test + fun `executor task cancellation is idempotent and worker threads are named`() { + val task = AppExecutor.AppPlatformTask() + val future = RecordingFuture() + task.cancel() + task.attach(future) + task.cancel() + assertTrue(task.isCancelled) + assertEquals(1, future.cancelCount) + + val cancellationSignal = CompletableFuture() + val compatibleTask = AppExecutor.AppPlatformTask(cancellationSignal) + compatibleTask.cancel() + compatibleTask.cancel() + assertTrue(cancellationSignal.isDone) + AppExecutor.AppPlatformTask::class.java.getConstructor(CompletableFuture::class.java) + + val factory = AppExecutorThreadFactory() + assertEquals("TabooLib-Application-Executor-1", factory.newThread {}.name) + assertEquals("TabooLib-Application-Executor-2", factory.newThread {}.name) + } + + private fun runnable(now: Boolean, block: PlatformExecutor.PlatformTask.() -> Unit): PlatformExecutor.PlatformRunnable { + return PlatformExecutor.PlatformRunnable(now, async = false, delay = 0, period = 0, executor = block) + } + + private class RecordingFuture : FutureTask(Runnable {}, Unit) { + + var cancelCount = 0 + + override fun cancel(mayInterruptIfRunning: Boolean): Boolean { + cancelCount++ + return super.cancel(mayInterruptIfRunning) + } + } + + private fun command(name: String, aliases: List = emptyList()): AppCommand.Command { + val structure = CommandStructure( + name, + aliases, + "", + "", + "", + "", + PermissionDefault.TRUE, + emptyMap(), + false + ) + val executor = object : CommandExecutor { + override fun execute( + sender: taboolib.common.platform.ProxyCommandSender, + command: CommandStructure, + name: String, + args: Array + ): Boolean = true + } + val completer = object : CommandCompleter { + override fun execute( + sender: taboolib.common.platform.ProxyCommandSender, + command: CommandStructure, + name: String, + args: Array + ): List = emptyList() + } + return AppCommand.Command(structure, executor, completer) {} + } +} diff --git a/platform/platform-bukkit-impl/src/main/kotlin/taboolib/platform/BukkitCommand.kt b/platform/platform-bukkit-impl/src/main/kotlin/taboolib/platform/BukkitCommand.kt index e68cda3f4..4749b03d2 100644 --- a/platform/platform-bukkit-impl/src/main/kotlin/taboolib/platform/BukkitCommand.kt +++ b/platform/platform-bukkit-impl/src/main/kotlin/taboolib/platform/BukkitCommand.kt @@ -31,6 +31,25 @@ import taboolib.common.platform.service.PlatformCommand import taboolib.common.util.unsafeLazy import java.lang.reflect.Constructor +internal fun commandLabelMatches(name: String, aliases: List, input: String, namespace: String): Boolean { + val separator = input.indexOf(':') + val label = if (separator >= 0) { + if (!input.substring(0, separator).equals(namespace, ignoreCase = true)) { + return false + } + input.substring(separator + 1) + } else { + input + } + return label.equals(name, ignoreCase = true) || aliases.any { it.equals(label, ignoreCase = true) } +} + +internal fun removeMappingsByIdentity(commands: MutableMap, target: T): Boolean { + val keys = commands.filterValues { it === target }.keys.toList() + keys.forEach(commands::remove) + return keys.isNotEmpty() +} + /** * TabooLib * taboolib.platform.BukkitCommand @@ -62,8 +81,12 @@ class BukkitCommand : PlatformCommand { val registeredCommands = ArrayList() + private val commandLock = Any() + private val registeredCommandBindings = ArrayList() private var isSupportedUnknownCommand = false + private data class RegisteredCommand(val structure: CommandStructure, val command: PluginCommand) + override fun registerCommand( command: CommandStructure, executor: CommandExecutor, @@ -109,14 +132,6 @@ class BukkitCommand : PlatformCommand { command.permissionChildren.forEach { registerPermission(it.key, it.value) } - // 注册命令 - knownCommands.remove(command.name) - knownCommands["${plugin.name.lowercase()}:${pluginCommand.name}"] = pluginCommand - knownCommands[pluginCommand.name] = pluginCommand - pluginCommand.aliases.forEach { - knownCommands[it] = pluginCommand - } - pluginCommand.register(commandMap) // 1.8 patch runCatching { if (pluginCommand.getProperty("timings") == null) { @@ -124,19 +139,55 @@ class BukkitCommand : PlatformCommand { pluginCommand.setProperty("timings", timingsManager.invokeMethod("getCommandTiming", plugin.name, pluginCommand, isStatic = true)) } } + // 注册命令及身份记录作为同一个事务；同名重注册前先清理旧实例的全部映射 + synchronized(commandLock) { + registeredCommandBindings + .filter { it.structure.name.equals(command.name, ignoreCase = true) } + .toList() + .forEach(::unregisterBinding) + knownCommands["${plugin.name.lowercase()}:${pluginCommand.name}"] = pluginCommand + knownCommands[pluginCommand.name] = pluginCommand + pluginCommand.aliases.forEach { + knownCommands[it] = pluginCommand + } + pluginCommand.register(commandMap) + registeredCommands.add(command) + registeredCommandBindings.add(RegisteredCommand(command, pluginCommand)) + } sync() - registeredCommands.add(command) } } override fun unregisterCommand(command: String) { - knownCommands.remove(command) - sync() + val removed = synchronized(commandLock) { + registeredCommandBindings + .filter { commandLabelMatches(it.structure.name, it.structure.aliases, command, plugin.name.lowercase()) } + .toList() + .also { it.forEach(::unregisterBinding) } + .isNotEmpty() + } + if (removed) { + sync() + } } override fun unregisterCommands() { - registeredCommands.forEach { taboolib.common.platform.function.unregisterCommand(it) } - sync() + val removed = synchronized(commandLock) { + registeredCommandBindings.toList().also { it.forEach(::unregisterBinding) }.isNotEmpty() + } + if (removed) { + sync() + } + } + + private fun unregisterBinding(binding: RegisteredCommand) { + removeMappingsByIdentity(knownCommands, binding.command) + binding.command.unregister(commandMap) + registeredCommandBindings.remove(binding) + val index = registeredCommands.indexOfFirst { it === binding.structure } + if (index >= 0) { + registeredCommands.removeAt(index) + } } override fun unknownCommand(sender: ProxyCommandSender, command: String, state: Int) { diff --git a/platform/platform-bukkit-impl/src/test/kotlin/taboolib/platform/BukkitCommandRegistryTest.kt b/platform/platform-bukkit-impl/src/test/kotlin/taboolib/platform/BukkitCommandRegistryTest.kt new file mode 100644 index 000000000..233b80bc4 --- /dev/null +++ b/platform/platform-bukkit-impl/src/test/kotlin/taboolib/platform/BukkitCommandRegistryTest.kt @@ -0,0 +1,60 @@ +package taboolib.platform + +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class BukkitCommandRegistryTest { + + @Test + fun `matches primary aliases and own namespace`() { + assertTrue(commandLabelMatches("main", listOf("alias"), "main", "plugin")) + assertTrue(commandLabelMatches("main", listOf("alias"), "alias", "plugin")) + assertTrue(commandLabelMatches("main", listOf("alias"), "plugin:main", "plugin")) + assertTrue(commandLabelMatches("main", listOf("alias"), "plugin:alias", "plugin")) + assertTrue(commandLabelMatches("main", listOf("alias"), "PLUGIN:MAIN", "plugin")) + assertTrue(commandLabelMatches("main", listOf("alias"), "ALIAS", "plugin")) + assertFalse(commandLabelMatches("main", listOf("alias"), "other:main", "plugin")) + assertFalse(commandLabelMatches("main", listOf("alias"), "missing", "plugin")) + } + + @Test + fun `re-registration cleanup removes old and new aliases by identity`() { + val old = Any() + val replacement = Any() + val commands = linkedMapOf( + "main" to old, + "old-alias" to old, + "plugin:main" to old, + ) + + assertTrue(removeMappingsByIdentity(commands, old)) + commands["main"] = replacement + commands["new-alias"] = replacement + commands["plugin:main"] = replacement + assertFalse(commands.containsKey("old-alias")) + + assertTrue(removeMappingsByIdentity(commands, replacement)) + assertTrue(commands.isEmpty()) + } + + @Test + fun `removes every mapping for the same command instance`() { + val target = Any() + val other = Any() + val commands = linkedMapOf( + "main" to target, + "alias" to target, + "plugin:main" to target, + "other" to other, + ) + + assertTrue(removeMappingsByIdentity(commands, target)) + assertSame(other, commands["other"]) + assertFalse(commands.containsKey("main")) + assertFalse(commands.containsKey("alias")) + assertFalse(commands.containsKey("plugin:main")) + assertFalse(removeMappingsByIdentity(commands, target)) + } +} diff --git a/platform/platform-velocity-impl/build.gradle.kts b/platform/platform-velocity-impl/build.gradle.kts index bed6616b4..db2936291 100644 --- a/platform/platform-velocity-impl/build.gradle.kts +++ b/platform/platform-velocity-impl/build.gradle.kts @@ -8,4 +8,10 @@ dependencies { compileOnly(project(":common-platform-api")) compileOnly(project(":platform:platform-velocity")) compileOnly("com.velocitypowered:velocity-api:3.1.1") + + testImplementation(project(":common")) + testImplementation(project(":common-util")) + testImplementation(project(":common-platform-api")) + testImplementation(project(":platform:platform-velocity")) + testImplementation("com.velocitypowered:velocity-api:3.1.1") } \ No newline at end of file diff --git a/platform/platform-velocity-impl/src/main/kotlin/taboolib/platform/VelocityExecutor.kt b/platform/platform-velocity-impl/src/main/kotlin/taboolib/platform/VelocityExecutor.kt index cfd1e2744..431cd9448 100644 --- a/platform/platform-velocity-impl/src/main/kotlin/taboolib/platform/VelocityExecutor.kt +++ b/platform/platform-velocity-impl/src/main/kotlin/taboolib/platform/VelocityExecutor.kt @@ -1,18 +1,24 @@ package taboolib.platform import com.velocitypowered.api.scheduler.ScheduledTask +import org.slf4j.LoggerFactory import taboolib.common.Inject import taboolib.common.LifeCycle import taboolib.common.platform.Awake import taboolib.common.platform.Platform import taboolib.common.platform.PlatformSide +import taboolib.common.platform.function.registerLifeCycleTask import taboolib.common.platform.service.PlatformExecutor import taboolib.common.util.unsafeLazy import java.io.Closeable -import java.util.concurrent.CompletableFuture +import java.util.concurrent.ExecutorService import java.util.concurrent.Executors +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.ThreadFactory import java.util.concurrent.TimeUnit -import kotlin.text.repeat +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference /** * TabooLib @@ -24,110 +30,325 @@ import kotlin.text.repeat @Awake @Inject @PlatformSide(Platform.VELOCITY) -class VelocityExecutor : PlatformExecutor { +class VelocityExecutor internal constructor( + private val taskScheduler: VelocityTaskScheduler?, + private val asyncExecutor: ExecutorService, + private val exceptionReporter: (Throwable) -> Unit, + registerStopTask: Boolean, +) : PlatformExecutor { - private val tasks = ArrayList() - private var started = false - private val executor = Executors.newFixedThreadPool(16) + constructor() : this(null, createAsyncExecutor(), ::reportTaskException, true) + + internal enum class State { + NEW, RUNNING, STOPPED + } + + private val lock = Any() + private val pendingTasks = LinkedHashSet() + private val activeTasks = LinkedHashSet() + + @Volatile + private var state = State.NEW val plugin by unsafeLazy { VelocityPlugin.getInstance() } + init { + if (registerStopTask) { + registerLifeCycleTask(LifeCycle.DISABLE, 2) { stop() } + } + } + @Awake(LifeCycle.ENABLE) override fun start() { - started = true + val tasks = synchronized(lock) { + when (state) { + State.NEW -> { + state = State.RUNNING + pendingTasks.filterNotTo(ArrayList()) { it.isCancelled }.also { + pendingTasks.clear() + activeTasks.addAll(it) + } + } + State.RUNNING, State.STOPPED -> return + } + } + var failure: Throwable? = null + tasks.forEach { + try { + launch(it) + } catch (ex: Throwable) { + if (failure == null) { + failure = ex + } else { + failure?.addSuppressed(ex) + } + } + } + failure?.let { throw it } + } + + fun stop() { + val tasks = synchronized(lock) { + if (state == State.STOPPED) { + return + } + state = State.STOPPED + LinkedHashSet().also { + it.addAll(pendingTasks) + it.addAll(activeTasks) + pendingTasks.clear() + activeTasks.clear() + } + } + var failure: Throwable? = null tasks.forEach { - if (it.runnable.now) { - it.executeNow() + try { + it.cancel() + } catch (ex: Throwable) { + if (failure == null) { + failure = ex + } else { + failure?.addSuppressed(ex) + } + } + } + try { + asyncExecutor.shutdownNow() + } catch (ex: Throwable) { + if (failure == null) { + failure = ex } else { - it.execute() + failure?.addSuppressed(ex) } } - tasks.clear() + failure?.let { throw it } } fun execute(velocityRunningTask: VelocityRunningTask, runnable: PlatformExecutor.PlatformRunnable): ScheduledTask { - + val action = Runnable { executeScheduled(velocityRunningTask, runnable) } + taskScheduler?.let { return it.schedule(velocityRunningTask, runnable, action) } return when { runnable.period > 0 -> plugin.server.scheduler - .buildTask(plugin) { - if (runnable.async) { - executor.submit { runnable.executor(velocityRunningTask.platformTask()) } - } else { - runnable.executor(velocityRunningTask.platformTask()) - } - } + .buildTask(plugin, action) .delay(runnable.delay * 50, TimeUnit.MILLISECONDS) .repeat(runnable.period * 50, TimeUnit.MILLISECONDS) .schedule() runnable.delay > 0 -> plugin.server.scheduler - .buildTask(plugin) { - if (runnable.async) { - executor.submit { runnable.executor(velocityRunningTask.platformTask()) } - } else { - runnable.executor(velocityRunningTask.platformTask()) - } - } + .buildTask(plugin, action) .delay(runnable.delay * 50, TimeUnit.MILLISECONDS) .schedule() - else -> plugin.server.scheduler - .buildTask(plugin) { - if (runnable.async) { - executor.submit { runnable.executor(velocityRunningTask.platformTask()) } - } else { - runnable.executor(velocityRunningTask.platformTask()) + else -> plugin.server.scheduler.buildTask(plugin, action).schedule() + } + } + + override fun submit(runnable: PlatformExecutor.PlatformRunnable): PlatformExecutor.PlatformTask { + val task = VelocityRunningTask(this, runnable) + val launchNow = synchronized(lock) { + when (state) { + State.NEW -> { + pendingTasks += task + false + } + State.RUNNING -> { + activeTasks += task + true + } + State.STOPPED -> throw RejectedExecutionException("VelocityExecutor has been stopped") + } + } + if (launchNow) { + launch(task) + } + return task.platformTask() + } + + private fun launch(task: VelocityRunningTask) { + if (task.isCancelled) { + taskFinished(task) + return + } + if (task.runnable.now) { + try { + task.executeNow() + } finally { + taskFinished(task) + } + } else { + try { + task.execute() + } catch (ex: Throwable) { + taskFinished(task) + throw ex + } + } + } + + private fun executeScheduled(task: VelocityRunningTask, runnable: PlatformExecutor.PlatformRunnable) { + if (task.isCancelled) { + return + } + if (runnable.async) { + val started = AtomicBoolean(false) + try { + asyncExecutor.execute { + started.set(true) + executeUserTask(task, runnable) + } + } catch (ex: Throwable) { + if (!started.get()) { + reportTaskFailure(ex) + try { + task.cancel() + } catch (cancellationFailure: Throwable) { + ex.addSuppressed(cancellationFailure) } - }.schedule() + } + throw ex + } + } else { + executeUserTask(task, runnable) + } + } + + private fun executeUserTask(task: VelocityRunningTask, runnable: PlatformExecutor.PlatformRunnable) { + if (task.isCancelled) { + return + } + try { + runnable.executor(task.platformTask()) + } catch (ex: Throwable) { + reportTaskFailure(ex) + throw ex + } finally { + if (runnable.period <= 0) { + taskFinished(task) + } + } + } + + private fun reportTaskFailure(ex: Throwable) { + try { + exceptionReporter(ex) + } catch (reportingFailure: Throwable) { + ex.addSuppressed(reportingFailure) + } + } + + private fun taskFinished(task: VelocityRunningTask) { + synchronized(lock) { + pendingTasks -= task + activeTasks -= task } } + internal fun taskCancelled(task: VelocityRunningTask) { + taskFinished(task) + } + + internal fun currentState(): State = state + + internal fun pendingTaskCount(): Int = synchronized(lock) { pendingTasks.size } + + internal fun activeTaskCount(): Int = synchronized(lock) { activeTasks.size } + class VelocityRunningTask(val executor: VelocityExecutor, val runnable: PlatformExecutor.PlatformRunnable) { lateinit var scheduledTask: ScheduledTask + private val cancelled = AtomicBoolean(false) + private val scheduledTaskReference = AtomicReference() + private val scheduledTaskCancelled = AtomicBoolean(false) + + internal val isCancelled: Boolean + get() = cancelled.get() + fun executeNow() { - runnable.executor(VelocityPlatformTask { }) + if (!isCancelled) { + executor.executeUserTask(this, runnable) + } } fun execute() { - scheduledTask = executor.execute(this, runnable) + if (isCancelled) { + return + } + val task = executor.execute(this, runnable) + scheduledTask = task + bind(task) } fun platformTask(): PlatformExecutor.PlatformTask { - return VelocityPlatformTask { scheduledTask.cancel() } + return VelocityPlatformTask { cancel() } } - } - override fun submit(runnable: PlatformExecutor.PlatformRunnable): PlatformExecutor.PlatformTask { - - val task = VelocityRunningTask(this, runnable) + fun cancel() { + if (cancelled.compareAndSet(false, true)) { + try { + val scheduled = scheduledTaskReference.get() + ?: if (this::scheduledTask.isInitialized) scheduledTask else null + scheduled?.let(::cancelScheduledTask) + } finally { + executor.taskCancelled(this) + } + } + } - return if (started) { - if (runnable.now) { - task.executeNow() - VelocityPlatformTask { } - } else { - task.execute() - task.platformTask() + private fun bind(task: ScheduledTask) { + check(scheduledTaskReference.compareAndSet(null, task)) { "Scheduled task is already bound" } + if (isCancelled) { + cancelScheduledTask(task) } - } else { - tasks += task - VelocityPlatformTask { - if (!runnable.now) { - task.platformTask().cancel() - } - tasks -= task + } + + private fun cancelScheduledTask(task: ScheduledTask) { + if (scheduledTaskCancelled.compareAndSet(false, true)) { + task.cancel() } } } class VelocityPlatformTask(val runnable: Closeable) : PlatformExecutor.PlatformTask { + private val cancelled = AtomicBoolean(false) + override fun cancel() { - runnable.close() + if (cancelled.compareAndSet(false, true)) { + runnable.close() + } } } -} \ No newline at end of file + + companion object { + + private fun createAsyncExecutor(): ExecutorService { + return Executors.newFixedThreadPool(16, VelocityAsyncThreadFactory()) + } + + private fun reportTaskException(ex: Throwable) { + val logger = try { + VelocityPlugin.getInstance().logger + } catch (_: Throwable) { + LoggerFactory.getLogger(VelocityExecutor::class.java) + } + logger.error("Unhandled exception in a TabooLib Velocity task", ex) + } + } +} + +internal interface VelocityTaskScheduler { + + fun schedule(task: VelocityExecutor.VelocityRunningTask, runnable: PlatformExecutor.PlatformRunnable, action: Runnable): ScheduledTask +} + +internal class VelocityAsyncThreadFactory : ThreadFactory { + + private val counter = AtomicInteger() + + override fun newThread(runnable: Runnable): Thread { + return Thread(runnable, "TabooLib-Velocity-Async-${counter.incrementAndGet()}") + } +} diff --git a/platform/platform-velocity-impl/src/test/kotlin/taboolib/platform/VelocityExecutorTest.kt b/platform/platform-velocity-impl/src/test/kotlin/taboolib/platform/VelocityExecutorTest.kt new file mode 100644 index 000000000..1e730b794 --- /dev/null +++ b/platform/platform-velocity-impl/src/test/kotlin/taboolib/platform/VelocityExecutorTest.kt @@ -0,0 +1,285 @@ +package taboolib.platform + +import com.velocitypowered.api.scheduler.ScheduledTask +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import taboolib.common.platform.service.PlatformExecutor +import java.lang.reflect.Proxy +import java.util.concurrent.AbstractExecutorService +import java.util.concurrent.ExecutorService +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.TimeUnit + +class VelocityExecutorTest { + + @Test + fun `cancelled pending task never reads lateinit or gets scheduled`() { + val scheduler = RecordingScheduler() + val executor = executor(scheduler) + val task = executor.submit(runnable()) + + task.cancel() + task.cancel() + executor.start() + + assertEquals(0, scheduler.scheduled.size) + assertEquals(0, executor.pendingTaskCount()) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `cancellation before scheduled handle binding cancels handle exactly once`() { + val scheduler = RecordingScheduler { task, _ -> task.cancel() } + val executor = executor(scheduler) + executor.start() + + val task = executor.submit(runnable()) + task.cancel() + + assertEquals(1, scheduler.scheduled.single().cancelCount) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `cancellation after scheduled handle binding is idempotent`() { + val scheduler = RecordingScheduler() + val executor = executor(scheduler) + executor.start() + + val task = executor.submit(runnable()) + task.cancel() + task.cancel() + + assertEquals(1, scheduler.scheduled.single().cancelCount) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `start moves pending tasks and stop is terminal`() { + val scheduler = RecordingScheduler() + val asyncExecutor = DirectExecutorService() + val executor = executor(scheduler, asyncExecutor) + executor.submit(runnable()) + + assertEquals(VelocityExecutor.State.NEW, executor.currentState()) + assertEquals(1, executor.pendingTaskCount()) + + executor.start() + assertEquals(VelocityExecutor.State.RUNNING, executor.currentState()) + assertEquals(0, executor.pendingTaskCount()) + assertEquals(1, executor.activeTaskCount()) + + executor.stop() + executor.stop() + + assertEquals(VelocityExecutor.State.STOPPED, executor.currentState()) + assertEquals(0, executor.activeTaskCount()) + assertEquals(1, scheduler.scheduled.single().cancelCount) + assertEquals(1, asyncExecutor.shutdownNowCount) + assertThrows(RejectedExecutionException::class.java) { executor.submit(runnable(now = true)) } + assertThrows(RejectedExecutionException::class.java) { executor.submit(runnable()) } + } + + @Test + fun `stop continues cleanup when one scheduled cancellation fails`() { + val scheduler = RecordingScheduler() + val asyncExecutor = DirectExecutorService() + val executor = executor(scheduler, asyncExecutor) + val failure = IllegalStateException("cancel failed") + executor.start() + executor.submit(runnable()) + executor.submit(runnable()) + scheduler.scheduled.first().cancelFailure = failure + + val thrown = assertThrows(IllegalStateException::class.java) { executor.stop() } + + assertSame(failure, thrown) + assertEquals(1, scheduler.scheduled.first().cancelCount) + assertEquals(1, scheduler.scheduled.last().cancelCount) + assertEquals(1, asyncExecutor.shutdownNowCount) + assertEquals(VelocityExecutor.State.STOPPED, executor.currentState()) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `now task queued before start runs without velocity scheduler`() { + val scheduler = RecordingScheduler() + val executor = executor(scheduler) + var executions = 0 + executor.submit(runnable(now = true) { executions++ }) + + executor.start() + + assertEquals(1, executions) + assertTrue(scheduler.scheduled.isEmpty()) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `user exception is reported and rethrown`() { + val scheduler = RecordingScheduler() + val reports = ArrayList() + val executor = executor(scheduler, reporter = reports::add) + val failure = IllegalStateException("boom") + executor.start() + executor.submit(runnable(async = true) { throw failure }) + + val thrown = assertThrows(IllegalStateException::class.java) { + scheduler.scheduled.single().action.run() + } + + assertSame(failure, thrown) + assertEquals(listOf(failure), reports) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `public task contract keeps scheduled task field`() { + val runningTask = VelocityExecutor.VelocityRunningTask::class.java + + assertEquals(ScheduledTask::class.java, runningTask.getField("scheduledTask").type) + assertEquals(ScheduledTask::class.java, runningTask.getMethod("getScheduledTask").returnType) + VelocityExecutor::class.java.getDeclaredConstructor() + } + + @Test + fun `async dispatch rejection reports failure and completes task`() { + val scheduler = RecordingScheduler() + val failure = RejectedExecutionException("dispatch rejected") + val reports = ArrayList() + val executor = executor(scheduler, RejectingExecutorService(failure)) { reports.add(it) } + executor.start() + executor.submit(runnable(async = true)) + + val thrown = assertThrows(RejectedExecutionException::class.java) { + scheduler.scheduled.single().action.run() + } + + assertSame(failure, thrown) + assertEquals(listOf(failure), reports) + assertEquals(1, scheduler.scheduled.single().cancelCount) + assertEquals(0, executor.activeTaskCount()) + } + + @Test + fun `async thread names use velocity prefix`() { + val factory = VelocityAsyncThreadFactory() + + val first = factory.newThread {} + val second = factory.newThread {} + + assertEquals("TabooLib-Velocity-Async-1", first.name) + assertEquals("TabooLib-Velocity-Async-2", second.name) + assertFalse(first.isAlive) + assertFalse(second.isAlive) + } + + private fun executor( + scheduler: RecordingScheduler, + asyncExecutor: ExecutorService = DirectExecutorService(), + reporter: (Throwable) -> Unit = {}, + ): VelocityExecutor { + return VelocityExecutor(scheduler, asyncExecutor, reporter, false) + } + + private fun runnable( + now: Boolean = false, + async: Boolean = false, + delay: Long = 0, + period: Long = 0, + block: PlatformExecutor.PlatformTask.() -> Unit = {}, + ): PlatformExecutor.PlatformRunnable { + return PlatformExecutor.PlatformRunnable(now, async, delay, period, block) + } + + private class RecordingScheduler( + private val beforeReturn: (VelocityExecutor.VelocityRunningTask, RecordedTask) -> Unit = { _, _ -> }, + ) : VelocityTaskScheduler { + + val scheduled = ArrayList() + + override fun schedule( + task: VelocityExecutor.VelocityRunningTask, + runnable: PlatformExecutor.PlatformRunnable, + action: Runnable, + ): ScheduledTask { + val recorded = RecordedTask(action) + scheduled += recorded + beforeReturn(task, recorded) + return recorded.handle + } + } + + private class RecordedTask(val action: Runnable) { + + var cancelCount = 0 + var cancelFailure: Throwable? = null + + val handle: ScheduledTask = Proxy.newProxyInstance( + ScheduledTask::class.java.classLoader, + arrayOf(ScheduledTask::class.java), + ) { proxy, method, args -> + when (method.name) { + "cancel" -> { + cancelCount++ + cancelFailure?.let { throw it } + null + } + "toString" -> "RecordedScheduledTask" + "hashCode" -> System.identityHashCode(proxy) + "equals" -> proxy === args?.get(0) + else -> null + } + } as ScheduledTask + } + + private class RejectingExecutorService(private val failure: RejectedExecutionException) : AbstractExecutorService() { + + override fun shutdown() = Unit + + override fun shutdownNow(): MutableList = ArrayList() + + override fun isShutdown(): Boolean = false + + override fun isTerminated(): Boolean = false + + override fun awaitTermination(timeout: Long, unit: TimeUnit): Boolean = false + + override fun execute(command: Runnable) { + throw failure + } + } + + private class DirectExecutorService : AbstractExecutorService() { + + private var shutdown = false + var shutdownNowCount = 0 + + override fun shutdown() { + shutdown = true + } + + override fun shutdownNow(): MutableList { + shutdown = true + shutdownNowCount++ + return ArrayList() + } + + override fun isShutdown(): Boolean = shutdown + + override fun isTerminated(): Boolean = shutdown + + override fun awaitTermination(timeout: Long, unit: TimeUnit): Boolean = shutdown + + override fun execute(command: Runnable) { + if (shutdown) { + throw RejectedExecutionException() + } + command.run() + } + } +} diff --git a/platform/platform-velocity/build.gradle.kts b/platform/platform-velocity/build.gradle.kts index 941a4099f..37705bbed 100644 --- a/platform/platform-velocity/build.gradle.kts +++ b/platform/platform-velocity/build.gradle.kts @@ -6,4 +6,8 @@ dependencies { compileOnly(project(":common")) compileOnly(project(":common-platform-api")) compileOnly("com.velocitypowered:velocity-api:3.1.1") + + testImplementation(project(":common")) + testImplementation(project(":common-platform-api")) + testImplementation("com.velocitypowered:velocity-api:3.1.1") } \ No newline at end of file diff --git a/platform/platform-velocity/src/main/java/taboolib/platform/VelocityActivationGate.java b/platform/platform-velocity/src/main/java/taboolib/platform/VelocityActivationGate.java new file mode 100644 index 000000000..5ea84cf2e --- /dev/null +++ b/platform/platform-velocity/src/main/java/taboolib/platform/VelocityActivationGate.java @@ -0,0 +1,39 @@ +package taboolib.platform; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Coordinates zero-delay activation with shutdown without blocking either thread. + */ +final class VelocityActivationGate { + + private enum State { + OPEN, + ACTIVATING, + CLOSED + } + + private final AtomicReference state = new AtomicReference<>(State.OPEN); + private final CompletableFuture activationClosed = new CompletableFuture<>(); + + boolean activate(Runnable action) { + if (!state.compareAndSet(State.OPEN, State.ACTIVATING)) { + return false; + } + try { + action.run(); + return true; + } finally { + state.set(State.CLOSED); + activationClosed.complete(null); + } + } + + CompletableFuture close() { + if (state.compareAndSet(State.OPEN, State.CLOSED)) { + activationClosed.complete(null); + } + return activationClosed; + } +} diff --git a/platform/platform-velocity/src/main/java/taboolib/platform/VelocityPlugin.java b/platform/platform-velocity/src/main/java/taboolib/platform/VelocityPlugin.java index 743a6ca12..5f55f6c76 100644 --- a/platform/platform-velocity/src/main/java/taboolib/platform/VelocityPlugin.java +++ b/platform/platform-velocity/src/main/java/taboolib/platform/VelocityPlugin.java @@ -1,6 +1,7 @@ package taboolib.platform; import com.google.inject.Inject; +import com.velocitypowered.api.event.EventTask; import com.velocitypowered.api.event.Subscribe; import com.velocitypowered.api.event.proxy.ProxyInitializeEvent; import com.velocitypowered.api.event.proxy.ProxyShutdownEvent; @@ -19,6 +20,8 @@ import taboolib.common.platform.Plugin; import java.nio.file.Path; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; import static taboolib.common.PrimitiveIO.t; @@ -84,6 +87,8 @@ public class VelocityPlugin { private final ProxyServer server; private final Logger logger; private final Path configDirectory; + private final VelocityActivationGate activationGate = new VelocityActivationGate(); + private final AtomicReference> disableFuture = new AtomicReference<>(); @Inject public VelocityPlugin(final ProxyServer server, final Logger logger, @DataDirectory final Path configDirectory) { @@ -116,25 +121,99 @@ public void e(ProxyInitializeEvent e) { // 因为插件可能在 onEnable() 下关闭 if (!TabooLib.isStopped()) { // 创建调度器，执行 onActive() 方法 - server.getScheduler().buildTask(this, () -> { + server.getScheduler().buildTask(this, () -> activationGate.activate(() -> { + if (TabooLib.isStopped()) { + return; + } // 生命周期任务 TabooLib.lifeCycle(LifeCycle.ACTIVE); // 调用 Plugin 实现的 onActive() 方法 if (pluginInstance != null) { pluginInstance.onActive(); } - }).schedule(); + })).schedule(); } } - @Subscribe + /** + * 保留旧同步入口；该入口无法向调用方表达异步完成，只负责观察失败。 + */ public void e(ProxyShutdownEvent e) { + observeDisable(disableAfterActivation()); + } + + @Subscribe + public EventTask eAsync(ProxyShutdownEvent e) { + return EventTask.resumeWhenComplete(disableAfterActivation()); + } + + private CompletableFuture disableAfterActivation() { + CompletableFuture current = disableFuture.get(); + if (current != null) { + return current; + } + CompletableFuture created = new CompletableFuture<>(); + if (!disableFuture.compareAndSet(null, created)) { + return disableFuture.get(); + } + activationGate.close().whenComplete((unused, failure) -> { + if (failure != null) { + created.completeExceptionally(failure); + return; + } + try { + disable(); + created.complete(null); + } catch (Throwable ex) { + created.completeExceptionally(ex); + } + }); + return created; + } + + private void observeDisable(CompletableFuture future) { + future.whenComplete((unused, failure) -> { + if (failure != null) { + try { + logger.error("Failed to disable the TabooLib Velocity plugin", failure); + } catch (Throwable ignored) { + try { + failure.printStackTrace(); + } catch (Throwable ignoredAgain) { + } + } + } + }); + } + + private void disable() { + Throwable failure = null; // 在插件未关闭的前提下，执行 onDisable() 方法 if (pluginInstance != null && !TabooLib.isStopped()) { - pluginInstance.onDisable(); + try { + pluginInstance.onDisable(); + } catch (Throwable ex) { + failure = ex; + } } - // 生命周期任务 - TabooLib.lifeCycle(LifeCycle.DISABLE); + // 生命周期任务必须执行，不能被用户回调异常跳过 + try { + TabooLib.lifeCycle(LifeCycle.DISABLE); + } catch (Throwable ex) { + if (failure == null) { + failure = ex; + } else { + failure.addSuppressed(ex); + } + } + if (failure != null) { + VelocityPlugin.rethrow(failure); + } + } + + @SuppressWarnings("unchecked") + private static void rethrow(Throwable throwable) throws T { + throw (T) throwable; } @Nullable diff --git a/platform/platform-velocity/src/main/kotlin/taboolib/platform/type/VelocityProxyEvent.kt b/platform/platform-velocity/src/main/kotlin/taboolib/platform/type/VelocityProxyEvent.kt index 02b98a71c..c83a588dd 100644 --- a/platform/platform-velocity/src/main/kotlin/taboolib/platform/type/VelocityProxyEvent.kt +++ b/platform/platform-velocity/src/main/kotlin/taboolib/platform/type/VelocityProxyEvent.kt @@ -2,12 +2,15 @@ package taboolib.platform.type import com.velocitypowered.api.event.ResultedEvent import com.velocitypowered.api.event.ResultedEvent.GenericResult +import org.slf4j.LoggerFactory import taboolib.common.PrimitiveIO.t import taboolib.platform.VelocityPlugin +import java.util.concurrent.CompletableFuture import java.util.function.Consumer open class VelocityProxyEvent : ResultedEvent { + @Volatile private var isCancelled = false private val cancelCallbacks = mutableListOf>>() @@ -39,8 +42,58 @@ open class VelocityProxyEvent : ResultedEvent { return this } + /** + * 调用事件，并在所有监听器完成后返回事件是否未被取消。 + */ + fun callAsync(): CompletableFuture { + return fireEvent().thenApply { !isCancelled } + } + + /** + * 调用事件但不等待异步监听器。 + * + * 若事件已同步完成，则返回最终状态；否则返回调用时可见的取消状态快照。 + */ fun call(): Boolean { - VelocityPlugin.getInstance().server.eventManager.fire(this) - return !isCancelled + val future = fireEvent() + val snapshot = !isCancelled + future.whenComplete { _, throwable -> + if (throwable != null) { + reportCallFailure(throwable) + } + } + return if (future.isDone) !isCancelled else snapshot + } + + /** + * 为测试保留的事件派发接缝。 + */ + protected open fun fireEvent(): CompletableFuture { + return VelocityPlugin.getInstance().server.eventManager.fire(this) + } + + private fun reportCallFailure(throwable: Throwable) { + try { + onCallFailure(throwable) + } catch (reportingFailure: Throwable) { + throwable.addSuppressed(reportingFailure) + try { + LoggerFactory.getLogger(VelocityProxyEvent::class.java) + .error("Failed to report an asynchronous Velocity event failure", throwable) + } catch (fallbackFailure: Throwable) { + throwable.addSuppressed(fallbackFailure) + try { + throwable.printStackTrace() + } catch (_: Throwable) { + } + } + } + } + + /** + * 兼容调用无法向调用方传播异步异常，因此至少将其记录下来。 + */ + protected open fun onCallFailure(throwable: Throwable) { + VelocityPlugin.getInstance().logger.error("Failed to fire Velocity event ${javaClass.name}", throwable) } } \ No newline at end of file diff --git a/platform/platform-velocity/src/test/java/taboolib/platform/VelocityActivationGateTest.java b/platform/platform-velocity/src/test/java/taboolib/platform/VelocityActivationGateTest.java new file mode 100644 index 000000000..12d4d0b43 --- /dev/null +++ b/platform/platform-velocity/src/test/java/taboolib/platform/VelocityActivationGateTest.java @@ -0,0 +1,73 @@ +package taboolib.platform; + +import com.velocitypowered.api.event.EventTask; +import com.velocitypowered.api.event.Subscribe; +import com.velocitypowered.api.event.proxy.ProxyShutdownEvent; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class VelocityActivationGateTest { + + @Test + void shutdownClosesGateBeforeLateActivation() { + VelocityActivationGate gate = new VelocityActivationGate(); + AtomicInteger activeCalls = new AtomicInteger(); + + CompletableFuture closed = gate.close(); + + assertTrue(closed.isDone()); + assertFalse(gate.activate(activeCalls::incrementAndGet)); + assertEquals(0, activeCalls.get()); + } + + @Test + void disableContinuationWaitsForClaimedActivation() { + VelocityActivationGate gate = new VelocityActivationGate(); + List order = new ArrayList<>(); + AtomicReference> closed = new AtomicReference<>(); + + assertTrue(gate.activate(() -> { + order.add("active-start"); + closed.set(gate.close()); + assertFalse(closed.get().isDone()); + closed.get().thenRun(() -> order.add("disable")); + order.add("active-end"); + })); + + assertTrue(closed.get().isDone()); + assertEquals(Arrays.asList("active-start", "active-end", "disable"), order); + } + + @Test + void activationCanOnlyBeClaimedOnce() { + VelocityActivationGate gate = new VelocityActivationGate(); + AtomicInteger activeCalls = new AtomicInteger(); + + assertTrue(gate.activate(activeCalls::incrementAndGet)); + assertFalse(gate.activate(activeCalls::incrementAndGet)); + assertTrue(gate.close().isDone()); + assertEquals(1, activeCalls.get()); + } + + @Test + void shutdownKeepsLegacyDescriptorAndUsesAsyncEventContract() throws NoSuchMethodException { + java.lang.reflect.Method legacy = VelocityPlugin.class.getDeclaredMethod("e", ProxyShutdownEvent.class); + java.lang.reflect.Method async = VelocityPlugin.class.getDeclaredMethod("eAsync", ProxyShutdownEvent.class); + + assertEquals(void.class, legacy.getReturnType()); + assertNull(legacy.getAnnotation(Subscribe.class)); + assertEquals(EventTask.class, async.getReturnType()); + assertTrue(async.isAnnotationPresent(Subscribe.class)); + } +} diff --git a/platform/platform-velocity/src/test/kotlin/taboolib/platform/type/VelocityProxyEventTest.kt b/platform/platform-velocity/src/test/kotlin/taboolib/platform/type/VelocityProxyEventTest.kt new file mode 100644 index 000000000..287a6a00e --- /dev/null +++ b/platform/platform-velocity/src/test/kotlin/taboolib/platform/type/VelocityProxyEventTest.kt @@ -0,0 +1,125 @@ +package taboolib.platform.type + +import com.velocitypowered.api.event.ResultedEvent.GenericResult +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionException + +class VelocityProxyEventTest { + + @Test + fun `callAsync completes with final cancellation state`() { + val fired = CompletableFuture() + val event = TestEvent(fired) + + val result = event.callAsync() + + assertEquals(1, event.fireCount) + assertFalse(result.isDone) + + event.result = GenericResult.denied() + fired.complete(event) + + assertFalse(result.getNow(true)) + } + + @Test + fun `callAsync propagates exceptional completion`() { + val fired = CompletableFuture() + val event = TestEvent(fired) + val failure = IllegalStateException("fire failed") + + val result = event.callAsync() + fired.completeExceptionally(failure) + + val thrown = assertThrows(CompletionException::class.java) { + result.getNow(true) + } + assertSame(failure, thrown.cause) + } + + @Test + fun `call returns current snapshot without waiting for unfinished fire`() { + val fired = CompletableFuture() + val event = TestEvent(fired) + + assertTrue(event.call()) + assertEquals(1, event.fireCount) + assertFalse(fired.isDone) + + event.result = GenericResult.denied() + fired.complete(event) + } + + @Test + fun `call returns final state when fire completes synchronously`() { + lateinit var event: TestEvent + event = TestEvent { + event.result = GenericResult.denied() + CompletableFuture.completedFuture(event) + } + + assertFalse(event.call()) + assertEquals(1, event.fireCount) + } + + @Test + fun `call observes later asynchronous failure`() { + val fired = CompletableFuture() + val event = TestEvent(fired) + val failure = IllegalArgumentException("listener failed") + + assertTrue(event.call()) + fired.completeExceptionally(failure) + + assertSame(failure, event.observedFailure) + } + + @Test + fun `call contains reporter failures without losing original failure`() { + val fired = CompletableFuture() + val reporterFailure = IllegalStateException("reporter failed") + val event = TestEvent(fired).also { it.reporterFailure = reporterFailure } + val failure = IllegalArgumentException("listener failed") + + assertTrue(event.call()) + fired.completeExceptionally(failure) + + assertSame(failure, event.observedFailure) + assertEquals(listOf(reporterFailure), failure.suppressed.toList()) + } + + @Test + fun `call keeps primitive boolean JVM signature`() { + val method = VelocityProxyEvent::class.java.getDeclaredMethod("call") + + assertEquals(Boolean::class.javaPrimitiveType, method.returnType) + assertEquals(0, method.parameterCount) + } + + private class TestEvent( + private val fire: () -> CompletableFuture + ) : VelocityProxyEvent() { + + constructor(future: CompletableFuture) : this({ future }) + + var fireCount = 0 + var observedFailure: Throwable? = null + var reporterFailure: Throwable? = null + + override fun fireEvent(): CompletableFuture { + fireCount++ + return fire() + } + + override fun onCallFailure(throwable: Throwable) { + observedFailure = throwable + reporterFailure?.let { throw it } + } + } +}