import org.apache.pekko.Done;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.stream.*;
import org.apache.pekko.stream.javadsl.*;
import org.apache.pekko.util.ByteString;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.CompletionStage;

/**
 * Pekko Stream 建议交互
 */
public class PekkoStreamExample {

    /**
     * 启动入口
     *
     * @param args 参数列表
     */
    public static void main(String[] args) {

        // 创建 ActorSystem（Pekko Stream 的运行时依赖）
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "SystemIO");

        // 获取 Stream 执行器 Materializer
        // Materializer 可以复用, 如果是频繁需要异步调用最好实例化句柄持有
        Materializer materializer = Materializer.createMaterializer(system);

        // 定义 Source: 包装 System.in 为数据生产者，按行读取
        InputStream inputStream = System.in;

        // 只要支持 Java 的 InputStream 的对象都可自己包装成 Source
        // 用 StreamConverters 包装 System.in(InputStream)
        // 建议观察下 StreamConverters 内部, 其中内部实现大量流转化工具
        Source<ByteString, ?> inputSource = StreamConverters.fromInputStream(
                        () -> inputStream, // 延迟提供 InputStream 实例
                        8192 // 读取缓冲区大小，按需调整（平衡性能和内存占用）
                )
                // 行分割 + 背压控制: 限制每行最大 1024 字节
                .via(Framing.delimiter(
                        ByteString.fromString("\n"),
                        1024,
                        FramingTruncation.ALLOW
                ))
                // 配置输入缓冲区, 强化背压(缓冲区满则暂停读取)
                .withAttributes(Attributes.inputBuffer(1, 1));


        // 中间处理流：过滤空行
        Flow<String, String, ?> processFlow = Flow.of(String.class)
                .filter(line -> !line.trim().isEmpty())
                .map((line) -> {
                    // 这里可以中间处理下, 比如在输出的时候追加特殊符号
                    return "Robot: %s".formatted(line);
                });


        // 创建共享的退出指令, 用于手动优雅退出数据流
        SharedKillSwitch killSwitch = KillSwitches.shared("quit-signal");


        // 输出 Sink: 打印到控制台
        // 定义输入处理进程流的处理
        Sink<String, CompletionStage<Done>> outputSink = Sink.foreach(
                line -> {
                    // 检测是否匹配 >>> quit 指令, 如果检测到就执行退出逻辑
                    if ("Robot: quit".equalsIgnoreCase(line)) {
                        killSwitch.shutdown(); // 主动关闭流
                        return;
                    }
                    // 照常输出
                    System.out.println(line);
                }
        );


        System.out.println("数据回显程序已启动");
        System.out.println("输入 quit 即可退出程序\n");


        // 构建数据流管道并启动
        CompletionStage<Done> done = inputSource
                // 字节流转字符串
                .map(byteString -> byteString.decodeString(StandardCharsets.UTF_8))
                .via(killSwitch.flow()) // 将 KillSwitch 接入流监听
                .via(processFlow)
                // Keep.right() 保留右侧 Sink 的 materialized 值
                .toMat(outputSink, Keep.right())
                .run(materializer);

        // 监听流结束/异常，释放资源
        done.whenComplete((result, exception) -> {
            if (Objects.nonNull(exception)) {
                System.err.println("数据流异常终止: " + exception.getMessage());
            } else {
                System.out.println("数据流正常结束");
            }
            system.terminate();
        });
    }
}

public static void main(String[] args) {
    // 1. 创建唯一的运行时 Materializer
    ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "demo");
    Materializer materializer = Materializer.createMaterializer(system);

    // 2. 定义固定拓扑
    Source<String, ?> source = Source.single("hello pekko stream");
    Sink<String, CompletionStage<Done>> sink = Sink.foreach(System.out::println);

    // 3. 同一运行时，不同 Keep 选型（只改变物化值，不改变运行时）
    // 选型1：Keep.right() → 拿 Sink 的物化值（监听结果）
    CompletionStage<Done> sinkMat = source.toMat(sink, Keep.right()).run(materializer);

    // 选型2：Keep.left() → 拿 Source 的物化值（这里 Source 物化值是 NotUsed, 其实也是类似的 Void）
    Object sourceMat = source.toMat(sink, Keep.left()).run(materializer);

    // 选型3：Keep.both() → 同时拿两个物化值
    var bothMat = source.toMat(sink, Keep.both()).run(materializer);

    // 4. 监听结果（验证运行时生效）
    sinkMat.whenComplete((done, ex) -> {
        System.out.println("流执行完成");
        system.terminate();
    });
}

/**
 * 这里引出了 pekko 的 Graph 概念
 * @param <In> 流输入产物
 * @param <Out> 流输出产物
 * @param <Mat> 物料值
 */
public final class Flow<In, Out, Mat> implements Graph<FlowShape<In, Out>, Mat> {
    // 其他略
}

/**
 * 空节点, 以下会直接跳出错误
 * <p>
 * 这里只有节点而没有边, 起点节点(Source)没有边连接到任何终点节点(Sink), 属于 "不可达的孤立节点", 拓扑无效
 */
void emptyVertex() {
    // Source 没有连接到 Sink，拓扑不闭合
    Source<String, NotUsed> source = Source.single("hello");
    // 没有调用 to()/via() 连接 Sink，运行时会提示“拓扑未闭合”
    source.run(materializer); // 编译/运行报错
}


/**
 * 匹配错误
 * <p>
 * 边的两端节点 "数据类型不兼容", 相当于数学图中 "边连接了两个不匹配的节点"
 */
void matchError() {
    // 错误示例：Flow 输入是 String，Source 输出是 Integer，类型不匹配
    Source<Integer, NotUsed> source = Source.single(123);
    Flow<String, String, NotUsed> flow = Flow.of(String.class).map(s -> s.toUpperCase());
    source.via(flow).to(Sink.foreach(System.out::println)); // 编译报错
}

import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.stream.Materializer;
import org.apache.pekko.stream.OverflowStrategy;
import org.apache.pekko.stream.javadsl.*;
import org.apache.pekko.util.ByteString;
import org.slf4j.Logger;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.util.Objects;

public class PekkoStreamQueueExample {

    /**
     * 功能入口
     */
    public static void main(String[] args) throws IOException {
        // 创建唯一的运行时 Materializer
        final ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "demo");
        final Materializer materializer = Materializer.createMaterializer(system);// 可以持有复用
        final Logger logger = system.log();


        // 入站数据处理 START -------------------------------------------------------------------------------

        // 创建入站消息通道, 第二个参数代表队列满了的策略
        // backpressure(): 自动回滚覆盖数据
        // dropTail(): 删除队列尾数据
        // dropHead(): 删除队列头数据
        SourceQueueWithComplete<ByteString> inbound = Source.<ByteString>queue(
                1024, // 队列长度
                OverflowStrategy.backpressure() // 队列满了之后的策略, 背压策略会回滚最早消息覆盖
        ).via(Framing.delimiter( // 这里也可以采用 Framing.lengthField 方式
                ByteString.fromString("\n"),// 数据分隔符
                1024, // 最大缓冲区
                FramingTruncation.ALLOW // 指定当输入数据流被截断无法处理时候策略: DISALLOW - 中断处理/ ALLOW - 直接丢弃
        )).to(Sink.foreach(message -> {
            // 捕获数据流转, 这里就可以考虑转发给特定 Actor
            // 比如 getSelf().tell(new FrameReceived(frame), getSelf());
            logger.info("Message received: {}", message);

        })).run(materializer); // 挂载运行时


        // 然后外部的 CMD/TCP/UDP/WebSocket 转发服务端数据就直接写入队列
        // 下面的数据因为没有分隔符直接被过滤跳过
        inbound.offer(ByteString.fromString("Hello World! ERROR!")) // 这里就是将数据推入队列
                .whenComplete((res, throwable) -> {
                    if (Objects.nonNull(throwable)) {
                        logger.error("push queue error", throwable); // 写入队列失败
                    } else {
                        logger.info("push queue success"); // 写入队列成功
                    }
                });

        // 这里追加 \n 符号用于分隔, 这条数据才会被捕获
        inbound.offer(ByteString.fromString("Hello World!Hello!!!\n")) // 这里就是将数据推入队列
                .whenComplete((res, throwable) -> {
                    if (Objects.nonNull(throwable)) {
                        logger.error("push queue error", throwable); // 写入队列失败
                    } else {
                        logger.info("push queue success"); // 写入队列成功
                    }
                });


        // 入站数据处理 END  -------------------------------------------------------------------------------


        // 出站数据处理 START -------------------------------------------------------------------------------

        // 其实思路也是一样, 只是反过来处理
        // 还是定义处理队列, 只是队列输出的地方变成客户端的 Socket 句柄
        SourceQueueWithComplete<ByteString> outbound = Source.<ByteString>queue(
                1024, // 队列长度
                OverflowStrategy.backpressure() // 队列满了之后的策略, 背压策略会回滚最早消息覆盖
        ).via(Framing.lengthField( // 这里采用 Framing.lengthField 方式
                Integer.BYTES, // sizeof(int32) 长度作为数据长度单位即可, 一般等于4
                0, // 长度字段的偏移, 一般长度都是放置于首位
                256 * 1024, // 数据帧数量最大值
                ByteOrder.BIG_ENDIAN // 网络序列是大端序列, 注意: 不声明默认采用小端序列, 会导致数据转化失败
        )).to(Sink.foreach(message -> {
            // 捕获数据流转, 这里就可以考虑转发给特定 Actor
            // 比如 getSelf().tell(new FrameReceived(frame), getSelf());
            logger.info("Frame message: {}", message);

        })).run(materializer); // 挂载运行时

        // 最后直接推入数据队列即可
        String msgBody = "Hello World!Frame!";
        byte[] msgBodyBytes = msgBody.getBytes(StandardCharsets.UTF_8);
        int msgId = 10001;

        // 构建数据二进制
        // 构建二进制帧：[总长度(4字节)] + [msgId(4字节)] + [消息体]
        ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES + Integer.BYTES + msgBodyBytes.length);
        buffer.putInt(msgBodyBytes.length + Integer.BYTES); // 长度字段只需要排除首位 int32 内容长度字段, 也就是 length = id + body 长度即可
        buffer.putInt(msgId);
        buffer.put(msgBodyBytes);
        ByteString bytes = ByteString.fromArray(buffer.array());

        // 最后投敌过去
        outbound.offer(bytes).whenComplete((res, throwable) -> {
            if (Objects.nonNull(throwable)) {
                logger.error("push queue error", throwable);
            } else {
                logger.info("push queue success");
            }
        });


        // 出站数据处理 END  -------------------------------------------------------------------------------


        // 最后执行, 确认队列执行被消费, 一般退出的时候需要提醒快点消费
        inbound.complete();
        outbound.complete();
        System.out.println("任意按键即可退出程序\n");
        int ignore = System.in.read();
        system.terminate();
    }
}