package io.meteorcat.game;

import org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.Behavior;
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;

/**
 * 实现数据打印功能的 Actor
 * AbstractBehavior: 就是声明该类是作为 Actor 继承
 * String: 则是代表内部做消息传递是采用 String(后续可以自定义结构体来交换数据)
 */
class PrintMyActorRefActor extends AbstractBehavior<String> {

    /**
     * 静态创建 Actor 并且利用 Behaviors.setup 初始化在 Actor 系统之中等待挂载
     * 注意: 调用该静态生成的实例化对象是 '等待挂载中' 的状态 
     */
    static Behavior<String> create() {
        return Behaviors.setup(PrintMyActorRefActor::new);
    }

    /**
     * 如果是声明为 Actor 类, 最好不允许外部做实例化, 而是要通过 Behaviors.setup 生成实例
     */
    private PrintMyActorRefActor(ActorContext<String> context) {
        super(context);
    }

    /**
     * Actor 必须实现的消息拦截方法
     * 拦截的消息就是 AbstractBehavior<T> 之中的 T 类型, 对于强类型来说
     */
    @Override
    public Receive<String> createReceive() {
        return newReceiveBuilder().onMessageEquals("printit", this::printIt).build();
    }

    /**
     * 自定义内部 Actor 功能
     * 注意调用之后都需要返回 Behavior<T> 来让 Actor 保持当前 Actor 状态
     */
    private Behavior<String> printIt() {
        // getContext() 是 Actor 内置方法, 用于获取 Actor 上下文
        // getContext().spawn() 调用 Actor 上下文动态在当前 Actor 节点额外创建 Actor 并且自动挂载
        // 其中 Behaviors.empty() 代表生成默认的为空(没有自定义) 的 Actor, 如果想指定自定义则是 Behaviors.setup(自定义类实例)
        // "second-actor" 则是指定挂载于当前 Actor 节点下的名称
        ActorRef<String> secondRef = getContext().spawn(Behaviors.empty(), "second-actor");
        System.out.println("Second: " + secondRef);
        return this;
    }
}

/**
 * 根节点下的第一节点 Actor
 */
class Main extends AbstractBehavior<String> {

    /**
     * 静态创建 Actor 并且利用 Behaviors.setup 初始化在 Actor 系统之中等待挂载
     * 注意: 调用该静态生成的实例化对象是 '等待挂载中' 的状态 
     */
    static Behavior<String> create() {
        return Behaviors.setup(Main::new);
    }

    private Main(ActorContext<String> context) {
        super(context);
    }

    @Override
    public Receive<String> createReceive() {
        return newReceiveBuilder().onMessageEquals("start", this::start).build();
    }

    /**
     * 自定义拦截方法
     * 这里则是动态创建自定义 Actor并且转发消息
     * Behaviors.same() 代表保持消息, 其实就是和返回 this 一致
     */
    private Behavior<String> start() {
        ActorRef<String> firstRef = getContext().spawn(PrintMyActorRefActor.create(), "first-actor");

        System.out.println("First: " + firstRef);
        firstRef.tell("printit");
        return Behaviors.same();
    }
}


/**
 * 全局启动入口类
 */
public class ActorHierarchyExperiments {
    /**
     * 全局启动入口方法
     */
    public static void main(String[] args) {
        // ActorSystem.create 代表创建 Actor 系统并且声明 Main.create() 构建的实例
        // ActorRef 就是 ActorSystem 创建 Actor 之后的引用地址, Actor 的消息都是引用地址做消息传递
        // 其中 testSystem 就是 Root Actor 的之下名称
        ActorRef<String> testSystem = ActorSystem.create(Main.create(), "testSystem");
        testSystem.tell("start");// 推送指定 String 类型消息给 actor 系统
    }
}

/**
 * 带参数的实例化构建 Actor 
 */
class PrintParamsActor extends AbstractBehavior<String> {

    final String message;


    private PrintParamsActor(ActorContext<String> context, String message) {
        super(context);// 传递给上层的上下文
        this.message = message;
    }

    /**
     * 附带参数一并实例化构建成 Actor
     */
    public static Behavior<String> create(String message) {
        // 内部其实是 (ActorContext)->AbstractBehavior<T> 的回调方法
        // 只需要按照回调方式传入即可, 其实就是动态实例化对象返回
        return Behaviors.setup((ctx) -> new PrintParamsActor(ctx, message));
    }

    // 其他略
}

class StartStopActor1 extends AbstractBehavior<String> {

    static Behavior<String> create() {
        return Behaviors.setup(StartStopActor1::new);
    }

    private StartStopActor1(ActorContext<String> context) {
        super(context);
        System.out.println("first started");

        // 初始化构建的同时动态构建其他 Actor 并挂载于 pekko://{父节点}/second 
        context.spawn(StartStopActor2.create(), "second");
    }

    @Override
    public Receive<String> createReceive() {
        return newReceiveBuilder()
                // 监听外部的消息是否有传递 'stop' 字符, 有的话触发 Behaviors::stopped 状态变动
                // Behaviors::stopped() 是系统定义的停止并退出当前 Actor 特殊状态变更消息
                .onMessageEquals("stop", Behaviors::stopped)
                // onSignal 是系统特殊信号拦截
                // PostStop.class 是系统内置的退出消息
                .onSignal(PostStop.class, signal -> onPostStop())
                .build();
    }

    private Behavior<String> onPostStop() {
        System.out.println("first stopped");
        return this;
    }
}

class StartStopActor2 extends AbstractBehavior<String> {

    static Behavior<String> create() {
        return Behaviors.setup(StartStopActor2::new);
    }

    private StartStopActor2(ActorContext<String> context) {
        super(context);
        System.out.println("second started");
    }

    @Override
    public Receive<String> createReceive() {
        // 监听推出的系统信号
        // 加入父级被关闭的时候, 会优先关闭子节点的 Actor, 最后关闭掉父 Actor 回调方法
        return newReceiveBuilder().onSignal(PostStop.class, signal -> onPostStop()).build();
    }

    private Behavior<String> onPostStop() {
        System.out.println("second stopped");
        return this;
    }
}

class SupervisingActor extends AbstractBehavior<String> {

    static Behavior<String> create() {
        return Behaviors.setup(SupervisingActor::new);
    }

    private final ActorRef<String> child;

    private SupervisingActor(ActorContext<String> context) {
        super(context);
        // 动态初始化子节点
        child = context.spawn(
                // Behaviors.supervise 设定高级配置的 Actor 实例化
                Behaviors.supervise(SupervisedActor.create())
                        // 注意这里追加 Actor 异常回调处理, 假如内部 Actor 抛出运行时相关异常( RuntimeException)
                        // 回调内部的 SupervisorStrategy.restart() 就是内置的重启方法, 会对子 Actor 立即执行服务重启
                        .onFailure(SupervisorStrategy.restart())
                , "supervised-actor");
    }

    @Override
    public Receive<String> createReceive() {
        return newReceiveBuilder().onMessageEquals("failChild", this::onFailChild).build();
    }

    private Behavior<String> onFailChild() {
        child.tell("fail");
        return this;
    }
}

/**
 * 被父Actor管理的子 Actor
 */
class SupervisedActor extends AbstractBehavior<String> {

    static Behavior<String> create() {
        return Behaviors.setup(SupervisedActor::new);
    }

    private SupervisedActor(ActorContext<String> context) {
        super(context);
        System.out.println("supervised actor started");
    }

    @Override
    public Receive<String> createReceive() {
        return newReceiveBuilder()
                // 接受到主动抛出异常的消息
                .onMessageEquals("fail", this::fail)
                // 系统内部预定义的信号:
                // - 重启(PreRestart)
                // - 关闭(PostStop)
                .onSignal(PreRestart.class, signal -> preRestart())
                // 注意: 重启的时候会默认执行关闭之后调用 PreRestart 再初始化
                .onSignal(PostStop.class, signal -> postStop())
                .build();
    }

    private Behavior<String> fail() {
        // !!! 注意: 这里代表子节点内部业务异常而抛出异常信号
        System.out.println("supervised actor fails now");
        throw new RuntimeException("I failed!");
    }

    private Behavior<String> preRestart() {
        System.out.println("supervised will be restarted");
        return this;
    }

    private Behavior<String> postStop() {
        System.out.println("supervised stopped");
        return this;
    }
}