// 10秒内最多重试3次的 Actor 管理器策略
RestartSupervisorStrategy strategy = SupervisorStrategy.restart()
                .withLoggingEnabled(true) // 启用日志记录
                .withStopChildren(true) // 重启时一并停止所有子 Actor, 默认为 true(可以不设置)
                .withLimit(3, Duration.ofSeconds(10)); // 异常最多重启3次, 10秒内超限则停止

Behavior<Object> behavior = Behaviors.supervise(Behaviors.empty()) // Behaviors.empty() 替换成具体 Actor
        .onFailure(RuntimeException.class, strategy) // 运行时异常: 重启策略, 10秒内最多重试3次, 超限则停止
        .onFailure(IllegalArgumentException.class, SupervisorStrategy.stop()) // 参数异常: 直接停止, 不重试
        ;
// 需要拦截不同异常可以自定义设置 onFailure 回调处理
// 不需要自动 Behaviors.setup, 直接移交 ActorSystem 即可

/**
 * 房间命令接口
 */
public sealed interface IRoomCommand {

    // 加入房间
    record JoinRoom(long uid) implements IRoomCommand {
    }

    // 广播消息
    record Broadcast(String message) implements IRoomCommand {
    }

    // 离开房间
    record LeaveRoom(long uid) implements IRoomCommand {
    }
}


/**
 * 房间管理器
 */
public static class RoomMonitor extends AbstractBehavior<IRoomCommand> {


    /**
     * 构造函数
     */
    private RoomMonitor(ActorContext<IRoomCommand> context) {
        super(context);
    }

    /**
     * 静态实例化
     */
    public static Behavior<IRoomCommand> create() {
        // 10秒内最多重试3次的 Actor 管理器策略
        RestartSupervisorStrategy strategy = SupervisorStrategy.restart()
                .withLoggingEnabled(true) // 启用日志记录
                .withStopChildren(true) // 重启时一并停止所有子 Actor, 默认为 true(可以不设置)
                .withLimit(3, Duration.ofSeconds(10)); // 异常最多重启3次, 10秒内超限则停止

        // 生成对应监督策略 Actor 管理器, 这里就是具体 OneForAllStrategy 初始化
        return Behaviors.supervise(Behaviors.setup(RoomMonitor::new))
                .onFailure(RuntimeException.class, strategy) // 运行时异常: 重启策略, 10秒内最多重试3次, 超限则停止
                .onFailure(IllegalArgumentException.class, SupervisorStrategy.stop()) // 参数异常: 直接停止, 不重试
                ;
    }


    /**
     * 用户列表
     */
    final Map<Long, ActorRef<IRoomCommand>> users = new ConcurrentHashMap<>();

    @Override
    public Receive<IRoomCommand> createReceive() {
        return newReceiveBuilder()

                // 系统处理
                .onSignal(Terminated.class, this::onTerminated)
                .onSignal(PreRestart.class, this::onPreRestart)

                // 业务处理
                .onMessage(IRoomCommand.JoinRoom.class, this::onJoinRoom)
                .onMessage(IRoomCommand.Broadcast.class, this::onBroadcast)
                .onMessage(IRoomCommand.LeaveRoom.class, this::onLeaveRoom)
                .build();
    }

    /**
     * 离开房间
     */
    private Behavior<IRoomCommand> onLeaveRoom(IRoomCommand.LeaveRoom leaveRoom) {
        ActorRef<IRoomCommand> user = users.remove(leaveRoom.uid());
        if (user != null) getContext().stop(user);
        return this;
    }

    /**
     * 广播消息
     */
    private Behavior<IRoomCommand> onBroadcast(IRoomCommand.Broadcast broadcast) {
        users.values().forEach(user -> user.tell(broadcast));
        return this;
    }

    /**
     * 监听 PreRestart 信号
     * restart 重建实例后 sessions 为全新空映射, 旧会话全部作废
     */
    private Behavior<IRoomCommand> onPreRestart(PreRestart preRestart) {
        users.values().forEach(ref -> getContext().stop(ref));
        users.clear(); // 清空用户列表等待重新复用
        return this;
    }

    /**
     * 监听子 Actor 停止信号
     */
    private Behavior<IRoomCommand> onTerminated(Terminated terminated) {
        ActorRef<?> ref = terminated.getRef();
        var iterator = users.entrySet().iterator();
        while (iterator.hasNext()) {
            var entry = iterator.next();
            if (entry.getValue().equals(ref)) {
                iterator.remove();

                // 打印内容
                getContext().getLog().info("Removing session with uid {}", entry.getKey());
                return this;
            }
        }
        return this;
    }

    /**
     * 加入房间
     */
    private Behavior<IRoomCommand> onJoinRoom(IRoomCommand.JoinRoom command) {
        if (users.containsKey(command.uid)) return this; // 已经加入会话


        // 包装监督策略: 1分钟内最多重启3次, 超过则停止
        // 一般子 Actor 推荐自定义再初始化一次
        Behavior<IRoomCommand> behavior = Behaviors.supervise(RoomSession.create(command.uid()))
                .onFailure(SupervisorStrategy
                        .restart()
                        .withLimit(3, Duration.ofMinutes(1))
                        .withLoggingEnabled(true));

        // 生成子Actor 纳入当前监督者管理
        ActorRef<IRoomCommand> ref = getContext().spawn(behavior, "session-" + command.uid);
        getContext().watch(ref); // 建立生命周期监视

        // 写入管理表
        users.put(command.uid, ref);
        return this;
    }
}


/**
 * 房间会话实体, 也就是最底层的 one 对象
 */
public static class RoomSession extends AbstractBehavior<IRoomCommand> {

    final long uid;

    private RoomSession(ActorContext<IRoomCommand> context, long uid) {
        super(context);
        this.uid = uid;
    }

    public static Behavior<IRoomCommand> create(long uid) {
        return Behaviors.setup(context -> new RoomSession(context, uid));
    }

    @Override
    public Receive<IRoomCommand> createReceive() {
        return newReceiveBuilder()
                .onSignal(PostStop.class, (ignore) -> {
                    // 退出信号
                    getContext().getLog().info("RoomSession stopped, uid: {}", uid);
                    return this;
                })
                .build();
    }
}

/**
 * 状态命令
 */
public sealed interface StorageCommand {
    record SaveSnapshot(StorageSnapshot snapshot) implements StorageCommand {
    }
}

/**
 * 状态 Actor
 */
public static class StorageActor extends AbstractBehavior<StorageCommand> {


    /**
     * 构造函数
     */
    private StorageActor(ActorContext<StorageCommand> context) {
        super(context);
    }

    /**
     * 直接注入 ActorSystem 而不需要手动 Behavior 挂载
     */
    public static ActorRef<StorageCommand> create(ActorContext<StorageCommand> context, String name, int size) {
        // Routers 就是内置 Actor 池功能, withRoundRobinRouting 就是声明采用负载均衡
        return context.spawn(
                Routers.pool(size, Behaviors.setup(StorageActor::new))
                        .withRoundRobinRouting(), name
        );
    }


    @Override
    public Receive<StorageCommand> createReceive() {
        return newReceiveBuilder()
                .onMessage(StorageCommand.SaveSnapshot.class, this::onSaveSnapshot)
                .build();
    }

    private Behavior<StorageCommand> onSaveSnapshot(StorageCommand.SaveSnapshot saveSnapshot) {
        // 处理数据落地业务
        return this;
    }
}

/**
 * WebSocket 运行时
 * 这是一套伪代码运行时
 */
public static class WebSocketRunner<T> {

    final ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "WebSocketRunner");

    protected ActorRef<T> actor = null;

    /**
     * 会话连接
     */
    public void onConnected(String sessionId) {
        // 假设已经创建好 WebSocketManager 管理器实现了 one_for_all
        // 后面将 Behaviors.empty() 替换成具体 WebSocketManager 实现
        this.actor = system.systemActorOf(Behaviors.empty(), "session", Props.empty());
    }

    /**
     * 会话断开
     */
    public void onDisconnected(String sessionId, Throwable throwable) {
        if (Objects.nonNull(actor)) {
            // 通知退出 PostStop 停止信号
            //this.actor.tell({自定义断开 Command});
        }
    }

    /**
     * 消息传递
     */
    public void onMessage(String sessionId, String message) {
        if (Objects.nonNull(actor)) {
            //actor.tell(sessionId,message);
        }
    }
}

/**
 * WebSocket 命令
 */
public sealed interface IWebSocketCommand {
    record Connected(String sessionId) implements IWebSocketCommand {
    }

    record Disconnected(String sessionId, Throwable throwable) implements IWebSocketCommand {
    }

    record Message(String sessionId, String message) implements IWebSocketCommand {
    }
}


/**
 * WebSocket 管理器
 */
public static class WebSocketManager extends AbstractBehavior<IWebSocketCommand> {

    public WebSocketManager(ActorContext<IWebSocketCommand> context) {
        super(context);
    }

    /**
     * 静态实例化
     */
    public static Behavior<IWebSocketCommand> create() {
        // 10秒内最多重试3次的 Actor 管理器策略
        RestartSupervisorStrategy strategy = SupervisorStrategy.restart()
                .withLoggingEnabled(true) // 启用日志记录
                .withStopChildren(true) // 重启时一并停止所有子 Actor, 默认为 true(可以不设置)
                .withLimit(3, Duration.ofSeconds(10)); // 异常最多重启3次, 10秒内超限则停止

        // 生成对应监督策略 Actor 管理器, 这里就是具体 OneForAllStrategy 初始化
        return Behaviors.supervise(Behaviors.setup(WebSocketManager::new))
                .onFailure(RuntimeException.class, strategy) // 运行时异常: 重启策略, 10秒内最多重试3次, 超限则停止
                .onFailure(IllegalArgumentException.class, SupervisorStrategy.stop()) // 参数异常: 直接停止, 不重试
                ;
    }

    /**
     * 创建接收器
     */
    @Override
    public Receive<IWebSocketCommand> createReceive() {
        return newReceiveBuilder()
                .onMessage(IWebSocketCommand.Connected.class, this::onConnected)
                .onMessage(IWebSocketCommand.Disconnected.class, this::onDisconnected)
                .build();
    }

    final Map<String, ActorRef<IWebSocketCommand>> sessions = new ConcurrentHashMap<>();

    /**
     * 断开连接
     */
    private Behavior<IWebSocketCommand> onDisconnected(IWebSocketCommand.Disconnected disconnected) {
        ActorRef<IWebSocketCommand> session = sessions.remove(disconnected.sessionId);
        if (Objects.nonNull(session)) {
            getContext().stop(session);
        }

        return this;
    }

    /**
     * 处理连接
     */
    private Behavior<IWebSocketCommand> onConnected(IWebSocketCommand.Connected command) {
        if (sessions.containsKey(command.sessionId())) return this; // 已经加入会话


        // 包装监督策略: 1分钟内最多重启3次, 超过则停止
        // 一般子 Actor 推荐自定义再初始化一次
        Behavior<IWebSocketCommand> behavior = Behaviors.supervise(WebSocketSession.create(command.sessionId()))
                .onFailure(SupervisorStrategy
                        .restart()
                        .withLimit(3, Duration.ofMinutes(1))
                        .withLoggingEnabled(true));

        // 生成子Actor 纳入当前监督者管理
        ActorRef<IWebSocketCommand> ref = getContext().spawn(behavior, "session-" + command.sessionId());
        getContext().watch(ref); // 建立生命周期监视

        // 写入管理表
        sessions.put(command.sessionId(), ref);

        return this;
    }
}


/**
 * 具体会话对象
 */
public static class WebSocketSession extends AbstractBehavior<IWebSocketCommand> {

    protected final String sessionId;

    public WebSocketSession(ActorContext<IWebSocketCommand> context, String sessionId) {
        super(context);
        this.sessionId = sessionId;
    }

    public static Behavior<IWebSocketCommand> create(String sessionId) {
        return Behaviors.setup(context -> new WebSocketSession(context, sessionId));
    }

    @Override
    public Receive<IWebSocketCommand> createReceive() {
        return newReceiveBuilder().build();
    }
}

/**
 * 授权命令
 */
public sealed interface IAuthorizationCommand {

    /**
     * 验证授权
     */
    record Verify(
            long uid,
            String token,
            ActorRef<IAuthorizationCommand> replyTo
    ) implements IAuthorizationCommand {

    }

    /**
     * 验证结果
     */
    record Result(
            boolean success,
            long uid,
            int state,
            String body
    ) implements IAuthorizationCommand {

    }
}

/**
 * 权限验证池
 */
public static class AuthorizationWorker extends AbstractBehavior<IAuthorizationCommand> {

    /**
     * 构造函数
     */
    private AuthorizationWorker(ActorContext<IAuthorizationCommand> context) {
        super(context);
        this.client = HttpClient.newHttpClient();
    }

    /**
     * 创建授权验证池
     */
    public static <T> ActorRef<IAuthorizationCommand> create(ActorContext<T> ctx, int size, String name) {
        return ctx.spawn(
                Routers.pool(size, Behaviors.setup(AuthorizationWorker::new)).withRoundRobinRouting(),
                name
        );
    }

    @Override
    public Receive<IAuthorizationCommand> createReceive() {
        return newReceiveBuilder()
                .onMessage(IAuthorizationCommand.Verify.class, this::onVerify)
                .build();
    }


    /**
     * Http 异步客户端
     */
    private final HttpClient client;

    /**
     * 验证授权
     */
    private Behavior<IAuthorizationCommand> onVerify(IAuthorizationCommand.Verify verify) {

        // 异步 HTTP 调用
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create("https://example.com"))
                .timeout(Duration.ofSeconds(3))
                .header("Authorization", "Bearer " + verify.token)
                .header("Content-Type", "application/json")
                .POST(HttpRequest.BodyPublishers.ofString(
                        "{\"token\":\"" + verify.token() + "\"}"
                ))
                .build();


        // 异步推送
        // 验证数据验证之后返回给 Actor
        client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
                .thenApply(response -> {
                    // 解析验证结果
                    return new IAuthorizationCommand.Result(
                            response.statusCode() == 200,
                            verify.uid,
                            response.statusCode(),
                            response.body()
                    );
                }).exceptionally(ex -> new IAuthorizationCommand.Result(
                        false,
                        verify.uid,
                        500,
                        ex.getMessage()
                )).thenAccept(verify.replyTo::tell);// 完成数据验证后回写给来源 ActorRef

        return Behaviors.same();
    }
}

/**
 * 具体会话对象
 */
public static class WebSocketSession extends AbstractBehavior<IWebSocketCommand> {

    protected final String sessionId;

    public WebSocketSession(ActorContext<IWebSocketCommand> context, String sessionId) {
        super(context);
        this.sessionId = sessionId;
    }

    public static Behavior<IWebSocketCommand> create(String sessionId) {
        return Behaviors.setup(context -> {
            WebSocketSession that = new WebSocketSession(context, sessionId);
            // 这里是个小技巧, 让 createReceive 指定运行消息处理器
            return that.unauthenticated();
        });
    }

    @Override
    public Receive<IWebSocketCommand> createReceive() {
        return newReceiveBuilder().build();
    }


    /**
     * 已授权的消息处理器
     */
    private Behavior<IWebSocketCommand> authenticated() {
        // 只放置已授权的业务
        return newReceiveBuilder()
                .build();
    }

    /**
     * 未授权的消息处理器
     */
    private Behavior<IWebSocketCommand> unauthenticated() {
        // 只放置未授权的业务
        return newReceiveBuilder()
                .build();
    }

}

// 示例: 普通业务消息处理
private Behavior<IWebSocketCommand> onMessage(IWebSocketCommand.Message msg) {
    // 下面两种写法效果基本一致, 但是更加推荐 Behaviors.same() 返回值
    return Behaviors.same(); // 标准写法
    return this;             // 不推荐
}

/**
 * 认证超时定时器
 */
private Cancelable timeoutTimer;


// ========================================================================
// 未认证态 —— 只处理认证消息, 超时强制释放连接
// ========================================================================
private Behavior<IWebSocketCommand> unauthenticated() {
    // 启动认证超时定时器: 5秒未完成认证直接断开
    timeoutTimer = getContext().scheduleOnce(
            Duration.ofSeconds(5),
            getContext().getSelf(),
            new IWebSocketCommand.AuthTimeout()
    );

    return newReceiveBuilder()
            // 处理客户端消息
            .onMessage(IWebSocketCommand.Message.class, this::onUnauthenticatedMessage)
            // 认证超时
            .onMessage(IWebSocketCommand.AuthTimeout.class, this::onAuthTimeout)
            // 异步验证结果回调
            .onMessage(IWebSocketCommand.AuthResult.class, this::onAuthResult)
            .build();
}


/**
 * 认证超时: 主动关闭连接, 销毁会话并释放端口资源
 */
private Behavior<IWebSocketCommand> onAuthTimeout(IWebSocketCommand.AuthTimeout timeout) {
    getContext().getLog().warn("会话认证超时, 释放连接: {}", sessionId);
    //writeRef.tell(new IWriteCommand.CloseConnection(408, "认证超时")); // 推送 socket/websocket 数据
    return Behaviors.stopped(); // 直接停止当前 Actor
}

/**
 * 处理认证结果: 在 Actor 主线程串行执行, 保证线程安全
 */
private Behavior<IWebSocketCommand> onAuthResult(IWebSocketCommand.AuthResult result) {
    // 取消超时定时器
    timeoutTimer.cancel();

    if (result.success()) {
        this.uid = result.uid();
        getContext().getLog().info("会话认证通过: uid={}, session={}", uid, sessionId);
        // 通知客户端认证成功
        writeRef.tell(new IWriteCommand.WriteMessage("{\"code\":0,\"msg\":\"auth ok\"}"));
        // 核心: 切换到 [已认证业务态], 这就是典型的状态机切换状态行为
        return authenticated();
    } else {
        // 认证失败: 直接断开连接, 释放资源
        writeRef.tell(new IWriteCommand.CloseConnection(401, "认证失败: " + result.body()));
        return Behaviors.stopped();
    }
}

private static final int MIN_FRAME_LENGTH = 8;       // 最小帧长: 4字节长度+4字节msgId
private static final int MAX_FRAME_LENGTH = 1024 * 1024; // 最大单包1MB
private static final int MSG_ID_C2S_LOGIN = 10001; // 未认证唯一允许的msgId

/**
 * 消息解包并启动验证
 */
private Behavior<SessionCommand> onUnauthenticatedMessage(SessionCommand.Message command) {
    final String connectionId = command.connection().id();

    // 1. 显式大端序
    final ByteBuffer buffer = ByteBuffer
            .wrap(command.message())
            .order(ByteOrder.BIG_ENDIAN);

    // 空包直接丢弃
    if (!buffer.hasRemaining()) {
        return Behaviors.same();
    }


    // 2. 最小帧长校验
    if (buffer.remaining() < MIN_FRAME_LENGTH) {
        logger.warnf("session actor %s frame parse error", connectionId);
        return Behaviors.stopped();
    }

    // 3. 读取帧体总长度(=msgId 4字节 + 业务载荷长度)
    int bodyLength = buffer.getInt();
    if (bodyLength < 4 || bodyLength > MAX_FRAME_LENGTH) {
        logger.warnf("session actor %s frame invalid length: length = %d", connectionId, bodyLength);
        return Behaviors.stopped();
    }
    if (buffer.remaining() != bodyLength) {
        logger.warnf("session actor %s frame length mismatch: length = %d, remaining = %d", connectionId, bodyLength, buffer.remaining());
        return Behaviors.stopped();
    }

    // 4. 读取msgId, 未认证态白名单拦截
    int msgId = buffer.getInt();
    if (msgId != MSG_ID_C2S_LOGIN) {
        logger.warnf("session actor %s received illegal msgId: %d", connectionId, msgId);
        return Behaviors.stopped();
    }

    // 5. 提取纯业务载荷(跳过msgId的4字节)
    int payloadLength = bodyLength - 4;
    byte[] payload = new byte[payloadLength];
    buffer.get(payload);

    // 用指定构建器来转化数据实体
    // 异常直接终端 Actor 会话
    try {
        C2SLogin protobuf = C2SLogin.parseFrom(payload);
        // todo: 转发到验证 Actor

        // 通知验证器处理消息
        final var ref = verifier.actorRef();
        ref.tell(new VerifierCommand.VerifyRequest(1001, "meteorcat", getContext().getSelf()));
        return Behaviors.same();
    } catch (InvalidProtocolBufferException e) {
        logger.errorf(e, "session actor %s protobuf parse error", command.connection().id());
        return Behaviors.stopped();
    }
}

/**
 * 广播消息
 */
private Behavior<IRoomCommand> onBroadcast(IRoomCommand.Broadcast broadcast) {
    users.values().forEach(user -> user.tell(broadcast));
    return this;
}

/**
 * 房间管理器 - 消息发布端
 */
public static class RoomMonitor extends AbstractBehavior<IRoomCommand> {

    // 其他略

    /**
     * 广播消息
     */
    private Behavior<IRoomCommand> onBroadcast(IRoomCommand.Broadcast broadcast) {
        // 直接发布消息 , 让所有订阅者接收(也就是子 Actor 接收)
        getContext().getSystem().eventStream().tell(new EventStream.Publish<>(broadcast));
        return Behaviors.same();
    }
}


/**
 * 房间会话实体, 也就是最底层的 one 对象 - 消息订阅端
 */
public static class RoomSession extends AbstractBehavior<IRoomCommand> {

    // 其他略

    /**
     * 构造函数
     */
    private RoomSession(ActorContext<IRoomCommand> context, long uid) {
        super(context);
        this.uid = uid;

        // 订阅句柄
        final ActorRef<EventStream.Command> event = context.getSystem().eventStream();

        // 订阅父级广播消息
        event.tell(new EventStream.Subscribe<>(
                IRoomCommand.Broadcast.class, getContext().getSelf().narrow()
        ));
    }


    @Override
    public Receive<IRoomCommand> createReceive() {
        return newReceiveBuilder()
                .onSignal(PostStop.class, (ignore) -> {
                    // 退出信号
                    getContext().getLog().info("RoomSession stopped, uid: {}", uid);

                    // 订阅句柄, 退订广播消息
                    // 取消订阅一个 Actor 时, 默认移除该 Actor 在事件总线上的全部订阅
                    final ActorRef<EventStream.Command> event = getContext().getSystem().eventStream();
                    event.tell(new EventStream.Unsubscribe<>(getContext().getSelf()));
                    return this;
                })
                .onMessage(IRoomCommand.Broadcast.class, this::onBroadcast) // 获取订阅的消息
                .build();
    }
}