/**
 * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
 * <p>
 * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.gnu.org/licenses/lgpl-3.0.txt
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.easyagents.flow.core.test;

import com.easyagents.flow.core.chain.EventManager;
import com.easyagents.flow.core.chain.event.ChainEndEvent;
import com.easyagents.flow.core.chain.listener.ChainEventListener;
import org.junit.Assert;
import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * {@link EventManager} 并发行为测试。
 */
public class EventManagerConcurrencyTest {

    /**
     * 验证事件分发期间可以从其他线程移除监听器，当前分发使用稳定快照，后续分发不再调用已移除监听器。
     *
     * @throws Exception 等待测试线程超时或中断时抛出
     */
    @Test
    public void shouldAllowListenerRemovalDuringEventDispatch() throws Exception {
        EventManager eventManager = new EventManager();
        CountDownLatch dispatchStarted = new CountDownLatch(1);
        CountDownLatch continueDispatch = new CountDownLatch(1);
        AtomicInteger retainedInvocationCount = new AtomicInteger();
        AtomicInteger removedInvocationCount = new AtomicInteger();
        AtomicReference<Throwable> dispatchFailure = new AtomicReference<>();

        ChainEventListener blockingListener = (event, chain) -> {
            dispatchStarted.countDown();
            try {
                if (!continueDispatch.await(3, TimeUnit.SECONDS)) {
                    throw new AssertionError("事件分发等待超时");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new AssertionError("事件分发线程被中断", e);
            }
        };
        ChainEventListener retainedListener = (event, chain) -> retainedInvocationCount.incrementAndGet();
        ChainEventListener removedListener = (event, chain) -> removedInvocationCount.incrementAndGet();
        eventManager.addEventListener(blockingListener);
        eventManager.addEventListener(retainedListener);
        eventManager.addEventListener(removedListener);

        Thread dispatchThread = new Thread(() -> {
            try {
                eventManager.notifyEvent(new ChainEndEvent(null), null);
            } catch (Throwable error) {
                dispatchFailure.set(error);
            }
        }, "event-manager-dispatch-test");
        dispatchThread.start();

        try {
            Assert.assertTrue("事件分发应在测试超时前开始", dispatchStarted.await(3, TimeUnit.SECONDS));
            eventManager.removeEventListener(removedListener);
        } finally {
            continueDispatch.countDown();
        }

        dispatchThread.join(3000L);
        Assert.assertFalse("事件分发线程应正常结束", dispatchThread.isAlive());
        Assert.assertNull("并发移除监听器不应中断事件分发", dispatchFailure.get());
        Assert.assertEquals(1, retainedInvocationCount.get());
        Assert.assertEquals("当前事件应按分发开始时的快照通知", 1, removedInvocationCount.get());

        eventManager.notifyEvent(new ChainEndEvent(null), null);
        Assert.assertEquals(2, retainedInvocationCount.get());
        Assert.assertEquals("后续事件不应通知已移除监听器", 1, removedInvocationCount.get());
    }
}
