package io.meteorcat.game;

import io.quarkus.runtime.Quarkus;
import io.quarkus.runtime.annotations.QuarkusMain;

/**
 * 项目启动入口
 * <p>
 * 参考文档: <a href="https://cn.quarkus.io/guides/lifecycle">应用程序的初始化和终止</a>
 */
@QuarkusMain
public class PinoApplication {

    /**
     * 程序启动入口
     *
     * @param args 参数列表
     */
    public static void main(String[] args) {
        Quarkus.run(args);
    }
}

package io.meteorcat.game.config;

import com.google.protobuf.MessageLite;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import io.smallrye.config.ConfigMapping;
import io.smallrye.config.WithDefault;
import org.apache.pekko.Done;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import scala.util.Try;

import java.util.Map;
import java.util.function.Function;

/**
 * Pekko Actor 系统的配置器接口，用于标准化 Actor 系统的初始化流程（配置加载、日志适配、自定义参数注入）。
 * <p>核心作用：
 * 1. 封装 Pekko 配置的加载逻辑（支持多类型默认配置切换）；
 * 2. 自动适配 Quarkus 日志系统（复用 Slf4j 日志配置，避免多日志系统冲突）；
 * 3. 支持外部自定义配置注入（通过 settings 传递 Pekko 配置项）；
 * 4. 提供 Actor 系统创建的默认实现和终止回调扩展。
 * <p>使用场景：Quarkus 环境下 Pekko Actor 系统的初始化，需配合 SmallRye Config 实现配置注入（如 @WithDefault 注解）。
 */
@SuppressWarnings("unused")
@ConfigMapping(prefix = "actor")
public interface ActorPropsConfigurator {

    /**
     * Pekko 配置加载类型枚举，对应 {@link ConfigFactory} 提供的四种默认配置加载方式，用于切换不同场景的基础配置。
     * <p>枚举值与 ConfigFactory 方法的映射关系：
     * - Reference: 加载 Pekko 框架默认参考配置（{@link ConfigFactory#defaultReference()}）；
     * - Application: 加载应用自定义配置（{@link ConfigFactory#defaultApplication()}）；
     * - Overrides: 加载覆盖配置（优先级最高，用于临时覆盖默认配置，{@link ConfigFactory#defaultOverrides()}）；
     * - ReferenceUnresolved: 加载未解析的默认参考配置（需手动调用 resolve()，{@link ConfigFactory#defaultReferenceUnresolved()}）。
     */
    enum ActorReferences {
        /**
         * 对应 {@link ConfigFactory#defaultReference()}，加载 Pekko 框架内置的默认参考配置
         */
        Reference,
        /**
         * 对应 {@link ConfigFactory#defaultApplication()}，加载应用级自定义配置（如 application.conf）
         */
        Application,
        /**
         * 对应 {@link ConfigFactory#defaultOverrides()}，加载覆盖配置（优先级高于 Reference 和 Application）
         */
        Overrides,
        /**
         * 对应 {@link ConfigFactory#defaultReferenceUnresolved()}，加载未解析的默认参考配置（需后续手动解析）
         */
        ReferenceUnresolved
    }

    /**
     * 获取 Pekko 配置加载类型（通过 SmallRye Config 注入，支持配置文件指定）。
     * <p>默认值为 {@link ActorReferences#Reference}，即默认加载 Pekko 框架的默认参考配置；
     * 可通过配置文件（如 application.properties）指定：{@code io.fortress.actor.configurator.reference=application} 切换为应用配置。
     *
     * @return 配置加载类型，非 null
     */
    @WithDefault("reference")
    ActorReferences reference();


    /**
     * 获取 Actor 系统的名称（用于创建 Actor 系统实例的唯一标识）。
     * <p>最终会传递给 {@link ActorSystem<MessageLite>#create(String, Config)} 的第一个参数，作为 Actor 系统的全局名称；
     * 建议通过配置文件注入，确保不同环境（开发/测试/生产）可灵活配置。
     *
     * @return Actor 系统名称，非空字符串
     */
    String name();

    /**
     * 获取 Pekko 自定义配置集合（用于注入额外的 Pekko 配置项，覆盖默认配置）。
     * <p>配置项格式要求：
     * - Key: 需符合 Pekko 配置路径（如 "pekko.actor.serializers.proto"）；
     * - Value: 配置值（如 "org.apache.pekko.remote.serialization.ProtobufSerializer"）；
     * <p>使用场景：外部通过配置文件或代码注入 Pekko 特定配置（如序列化器、线程池参数、远程通信配置等）。
     *
     * @return 自定义配置键值对，可为空（无额外配置时返回空 Map）
     */
    Map<String, String> settings();


    /**
     * 默认实现：创建 Pekko Actor 系统实例（核心初始化方法，封装完整配置流程）。
     * <p>初始化流程：
     * 1. 加载基础配置：通过 {@link #reference()} 指定的类型加载默认配置；
     * 2. 适配 Quarkus 日志：自动注入 Slf4j 日志相关配置，复用 Quarkus 日志系统（避免 Pekko 自带日志与 Quarkus 冲突）；
     * 3. 合并自定义配置：将 {@link #settings()} 中的配置项合并到基础配置，覆盖默认值；
     * 4. 创建 Actor 系统：使用合并后的配置和 {@link #name()} 生成的系统名，创建 ActorSystem 实例。
     *
     * @return 初始化完成的 Pekko Actor 系统实例，非 null
     */
    default ActorSystem<MessageLite> createActorSystem() {

        // 1. 加载应用默认基础配置（如 application.conf 中的配置）
        Config originalConfig = ConfigFactory.load();


        // 2. 根据 reference() 选择的类型，合并对应的 Pekko 默认配置（优先级：Overrides > Application > Reference）
        originalConfig = switch (reference()) {
            case Application -> originalConfig.withFallback(ConfigFactory.defaultApplication());
            case Overrides -> originalConfig.withFallback(ConfigFactory.defaultOverrides());
            case ReferenceUnresolved -> originalConfig.withFallback(ConfigFactory.defaultReferenceUnresolved());
            // 默认使用 Pekko 框架的默认参考配置
            default -> originalConfig.withFallback(ConfigFactory.defaultReference());
        };

        // 3. 获取外部注入的自定义配置（settings()），适配 Quarkus 日志系统
        Map<String, String> settings = settings();
        // 若未配置 Pekko 日志过滤器，自动注入 Slf4j 日志相关配置（复用 Quarkus 日志，避免多日志实现冲突）
        if (!settings.containsKey("pekko.logging-filter")) {
            settings.put("pekko.logging-filter", "org.apache.pekko.event.slf4j.Slf4jLoggingFilter"); // 日志过滤器：使用 Slf4j 实现
            settings.put("pekko.loggers.0", "org.apache.pekko.event.slf4j.Slf4jLogger"); // 日志实现：Slf4j  logger
            settings.put("pekko.log-dead-letters", "off"); // 关闭死信日志（避免开发环境日志冗余）
            settings.put("pekko.log-dead-letters-during-shutdown", "off"); // 关闭阶段的死信日志
        }


        // 4. 合并自定义配置到基础配置（自定义配置优先级最高，覆盖默认配置）
        Config mergedConfig = ConfigFactory
                .parseMap(settings)
                .withFallback(originalConfig)
                .resolve();

        // 5. 检查是否配置正确的 Pekko 系统名称
        if (Objects.isNull(name()) || name().isBlank()) {
            throw new IllegalArgumentException("Pekko ActorSystem name cannot be empty! Please configure 'pekko.name'");
        }


        // 6. 注册 JVM 关闭钩子，保证应用退出时 ActorSystem 优雅关闭
        ActorSystem<MessageLite> system = ActorSystem.create(Behaviors.empty(), name(), mergedConfig);
        Runtime.getRuntime().addShutdownHook(new Thread(system::terminate));
        return system;
    }


    /**
     * 扩展实现：创建 Actor 系统并绑定终止回调（支持在 Actor 系统终止时执行自定义逻辑）。
     * <p>适用场景：需要监听 Actor 系统生命周期的场景（如资源释放、日志记录、告警通知等）；
     * 回调函数会接收 Actor 系统终止的结果（{@link Try<Done>}），成功终止时 Try 为 Success，异常时为 Failure。
     *
     * @param closeable Actor 系统终止时的回调函数，入参为终止结果（Try<Done>），返回值无特定含义（可忽略）
     * @param <U>       回调函数返回值类型（泛型，无实际约束）
     * @return 初始化完成的 Actor 系统实例（已绑定终止回调），非 null
     */
    default <U> ActorSystem<MessageLite> createActorSystem(Function<Try<Done>, U> closeable) {
        // 1. 复用默认方法创建 Actor 系统
        ActorSystem<MessageLite> actorSystem = createActorSystem();
        // 2. 绑定终止回调：当 Actor 系统终止时，执行 closeable 函数（使用 Actor 系统的调度器线程池）
        actorSystem.whenTerminated().onComplete(closeable::apply, actorSystem.executionContext());
        return actorSystem;
    }

}

package io.meteorcat.game;

import com.google.protobuf.MessageLite;
import io.meteorcat.game.config.ActorPropsConfigurator;
import io.quarkus.arc.DefaultBean;
import io.quarkus.runtime.Quarkus;
import io.quarkus.runtime.Startup;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Produces;
import jakarta.inject.Inject;
import org.apache.pekko.Done;
import org.apache.pekko.actor.typed.ActorSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 生成全局持久化唯一的 ActorSystem<br>
 * '@ApplicationScoped' 会默认将该类实例化并加入容器托管
 * 后续通过 '@Inject ActorSystem<MessageLite> system' 就能全局获取到对应的 ActorSystem 句柄
 */
@ApplicationScoped
public class ActorSystemApplication {

    /**
     * 日志对象
     */
    private final Logger logger = LoggerFactory.getLogger(this.getClass());


    /**
     * 加载 ActorSystem 配置
     */
    @Inject
    ActorPropsConfigurator configurator;


    /**
     * 生成全局唯一Actor系统
     * DefaultBean 设置为全局默认 ActorSystem 的 Bean
     * Startup 代表优先启动加载
     * Produces 代表对象生成方法
     * ApplicationScoped 代表启动为系统应用级别
     */
    @Produces
    @Startup
    @DefaultBean
    @ApplicationScoped
    public ActorSystem<MessageLite> actorSystem() {
        // 打印配置
        logger.info("Creating ActorSystem: {}", configurator.name());
        configurator.settings().forEach((key, value) -> logger.info("ActorSystem Setting: {} = {}", key, value));

        // 生成 Actor System 并注册系统退出方法, 当 Actor 系统退出的时候要求关闭整个服务
        return configurator.createActorSystem((terminatedTry -> {
            // 无论成功失败都关闭 Quarkus 系统
            logger.info("Close ActorSystem, Status: {}", terminatedTry.isSuccess() ? "Success" : "Failure");
            Quarkus.asyncExit();
            return Done.getInstance();
        }));
    }
}

package io.meteorcat.game.utils;


import io.quarkus.arc.Arc;
import io.quarkus.arc.ArcContainer;
import jakarta.enterprise.inject.Instance;
import jakarta.enterprise.inject.spi.BeanManager;
import jakarta.inject.Inject;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.util.Optional;

/**
 * Bean容器注入
 * 对 Actor 实例化的句柄注入 Quarkus 的全局实例化对象
 * 需要的该功能的 Actor 只需要组合继承即可
 */
public interface ActorBeanContainer {

    /**
     * 日志对象
     */
    Logger LOG = LoggerFactory.getLogger(ActorBeanContainer.class);

    /**
     * 通过反射为 Actor 注入 @Inject 注解的字段（兼容 CDI 上下文 + 异常安全）
     *
     * @param actor Pekko Actor 实例
     * @param <T>   Actor 消息类型
     * @throws IllegalAccessException 反射访问字段失败时抛出
     */
    default <T> void injectFields(AbstractBehavior<T> actor) throws IllegalAccessException {
        final ArcContainer arcContainer = Arc.container();
        final BeanManager beanManager = arcContainer.beanManager();
        final Instance<Object> cdiInstance = beanManager.createInstance();

        // 递归扫描 Actor 类继承链的所有字段
        Class<?> currentClass = actor.getClass();
        while (currentClass != Object.class && currentClass != AbstractBehavior.class) {
            Field[] fields = currentClass.getDeclaredFields();
            for (Field field : fields) {
                if (field.isAnnotationPresent(Inject.class)) {
                    injectSingleField(actor, field, cdiInstance);
                }
            }
            currentClass = currentClass.getSuperclass();
        }
    }

    /**
     * 注入单个 @Inject 字段（异常安全 + 类型匹配）
     *
     * @param actor       Actor 实例
     * @param field       待注入字段
     * @param cdiInstance CDI Instance 实例
     * @throws IllegalAccessException 反射赋值失败
     */
    private <T> void injectSingleField(AbstractBehavior<T> actor, Field field, Instance<Object> cdiInstance) throws IllegalAccessException {
        Class<?> fieldType = field.getType();
        field.setAccessible(true);

        // 安全获取 Bean：避免找不到 Bean 导致崩溃
        Optional<Object> dependency = Optional.ofNullable(
                cdiInstance.select(fieldType).get()
        );

        if (dependency.isPresent()) {
            field.set(actor, dependency.get());
            LOG.debug("Successfully injected field[{}] of type[{}] into Actor[{}]", field.getName(), fieldType.getName(), actor.getClass().getName());
        } else {
            LOG.warn("CDI Bean for field[{}] (type: {}) not found, skipping injection", field.getName(), fieldType.getName());
        }
    }
}

package io.meteorcat.game.actor;

import com.google.protobuf.MessageLite;
import io.meteorcat.game.config.ActorPropsConfigurator;
import io.meteorcat.game.utils.ActorBeanContainer;
import jakarta.inject.Inject;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.PostStop;
import org.apache.pekko.actor.typed.Terminated;
import org.apache.pekko.actor.typed.javadsl.AbstractBehavior;
import org.apache.pekko.actor.typed.javadsl.ActorContext;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.actor.typed.javadsl.Receive;
import org.slf4j.Logger;

import java.util.Objects;

/**
 * 客户端在服务端之中的 '操作代理'
 * 客户端通过 TCP/UDP/WebSocket 等连接上来的会被 ActorSystem 动态构建单独 Actor
 * 所以消息转发过滤都是经由此 Actor 检测过滤最终确认是转发还是执行本地逻辑
 */
public class AgentActor extends AbstractBehavior<MessageLite> implements ActorBeanContainer {

    /**
     * 日志对象
     */
    final Logger logger = getContext().getLog();

    /**
     * 确认是否能够接收到被注入的 Actor 全局配置
     * 如果成功代表 Quarkus 已经被顺利托管进来
     */
    @Inject
    ActorPropsConfigurator configurator;

    /**
     * 默认构造方法
     */
    private AgentActor(ActorContext<MessageLite> context) {
        super(context);
    }

    /**
     * 动态构建
     */
    public static Behavior<MessageLite> create() {
        return Behaviors.setup((ctx) -> {
            AgentActor agent = new AgentActor(ctx);
            agent.injectFields(agent); // 自己做依赖注册
            return agent;
        });
    }

    /**
     * 消息拦截
     */
    @Override
    public Receive<MessageLite> createReceive() {
        // 构建信号拦截
        return newReceiveBuilder()
                .onSignal(PostStop.class, (ignore) -> {
                    logger.info("Actor Stop!");

                    // 退出的时候打印注入配置
                    if (Objects.isNull(configurator)) {
                        logger.warn("Actor configurator is null!");
                    } else {
                        logger.warn("Actor configurator has been initialized!");
                    }

                    return Behaviors.same();
                })
                .onSignal(Terminated.class, (ignore) -> {
                    logger.info("Actor Terminated!");
                    return Behaviors.same();
                })
                .build();
    }
}

package io.meteorcat.game.actor;

import com.google.protobuf.MessageLite;
import io.quarkus.test.junit.QuarkusTest;
import jakarta.inject.Inject;
import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.PostStop;
import org.apache.pekko.actor.typed.Props;
import org.apache.pekko.testkit.javadsl.TestKit;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.UUID;

import static org.junit.jupiter.api.Assertions.*;

/**
 * AgentActor 测试单元入口
 */
@QuarkusTest
class AgentActorTest {


    /**
     * 获取全局挂载的 ActorSystem
     */
    @Inject
    ActorSystem<MessageLite> system;


    /**
     * 确认 Actor 名称
     */
    @Test
    void testSystemInfo() {
        assertNotNull(system);
        final Logger logger = system.log();
        logger.info("Actor System Name: {}", system.name());
    }


    /**
     * 测试启动 AgentActor
     */
    @Test
    void testStartAgentActor() {
        assertNotNull(system);
        TestKit testKit = new TestKit(system.classicSystem());

        // 创建 Actor
        String name = "agent-%s".formatted(UUID.randomUUID().toString());
        ActorRef<MessageLite> actorRef = system.systemActorOf(AgentActor.create(), name, Props.empty());

        // 延迟任务, 让 actor 退出
        system.scheduler().scheduleOnce(
                Duration.ofSeconds(2),
                () -> actorRef.unsafeUpcast().tell(PostStop.instance()),
                system.executionContext()
        );

        // 等待3s 让 Actor 反映之后退出
        testKit.expectNoMessage(Duration.ofSeconds(3));
    }

}