package com.muyu.parsing.manager;

import com.alibaba.fastjson2.JSONObject;
import com.muyu.common.kafka.constants.KafkaConstants;
import com.muyu.parsing.domain.KafKaData;
import com.muyu.parsing.domain.SysCarMessage;
import com.muyu.parsing.service.impl.SysCarMessageServiceImpl;
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.MqttMessage;
import org.springframework.stereotype.Service;

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

@Slf4j
@Service
public class MessageProcessor {


    private static final int ID = 1;
    @Resource
    private SysCarMessageServiceImpl sysCarMessageService;
    @Resource
    private KafkaProducer<String, String> kafkaProducer;


    public void processMessage(MqttMessage mqttMessage) {
        String payload = new String(mqttMessage.getPayload());
        log.info("====:{}", payload);

        List<SysCarMessage> carMessages = sysCarMessageService.selectSysCarMessageLists(ID);
        List<KafKaData> kafKaDataList = new ArrayList<>();

        String[] test = payload.split(" ");
        for (SysCarMessage carMessage : carMessages) {
            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 = JSONObject.toJSONString(kafKaDataList);
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.KafkaTopic, jsonString);
        kafkaProducer.send(producerRecord);
        log.info("kafka投产：{}", jsonString);
    }
}

