package com.muyu.mqtt;

import com.alibaba.fastjson2.JSONObject;

import com.muyu.domain.CarMessage;
import com.muyu.domain.KafKaData;
import com.muyu.service.CarMessageService;
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.stereotype.Component;

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


/**
 * mqtt
 *
 * @ClassName MqttTest
 * @Description
 * @Date 2024/9/28 23:49
 */
@Slf4j
@Component
public class MqttTest {
    private static final Integer ID = 1;
    private static final Integer CODE = 1;

    @Resource
    private KafkaProducer<String, String> kafkaProducer;
    @Resource
    private CarMessageService sysCarMessageService;

    @PostConstruct
    public void Test() {
        String topic = "vehicle";
        String content = "Message from MqttPublishSample";
        int qos = 2;
        String broker = "tcp://60.204.221.52: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 {
                    List<CarMessage> list = sysCarMessageService.selectCarMessage(1);
                    String string = new String(mqttMessage.getPayload());
                    log.info(new String(mqttMessage.getPayload()));
                    List<KafKaData> kafKaDataList = new ArrayList<>();
                    String[] test = string.split(" ");
//                    String[] results = new String[list.size()];
                    for (CarMessage carMessage : list) {
                        int startIndex = Integer.parseInt(String.valueOf(carMessage.getCarMessageStartIndex())) - 1;
                        int endIndex = Integer.parseInt(String.valueOf(carMessage.getCarMessageEndIndex()));
                        StringBuilder hexBuilder = new StringBuilder();
                        for (int i = startIndex; i < endIndex; 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("String")
                                .build());
                    }
                    String jsonString = JSONObject.toJSONString(kafKaDataList);
                    ProducerRecord<String, String> producerRecord = new ProducerRecord<>("carJsons",jsonString);

                    kafkaProducer.send(producerRecord);
                    log.info("kafka投产：{}", jsonString);
//                    HashMap<String, String> stringStringHashMap = new HashMap<>();
//                    kafKaDataList.forEach(data -> stringStringHashMap.put(data.getKey(), data.getValue()));
//                    jsonString = JSONObject.toJSONString(stringStringHashMap);
//                    System.out.println(jsonString);
                }

                // 接收信息
                @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();
        }
    }
}

