/**
 * Pekko 定时调度 Actor, 最好把任务设计成 Actor 投递, 结合 Actor 做业务调度, 而不要直接调用内部的任何函数和方法
 */
public static class PekkoSchedule extends AbstractActor {

    /**
     * Actor 定时器
     */
    final Scheduler scheduler = getContext().getSystem().getScheduler();

    /**
     * Actor 调度器
     */
    final ExecutionContextExecutor dispatcher = getContext().getDispatcher();

    /**
     * 延迟 2 秒执行任务
     */
    final Cancellable onceTask = scheduler.scheduleOnce(
            Duration.ofSeconds(2), // 延迟时间
            () -> getSelf().tell("Hello.World", ActorRef.noSender()), // 执行任务
            dispatcher // 绑定的调度器
    );


    /**
     * 周期性执行任务
     */
    final Cancellable fixedRateTask = scheduler.scheduleAtFixedRate(
            Duration.ofSeconds(3), // 初始延迟
            Duration.ofSeconds(3), // 执行间隔
            () -> getSelf().tell("Hello.EachTask", ActorRef.noSender()), // 执行任务
            dispatcher  // 绑定的调度器
    );


    /**
     * 退出 Actor 回调
     */
    @Override
    public void postStop() {
        // 取消定时任务
        onceTask.cancel();
        fixedRateTask.cancel();
    }
}

/**
 * Pekko 请求-响应调度 Actor
 */
public static class PekkoPatterns extends AbstractActor {

    /**
     * 定义传输数据结构
     *
     * @param values
     */
    record AskSum(int[] values) {

    }

    /**
     * 启动 Actor 回调
     */
    @Override
    public void preStart() {
        // 设计请求: 要求计算 int[] 的合计总额, 5秒获取结果否则超时弹出异常
        // Patterns.ask 响应结果只能返回 CompletionStage<Object>, 内部类型需要自行去匹配推断
        CompletionStage<Object> future = Patterns.ask(getSelf(), new AskSum(new int[]{10, 20, 30}), Duration.ofSeconds(5));

        // 注册异步调用
        future.thenApply(res -> {
            // 过滤判断结果
            if (res instanceof Integer v) {
                return v;
            } else {
                throw new IllegalArgumentException("合计总额参数异常");
            }
        }).thenAccept(res -> {
            System.out.printf("最后得结果: %d%n", res);
        }).exceptionally(throwable -> {
            // 拦截异步调用过程之中的异常信息
            System.err.println(throwable.getMessage());
            return null; // 这里返回 null 代表外部无需捕获最后调用值, 指代 Void
        });

    }

    /**
     * 消息拦截
     */
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .matchk(AskSum.class, req -> {
                    final ActorRef sender = getSender(); // 获取请求过来的响应地址

                    // 将计算结果响应返回过去, 如果没有返回会导致外层异步超时失败
                    sender.tell(
                            Arrays.stream(req.values()).sum(),
                            ActorRef.noSender()
                    );
                }).build();
    }
}

/**
 * Pekko 自定义调度 CompletableFuture
 */
public static class PekkoCompletableFuture extends AbstractActor {
    /**
     * Actor 调度器
     */
    final ExecutionContextExecutor dispatcher = getContext().getDispatcher();

    /**
     * 启动 Actor 回调
     */
    @Override
    public void preStart() {
        // 注意: 这里仅仅是参考, Actor 内部任务调度一定要让内部自行调度, 而不要手动采用 dispatcher 调度
        // runAsync 第二个参数支持 Java 内部 Executor 线程执行器
        CompletableFuture.runAsync(() -> {
                    System.out.println("Hello.World");
                }, dispatcher)
                .orTimeout(3, TimeUnit.SECONDS) // 设置异步超时
                .thenAccept(ignore -> {
                    System.out.println("执行完成");
                }).exceptionally(throwable -> {
                    System.err.printf("执行异常：%s%n", throwable.toString());
                    return null;
                });
    }
}

/**
 * Pekko CompletableFuture 投递 Actor
 */
public static class PekkoCompletableFuture extends AbstractActor {
    /**
     * Actor 调度器
     */
    final ExecutionContextExecutor dispatcher = getContext().getDispatcher();

    /**
     * 构建独立的线程池而非 Actor 的 dispatcher 来执行异步任务, 避免挤占消息处理资源
     */
    final ExecutorService executor = Executors.newFixedThreadPool(3);


    /**
     * 启动 Actor 回调
     */
    @Override
    public void preStart() {
        // 创建自定义异步任务, 将调度器设置为自定义线程池上
        CompletableFuture<Long> task = CompletableFuture.supplyAsync(() -> {
                    System.out.println("执行完成");
                    return System.currentTimeMillis();
                }, executor)
                .orTimeout(3, TimeUnit.SECONDS) // 设置异步超时
                .exceptionally(throwable -> {
                    System.err.printf("执行异常：%s%n", throwable.toString());

                    // 这里也可以实现任务充实机制
                    // 首次任务失败可以再次运行一次
                    return CompletableFuture.supplyAsync(() -> {
                                System.out.println("任务重试");
                                return System.currentTimeMillis();
                            }, executor)
                            .orTimeout(3, TimeUnit.SECONDS)
                            .exceptionally(throwable1 -> {
                                // 重试任务还是失败, 直接可以设置异常消息通知
                                System.err.printf("重试异常：%s%n", throwable1.toString());
                                return null;
                            })
                            .join();
                });

        // 将异步结果 pipe 到 Actor 自身的消息队列, 也就是发送到 Match 消息拦截之中
        // 这种就是通过自定义线程池来调度异步任务, 然后获取结果之后 actorRef.tell(msg) 将结果投递到 Actor
        Patterns.pipe(task, dispatcher).to(getSelf());
    }
}