package com.fenydata.amqpdemo;

import cn.hutool.core.util.HexUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.rabbitmq.client.DeliverCallback;
import org.iot.AmqpClient;
import org.iot.dto.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Configuration
public class SdkDemo {

    private final static Logger logger = LoggerFactory.getLogger(SdkDemo.class);

    //业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。
    private final static ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000));

    @PostConstruct
    public void init() {
        callback("2025-client-a");
    }

    public void callback(String clientId) {
        logger.info("服务启动开始监听消息------------------------------------------------");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            try {

                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);

                // 1.建议异步处理收到的消息，确保没有耗时逻辑。
                // 如果业务处理耗时过程过长阻塞住线程，可能会可能会有oom发生。
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        processMessage(message);
                    }
                });


            } catch (Exception e) {
                logger.error("sdk收到消息，处理消息异常:{}", e.getMessage());
            }
        };

        // 监听所有队列的消息
        AmqpClient.consumer("AccessKeyId", "AccessKeySecret", "实例id", "消费组id", clientId, deliverCallback, "地区");

        // 监听指定队列的消息
        // List<String> needQueue = new ArrayList<>();
        // needQueue.add("qYJ3iFdY***.up_raw"); // 透传上报
        // AmqpClient.consumerNeedQueue("AccessKeyId", "AccessKeySecret", "实例id", "消费组id", clientId, deliverCallback, "地区", needQueue);


    }

    /**
     * 在这里处理您收到消息后的具体业务逻辑。
     */
    private static void processMessage(String message) {
        try {
            logger.info("收到消息，message:{}", message);

            JSONObject jsonObject = JSONUtil.parseObj(message);
            System.out.println(jsonObject);

            Message msg = JSONUtil.toBean(message, Message.class);
            // TODO: 在这里处理您收到消息后的具体业务逻辑。
            System.out.println("generateTime:" + msg.getGenerateTime());
            System.out.println("traceId:" + msg.getTraceId());
            System.out.println("messageId:" + msg.getMessageId());
            System.out.println("instanceId:" + msg.getInstanceId());
            System.out.println("productKey:" + msg.getProductKey());
            System.out.println("deviceName:" + msg.getDeviceName());
            System.out.println("topic:" + msg.getTopic());
            System.out.println("content:" + new String(msg.getContent())); // 默认使用此方式
            //System.out.println("content:" + HexUtil.encodeHexStr(msg.getContent())); // 透传/自定义Topic且数据为二进制，使用本方式

            System.out.println("------------------------------------------------");

            // 设备上下线消息content结构：org.iot.dto.ContentOfOnlineOffLine
            // 设备上报物模型属性消息content结构：org.iot.dto.ContentOfProperty
            // 设备上报物模型事件消息content结构：org.iot.dto.ContentOfEvent
        } catch (Exception e) {
            logger.error("error ", e);
        }
    }


}
