package com.muyu.cloud.protocol.parsing.test;

import cn.hutool.json.JSONObject;
import com.muyu.domain.Vehicle;
import com.muyu.domain.VehicleType;
import com.muyu.domain.resp.MessageValueListResp;
import com.muyu.enterprise.cache.AllMessageValueCacheService;
import com.muyu.enterprise.cache.VehicleCacheService;
import com.muyu.enterprise.cache.VehicleTypeCacheService;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

/**
 * @Author: LiDongJia
 * @Package: com.muyu.cloud.protocol.parsing.service.impl
 * @Project: 2112-car-cloud-server
 * @name: ParsingServiceImpl
 * @Date: 2024/9/28  14:31
 * @Description: 协议解析实现层
 */
@Log4j2
@Component
public class ParsingMessage {

    @Resource
    private KafkaProducer<String, String> kafkaProducer;

    @Autowired
    private VehicleCacheService vehicleCacheService;

    @Autowired
    private VehicleTypeCacheService vehicleTypeCacheService;

    @Autowired
    private AllMessageValueCacheService allMessageValueCacheService;

    /**
     * 协议解析
     */
    @PostConstruct
    public void mqttClient() {
        String topic = "vehicle";
        String broker = "tcp://111.231.50.146:1883";
        String clientId = "JavaSample";

        try {
            // 第三个参数为空，默认持久化策略
            MqttClient sampleClient = new MqttClient(broker, clientId);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker: " + broker);
            sampleClient.connect(connOpts);
            sampleClient.subscribe(topic, 0);
            sampleClient.setCallback(new MqttCallback() {
                // 连接丢失
                @Override
                public void connectionLost(Throwable throwable) {

                }

                // 连接成功
                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    System.out.println(new String(mqttMessage.getPayload()));
                    JSONObject entries = this.protocolParsing(new String(mqttMessage.getPayload()));

                    ProducerRecord<String, String> producerRecord = new ProducerRecord<>("zeshi",
                            entries.toString()      );
                    kafkaProducer.send(producerRecord);
                    System.out.println(entries);
                }

                /**
                 * 协议解析
                 * @param messageStr
                 * @return
                 */
                public JSONObject protocolParsing(String messageStr) {
                    //根据空格切割数据
                    String[] hexArray = messageStr.split(" ");
                    StringBuilder result = new StringBuilder();
                    //遍历十六进制数据转换为字符
                    for (String hex : hexArray) {
                        int decimal = Integer.parseInt(hex, 16);
                        result.append((char) decimal);
                    }
                    //取出车辆VIN码
                    String vehicleVin = result.substring(1, 18);
                    log.info("车辆VIN码: " + vehicleVin);
                    //根据车辆VIN码查询报文模板ID
                    Vehicle vehicle = vehicleCacheService.get(vehicleVin);
                    Long vehicleTypeId = vehicle.getVehicleTypeId();
                    VehicleType vehicleType = vehicleTypeCacheService.get(String.valueOf(vehicleTypeId));
                    Long templateId = vehicleType.getMessageTemplateId();
                    List<MessageValueListResp> templateList = allMessageValueCacheService.get(String.valueOf(templateId));
//                    //从redis缓存中获取报文模板数据
//                    try {
//                        String redisKey = "messageTemplate" + templateId;
//                        if (redisTemplate.hasKey(redisKey)) {
//                            List<Object> list = redisTemplate.opsForList().range(redisKey, 0, -1);
//                            templateList = list.stream()
//                                    .map(obj -> JSON.parseObject(obj.toString(), MessageValueListResp.class))
//                                    .toList();
//                            log.info("Redis缓存查询成功");
//                        } else {
//                            Result<List<MessageValueListResp>> byTemplateId = remoteServiceClient.findByTemplateId(templateId);
//                            templateList = byTemplateId.getData();
//                            templateList.forEach(
//                                    listResp ->
//                                            redisTemplate.opsForList().rightPush(
//                                                    redisKey, JSON.toJSONString(listResp)
//                                            )
//                            );
//                            log.info("数据库查询成功");
//                        }
//                    } catch (Exception e) {
//                        throw new RuntimeException("获取报文模板失败");
//                    }
                    //判断报文模板列表不为空
                    if (templateList.isEmpty()) {
                        throw new RuntimeException("报文模版为空");
                    }
                    //存储报文模版解析后的数据
                    JSONObject jsonObject = new JSONObject();
                    for (MessageValueListResp messageValue : templateList) {
                        //起始位下标
                        Integer startIndex = messageValue.getMessageStartIndex() - 1;
                        //结束位下标
                        Integer endIndex = messageValue.getMessageEndIndex();
                        //根据报文模版截取数据
                        String value = result.substring(startIndex, endIndex);
                        //存入数据
                        jsonObject.put(messageValue.getMessageLabel(), value);
                    }
                    return jsonObject;
                }

                // 接收信息
                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

                }
            });
        } catch (MqttException 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();
        }
    }
}
