import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.http.javadsl.Http;
import org.apache.pekko.http.javadsl.ServerBinding;
import org.apache.pekko.http.javadsl.server.AllDirectives;
import org.apache.pekko.http.javadsl.server.Route;

import java.util.concurrent.CompletionStage;

/**
 * 构建 PekkoHTTP 服务
 */
public class PekkoHttpExample extends AllDirectives {


    /**
     * 服务请求入口
     */
    public static void main(String[] args) throws Exception {

        // 创建初始化默认的 ActorSystem Boot
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "routes");

        // 生成 HTTP 功能服务
        final Http http = Http.get(system);

        // 生成管理的内部类对象实例, 这个类必须继承 AllDirectives 类
        PekkoHttpExample app = new PekkoHttpExample();

        // 将 HTTP 服务和内部类做好映射, 并且设置监听的地址和端口
        final CompletionStage<ServerBinding> binding =
                http.newServerAt("localhost", 18880).bind(app.createRoute());

        // 这里采用任意输入回车中断监听, 如果未做任何键盘操作则是保持运行
        System.out.println("Server online at http://localhost:18880/\nPress RETURN to stop...");
        int ignore = System.in.read();

        // 退出之后回收这部分 HTTP 资源并且直接中断 Actor 系统服务
        binding
                .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
                .thenAccept(unbound -> system.terminate()); // and shutdown when done

    }

    /**
     * 构建请求路径的路由
     */
    private Route createRoute() {
        // 添加 GET /hello 访问路径, 映射到内部函数调用输出
        return concat(path("hello", () -> get(() -> complete("Hello.World"))));
    }
}

// 具体代码参照: https://pekko.apache.org/docs/pekko-http/current/introduction.html

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.http.javadsl.Http;
import org.apache.pekko.http.javadsl.ServerBinding;
import org.apache.pekko.http.javadsl.marshallers.jackson.Jackson;
import org.apache.pekko.http.javadsl.server.AllDirectives;
import org.apache.pekko.http.javadsl.server.Route;

import java.util.concurrent.CompletionStage;

/**
 * PekkoHttp JSON 响应
 */
public class PekkoJsonExample extends AllDirectives {
    /**
     * 服务请求入口
     */
    public static void main(String[] args) throws Exception {

        // 创建初始化默认的 ActorSystem Boot
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "routes");

        // 生成 HTTP 功能服务
        final Http http = Http.get(system);

        // 生成管理的内部类对象实例, 这个类必须继承 AllDirectives 类
        PekkoJsonExample app = new PekkoJsonExample();

        // 将 HTTP 服务和内部类做好映射, 并且设置监听的地址和端口
        final CompletionStage<ServerBinding> binding =
                http.newServerAt("localhost", 18880).bind(app.createRoute());

        // 这里采用任意输入回车中断监听, 如果未做任何键盘操作则是保持运行
        System.out.println("Server online at http://localhost:18880/\nPress RETURN to stop...");
        int ignore = System.in.read();

        // 退出之后回收这部分 HTTP 资源并且直接中断 Actor 系统服务
        binding
                .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
                .thenAccept(unbound -> system.terminate()); // and shutdown when done

    }

    /**
     * 构建请求路径的路由
     */
    private Route createRoute() {
        // 添加 GET /hello 访问路径, 映射到内部函数调用输出
        return concat(
                get(() -> pathPrefix("hello", () -> complete("Hello.World"))),
                get(() -> pathPrefix("item", () -> {
                    Item item = new Item("MeteorCat", 2026L);
                    return completeOK(item, Jackson.marshaller());
                }))
        );
    }

    /**
     * 采用 record 结构, 其实和 static class 差不多
     *
     * @param name
     * @param id
     */
    private record Item(String name, long id) {

        @JsonCreator
        private Item(@JsonProperty("name") String name, @JsonProperty("id") long id) {
            this.name = name;
            this.id = id;
        }
    }
}

import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.http.javadsl.Http;
import org.apache.pekko.http.javadsl.model.*;
import org.apache.pekko.http.javadsl.server.examples.petstore.Pet;
import org.apache.pekko.stream.SystemMaterializer;

import java.util.concurrent.CompletionStage;

public class ClientSingleRequestExample {

    public static void main(String[] args) {
        final ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "SingleRequest");

        // Actor 如果内部需要做 HTTP 请求转发可以直接这样使用
        // 这里 CompletionStage 是内部的异步运行时, 会将请求数据结果的 HttpResponse 做回调
        final CompletionStage<HttpResponse> responseFuture =
                Http.get(system).singleRequest(HttpRequest.create("https://pekko.apache.org"));
    }

    /**
     * 构建请求路径的路由, 如果需要内部请求动态返回可以参考下面处理 /bing 的路由处理
     */
    private Route createRoute(ActorSystem<Void> system) {
        // 添加 GET /hello 访问路径, 映射到内部函数调用输出
        return concat(
                get(() -> pathPrefix("hello", () -> complete("Hello.World"))),
                get(() -> pathPrefix("item", () -> {
                    Item item = new Item("MeteorCat", 2026L);
                    return completeOK(item, Jackson.marshaller());
                })),
                get(() -> pathPrefix("bing", () -> {
                    // 利用异步包装返回对应请求
                    // 动态加载远程请求网址信息信息
                    final CompletionStage<HttpResponse> responseFuture = Http.get(system).singleRequest(HttpRequest.create("https://bing.com"));
                    return onComplete(responseFuture, resp -> resp.fold(
                            ex -> complete(StatusCodes.BAD_GATEWAY, "upstream error"), // 如果检测到异常直接返回 BAD_GATEWAY 错误
                            this::complete // 没有异常就将对应数据返回给客户端
                    ));
                }))
        );
    }
}

// 只要返回 handleWebSocketMessages, Pekko-HTTP 会自动回 101 Switching Protocols
Route wsRoute = path("ws", () -> handleWebSocketMessages(flow));
// WebSocket 服务构建方式, flow 其实是 Flow, 后面会说明

// 1. 打开水龙头: 服务端 → 客户端
Source<Message, NotUsed> faucet = Source
                .queue(50, OverflowStrategy.dropHead())  // 业务线程往里倒消息, 如果触发背压信号就会默认抛弃50条最老数据
                .mapMaterializedValue(queueRef -> { /* 把 queueRef 存起来别处用 */ });

// 2. 再接下水口: 客户端 → 服务端
Sink<Message, NotUsed> drain = Sink
        .foreach(msg -> handle(msg));            // 客户端来的帧在这里处理

// 3. 中间装净水器
Flow<Message, Message, NotUsed> filter =
        Flow.of(Message.class)
                .filter(Message::isBinary)
                .map(this::decrypt)                 // 你的业务
                .map(this::encrypt);

// 4. 把三口拼成一根水管
Flow<Message, Message, NotUsed> pipeline =
        Flow.fromSinkAndSource(drain, faucet).via(filter);

// 最后就是设置 WebSocket 的消息路由功能
// return path("ws", () -> handleWebSocketMessages(pipeline));

import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.http.javadsl.Http;
import org.apache.pekko.http.javadsl.ServerBinding;
import org.apache.pekko.http.javadsl.model.ws.BinaryMessage;
import org.apache.pekko.http.javadsl.model.ws.Message;
import org.apache.pekko.http.javadsl.server.AllDirectives;
import org.apache.pekko.http.javadsl.server.Route;
import org.apache.pekko.stream.OverflowStrategy;
import org.apache.pekko.stream.javadsl.Flow;
import org.apache.pekko.stream.javadsl.Sink;
import org.apache.pekko.stream.javadsl.Source;
import org.apache.pekko.stream.javadsl.SourceQueueWithComplete;
import org.apache.pekko.util.ByteString;

import java.util.concurrent.CompletionStage;

/**
 * Pekko WebSocket 服务挂载
 */
public class PekkoWebSocketExample extends AllDirectives {

    /**
     * Actor 系统
     */
    final ActorSystem<Void> system;

    /**
     * 服务端推送客户端消息队列
     */
    private SourceQueueWithComplete<Message> queueRef;

    /**
     * 构造方法
     */
    public PekkoWebSocketExample(ActorSystem<Void> system) {
        this.system = system;
    }

    /**
     * 服务请求入口
     */
    public static void main(String[] args) throws Exception {

        // 创建初始化默认的 ActorSystem Boot
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "routes");

        // 生成 HTTP 功能服务
        final Http http = Http.get(system);

        // 生成管理的内部类对象实例, 这个类必须继承 AllDirectives 类
        PekkoWebSocketExample app = new PekkoWebSocketExample(system);

        // 将 HTTP 服务和内部类做好映射, 并且设置监听的地址和端口
        final CompletionStage<ServerBinding> binding =
                http.newServerAt("localhost", 18880).bind(app.createRoute());

        // 这里采用任意输入回车中断监听, 如果未做任何键盘操作则是保持运行
        System.out.println("Server online at http://localhost:18880/\nPress RETURN to stop...");
        int ignore = System.in.read();

        // 退出之后回收这部分 HTTP 资源并且直接中断 Actor 系统服务
        binding
                .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
                .thenAccept(unbound -> system.terminate()); // and shutdown when done
    }

    /**
     * 构建请求路径的路由
     */
    private Route createRoute() {
        // 添加 GET /hello 访问路径, 映射到内部函数调用输出
        return concat(
                get(() -> pathPrefix("hello", () -> complete("Hello.World"))),
                get(() -> pathPrefix("echo", () -> handleWebSocketMessages(createEchoFlow())))
        );
    }

    /**
     * 创建 echo(回显) WebSocket 消息管道
     */
    private Flow<Message, Message, NotUsed> createEchoFlow() {
        // 1. 水龙头：服务端→客户端
        Source<Message, NotUsed> faucet =
                Source.<Message>queue(50, OverflowStrategy.dropHead())
                        .mapMaterializedValue(queueRef -> {
                            // 把 queueRef 存起来，别处可以主动推消息
                            this.queueRef = queueRef;
                            return NotUsed.getInstance();
                        });


        // 2. 下水口：客户端→服务端
        Sink<Message, NotUsed> drain =
                Flow.<Message>create()
                        .filter(BinaryMessage.class::isInstance)   // 验证并且拦截二进制消息
                        .map(BinaryMessage.class::cast)            // 转化成二进制消息内容
                        .flatMapConcat(BinaryMessage::getStreamedData)   // Source<ByteString, _>
                        .to(Sink.foreach(chunk -> queueRef.offer(BinaryMessage.create(chunk)))); // 消费消息队列

        // 3. 拼成双向管
        return Flow.fromSinkAndSource(drain, faucet);
    }
}

// 更换成大小帧判断
// 这里要定义个 droppedFrames 原子量, 用于当作丢帧计数器
// 在功能类定义 private final AtomicLong droppedFrames = new AtomicLong(); 即可
Sink<Message, NotUsed> drain =
        Flow.<Message>create()
                .filter(BinaryMessage.class::isInstance)   // 验证并且拦截二进制消息
                .map(BinaryMessage.class::cast)            // 转化成二进制消息内容
                .flatMapConcat(BinaryMessage::getStreamedData)
                .map(BinaryMessage::create)                      // 打包成二进制消息结构
                .to(Sink.foreach(chunk -> queueRef.offer(chunk).whenComplete((r, e) -> {
                            // 计算消息抛弃的帧数, 超过一定数值就可以做报警
                            if (r != QueueOfferResult.enqueued()) {
                                long drops = droppedFrames.incrementAndGet();
                                // 每 1000 次写入日志或者报警
                                if (drops % 1000 == 0) {
                                    System.err.printf("WebSocket dropped {%d} frames%n", drops);
                                }
                            }
                        })
                ));

/**
 * PekkoWebSocket 数据帧实现
 */
public class PekkoValuesExample extends AllDirectives {

    /**
     * 服务请求入口
     */
    public static void main(String[] args) throws Exception {
        // 创建初始化默认的 ActorSystem Boot
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "routes");

        // 异步流计算器

        // 1. 声明数据流异步调度运行时, Materializer 可以复用, 如果是频繁需要异步调用最好实例化句柄持有
        Materializer materializer = Materializer.createMaterializer(system);

        // 2. 创建带有 SourceQueueWithComplete<Integer> 结果的生成者管道
        Source<Integer, SourceQueueWithComplete<Integer>> source = Source.queue(100, OverflowStrategy.backpressure());

        // 3. 创建带有 CompletionStage<Integer> 结果的消费者管道
        Sink<Integer, CompletionStage<Integer>> sink = Sink.fold(0, Integer::sum);

        // 4. 通过 materializer 运行时将构建成异步程序执行, 最后返回队列和结果对象
        Pair<SourceQueueWithComplete<Integer>, CompletionStage<Integer>> pair = source
                .toMat(sink, Keep.both())
                .run(materializer);

        SourceQueueWithComplete<Integer> queue = pair.first(); // 数据队列对象
        CompletionStage<Integer> future = pair.second(); // 数据异步结果对象

        // 5. 将数据投递进去执行
        RandomGenerator randomGenerator = RandomGenerator.getDefault();
        for (int i = 1; i <= 10; i++) {
            int value = randomGenerator.nextInt(10);
            System.out.println("Value = " + value);
            queue.offer(value);
        }

        // 6. 投递完成之后通知完成并准备执行
        queue.complete();

        // 7. 最后异步获取到最终执行结果
        future.thenAccept(sum -> {
            System.out.println("Total = " + sum);
        });

        // 这里阻塞等待到最后获取结果
        future.toCompletableFuture().join();
        // 也可以采用超时, 比如下面的超时5s
        // future.toCompletableFuture().get(5, TimeUnit.SECONDS);
    }
}

public static Flow<ByteString, ByteString, NotUsed> lengthField(
        int fieldLength,
        int fieldOffset,
        int maximumFrameLength,
        java.nio.ByteOrder byteOrder) {
    // 内部实现
}

import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.http.javadsl.Http;
import org.apache.pekko.http.javadsl.ServerBinding;
import org.apache.pekko.http.javadsl.model.ws.BinaryMessage;
import org.apache.pekko.http.javadsl.model.ws.Message;
import org.apache.pekko.http.javadsl.server.AllDirectives;
import org.apache.pekko.http.javadsl.server.Route;
import org.apache.pekko.stream.*;
import org.apache.pekko.stream.javadsl.*;
import org.apache.pekko.util.ByteString;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Base64;
import java.util.Objects;
import java.util.concurrent.CompletionStage;

/**
 * PekkoWebSocket 数据帧实现
 */
public class PekkoFramedExample extends AllDirectives {
    /**
     * Actor 系统
     */
    final ActorSystem<Void> system;


    /**
     * 消息长度字段的字节数(int32 = 4字节)
     */
    private static final int FIELD_LENGTH = Integer.BYTES;

    /**
     * 长度字段的偏移量(0，因为第一个字段就是长度)
     */
    private static final int FIELD_OFFSET = 0;


    /**
     * 最大帧长度(256KB，覆盖大部分游戏/业务场景)
     */
    private static final int MAX_FRAME_LENGTH = 256 * 1024;


    /**
     * 字节序(网络传输通常用大端)
     */
    private static final ByteOrder BYTE_ORDER = ByteOrder.BIG_ENDIAN;


    /**
     * 消息ID字段的偏移量(长度字段之后，4字节)
     */
    private static final int MSG_ID_OFFSET = FIELD_OFFSET + FIELD_LENGTH;


    /**
     * 消息主体的起始偏移量(长度+ID之后，8字节)
     */
    private static final int MSG_BODY_OFFSET = MSG_ID_OFFSET + 4;


    /**
     * 构造方法
     */
    public PekkoFramedExample(ActorSystem<Void> system) {
        this.system = system;
    }

    /**
     * 服务请求入口
     */
    public static void main(String[] args) throws Exception {
        // 创建初始化默认的 ActorSystem Boot
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "routes");

        // 生成测试数据转 Base64, 方便部分 WebSocket 客户端以这部分数据做传输
        String message = "Hello.World";
        byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
        ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES * 2 + bytes.length);
        buffer.putInt(Integer.BYTES + bytes.length); // 注意这里的内容长度是 Body 不含有 Header
        buffer.putInt(10001);
        buffer.put(bytes);
        System.out.printf("Test Bytes: %s%n", Arrays.toString(buffer.array()));
        String base64 = Base64.getEncoder().encodeToString(buffer.array());
        System.out.printf("Test Base64: %s%n", base64);


        // 生成 HTTP 功能服务
        final Http http = Http.get(system);

        // 生成管理的内部类对象实例, 这个类必须继承 AllDirectives 类
        PekkoFramedExample app = new PekkoFramedExample(system);

        // 将 HTTP 服务和内部类做好映射, 并且设置监听的地址和端口
        final CompletionStage<ServerBinding> binding =
                http.newServerAt("localhost", 18880).bind(app.createRoute());

        // 这里采用任意输入回车中断监听, 如果未做任何键盘操作则是保持运行
        System.out.println("Server online at http://localhost:18880/\nPress RETURN to stop...");
        int ignore = System.in.read();

        // 退出之后回收这部分 HTTP 资源并且直接中断 Actor 系统服务
        binding
                .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
                .thenAccept(unbound -> system.terminate()); // and shutdown when done
    }

    /**
     * 构建请求路径的路由
     */
    private Route createRoute() {
        // 添加 GET /hello 访问路径, 映射到内部函数调用输出
        return concat(get(() -> pathPrefix("frame", () -> handleWebSocketMessages(createFrameFlow().mapMaterializedValue(queueRef -> {
            // 提取请求的消息队列保存到内部, 最好在内部维护好具体的请求队列列表
            // queues.put(sessionId(), queue);
            return queueRef;
        })))));
    }

    /**
     * 创建消息分帧管道
     */
    private Flow<Message, Message, ?> createFrameFlow() {

        // 1. 构建lengthField拆帧流的数据解码器
        Flow<ByteString, ByteString, ?> frameDecoder = Framing.lengthField(
                FIELD_LENGTH,
                FIELD_OFFSET,
                MAX_FRAME_LENGTH,
                BYTE_ORDER
        );


        // 2. 完整流处理链路：缓存合并 → 拆帧 → 错误处理 → 解析 → 投递Actor
        return Flow.<Message>create()
                // 异步处理
                .map(msg -> {
                    if (msg instanceof BinaryMessage bm) {
                        return bm.getStrictData();
                    } else {
                        return ByteString.emptyByteString();
                    }
                })
                // 步骤1：过滤空数据
                .filter(bs -> !bs.isEmpty())
                // 步骤2：执行精准拆帧
                .via(frameDecoder)
                // 步骤3：错误处理, 多个错误可以这样拦截处理
                .recover(Throwable.class, () -> {
                    system.log().error("[拆帧异常]");
                    return ByteString.emptyByteString();// 异常直接返回空数据
                })
                // 步骤4：过滤空字节串（异常帧的返回值）
                .filter(byteString -> !byteString.isEmpty())
                // 步骤5：解析帧内容（提取ID、主体）
                .map(this::parseFrame)
                // 步骤6：过滤解析失败的帧
                .filter(Objects::nonNull)
                // 步骤7：控制台打印解析结果（替代Actor投递）
                .map(body -> {
                    // 打印核心信息到控制台
                    System.out.println("===== 解析到完整消息帧 ==================================");
                    System.out.println("消息ID：" + body.id());
                    System.out.println("实际主体长度：" + body.message.size() + " 字节");
                    System.out.println("消息主体内容：" + Arrays.toString(body.message.toArray()));
                    System.out.println("======================================================\n");
                    return body; // 保留返回值，不影响后续流处理
                })
                // 步骤8：背压超时保护（5秒）
                .backpressureTimeout(Duration.ofSeconds(5))
                // 步骤9：超时异常处理
                .recover(java.util.concurrent.TimeoutException.class, () -> {
                    system.log().warn("[流超时] 消息帧处理超时");
                    return null;
                })
                // 步骤10：过滤超时的空值
                .filter(Objects::nonNull)
                // 步骤11：核心 - 不向客户端返回响应，返回空流
                // 大部分情况都不是 请求-响应 模型, 而是动态消息推送过来
                .flatMapConcat(frame -> Source.empty()); // 空Source，无消息返回客户端
    }


    /**
     * 解析单帧数据：按「长度int32 + IDint32 + 主体」格式提取核心信息
     */
    private MessageFrame parseFrame(ByteString frame) {
        try {
            // 校验帧最小长度（至少包含长度+ID字段，共8字节）
            if (frame.size() < MSG_BODY_OFFSET) {
                system.log().warn("[帧解析] 无效帧：长度{} < 最小8字节", frame.size());
                return null;
            }

            // 开始解析数据
            ByteBuffer buffer = frame
                    .asByteBuffer()
                    .order(BYTE_ORDER);

            // 1. 提取消息长度（int32，大端字节序）
            int msgLength = buffer.getInt(FIELD_OFFSET);

            // 校验长度合法性：声明的长度 = 实际帧长度 - 4（长度字段本身）
            if (msgLength + FIELD_LENGTH != frame.size()) {
                system.log().warn("[帧解析] 长度不匹配：声明{} vs 实际{}",
                        msgLength, frame.size() - FIELD_LENGTH);
                return null;
            }

            // 2. 提取消息ID（int32，大端字节序）
            int msgId = buffer.getInt(MSG_ID_OFFSET);

            // 3. 提取消息主体（从8字节偏移开始，长度为msgLength-4）
            ByteBuffer msgBody = buffer.slice(MSG_BODY_OFFSET, msgLength - 4);

            system.log().info("[帧解析] 成功 - ID：{}，总长度：{}，主体长度：{}",
                    msgId, frame.size(), msgBody.remaining());
            return new MessageFrame(msgId, ByteString.fromByteBuffer(msgBody));

        } catch (Exception e) {
            system.log().error("[帧解析] 失败", e);
            return null;
        }
    }

    /**
     * 消息帧, 也就是客户端提交的数据流单帧
     *
     * @param id
     * @param message
     */
    public record MessageFrame(
            int id,
            ByteString message
    ) {
    }
}

/**
 * 创建消息分帧管道
 */
private Flow<Message, Message, ?> createFrameFlow() {
    // 采用 pekko 内部封装更加易用的 Framing.simpleFramingProtocol
    // 内部其实就是上面提出的 Framing.lengthField(4,0,MAX,ORDER) 简易封装
    // 好处就是零手动缓存、低风险、易维护
    // 构建双向处理的简易分帧协议
    // 设置了最大帧长度(超过抛异常，防OOM)
    BidiFlow<ByteString, ByteString, ByteString, ByteString, NotUsed> bidiFraming =
            Framing.simpleFramingProtocol(MAX_FRAME_LENGTH);

    // BidiFlow 转为单向解码 Flow(仅保留入站解码，忽略出站编码)
    Flow<ByteString, ByteString, ?> frameDecoder = bidiFraming.join(Flow.create())
            // 拦截分帧异常（超大载荷/非法长度）
            .recover(Throwable.class, () -> {
                system.log().error("[分帧异常] 载荷超限/格式错误");
                return ByteString.emptyByteString();
            });

    // 之后就是使用 frameDecoder 照常解码
    // 其他略
}