package com.muyu.parsing.consumer;

import com.muyu.cargateway.domain.properties.MqttProperties;
import com.muyu.common.kafka.constants.KafkaConstants;
import com.muyu.parsing.domain.KafKaData;
import com.muyu.parsing.domain.SysCarMessage;
import com.muyu.parsing.manager.TaskManager;
import com.muyu.parsing.service.impl.SysCarMessageServiceImpl;
import com.rabbitmq.client.Channel;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName FormMessageConsumer
 * @Description 描述
 * @Author Chen
 * @Date 2024/10/9 15:55
 */
@Component
@Slf4j
public class FormMessageConsumer {
    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private KafkaProducer<String, String> kafkaProducer;
    @Resource
    private SysCarMessageServiceImpl sysCarMessageService;
    private final static String FORM_QUEUE = "queue_inform_sms";

    @RabbitListener(queuesToDeclare = {@Queue(FORM_QUEUE)})
    public void downline(MqttProperties mqttProperties, Message message, Channel channel) {
        log.info("基础信息 {} 。。。", mqttProperties);
        try {
            // 第三个参数为空，默认持久化策略
            MqttClient sampleClient = new MqttClient(mqttProperties.getBroker(), mqttProperties.getClientId());
            MqttConnectOptions connOpts = new MqttConnectOptions();
            //用户名
            connOpts.setUserName(mqttProperties.getUserName());
            //密码
            connOpts.setPassword(mqttProperties.getPassword().toCharArray());
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker: " + mqttProperties.getBroker());
            sampleClient.connect(connOpts);
            sampleClient.subscribe(mqttProperties.getTopic(), 0);
            sampleClient.setCallback(new MqttCallback() {
                // 连接丢失
                @Override
                public void connectionLost(Throwable throwable) {
                    log.error("连接丢失：{}", throwable.getMessage());
                }

                // 连接成功
                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) {
                    TaskManager manager = new TaskManager(10);
                    manager.execute(() -> processMessage(mqttMessage));
                }

                // 接收信息
                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    log.info("接收消息：{}", iMqttDeliveryToken);
                }
            });
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (MqttException | IOException me) {
//            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
            throw new RuntimeException(me);
        }
    }


    private void processMessage(MqttMessage mqttMessage) {
        String string = new String(mqttMessage.getPayload());
        log.info(string);
        List<SysCarMessage> list = sysCarMessageService.selectSysCarMessageLists(1);
        List<KafKaData> kafKaDataList = new ArrayList<>();
        String[] test = string.split(" ");
        for (SysCarMessage carMessage : list) {
            int start = Integer.parseInt(carMessage.getMessageStartIndex()) - 1;
            int end = Integer.parseInt(carMessage.getMessageEndIndex());
            StringBuilder hexBuilder = new StringBuilder();
            for (int i = start; i < end; i++) {
                hexBuilder.append(test[i]);
            }
            String hex = hexBuilder.toString();
            char[] result = new char[hex.length() / 2];
            for (int x = 0; x < hex.length(); x += 2) {
                int high = Character.digit(hex.charAt(x), 16);
                int low = Character.digit(hex.charAt(x + 1), 16);
                result[x / 2] = (char) ((high << 4) + low);
            }
            String value = new String(result);
            kafKaDataList.add(KafKaData.builder()
                    .key(carMessage.getMessageTypeCode())
                    .label(carMessage.getMessageTypeCode())
                    .value(value)
                    .type(carMessage.getMessageType())
                    .build());
        }
        kafKaDataList.add(KafKaData.builder()
                .key("firmCode")
                .label("企业编码")
                .value("firm01")
                .type("String")
                .build());

        String jsonString = com.alibaba.fastjson.JSONObject.toJSONString(kafKaDataList);
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.KafkaTopic, jsonString);
        kafkaProducer.send(producerRecord);
        log.info("kafka投产：{}", jsonString);
    }
}
