package com.yeejoin.equip.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
import com.yeejoin.equip.entity.EquipmentIndexVO;
import com.yeejoin.equip.service.KafkaMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author LiuLin
 * @date 2023年07月12日 10:44
 */
@Slf4j
@Service
public class KafkaMessageServiceImpl implements KafkaMessageService {
    private Executor dataExecutor = new ThreadPoolTaskExecutor();

    @Autowired
    private InfluxDbConnection influxDbConnection;

    @Value("${kafka.alarm.topic}")
    private String alarmTopic;

    //iot转发实时消息存入influxdb前缀
    private static final String MEASUREMENT = "iot_data_";
    //装备更新最新消息存入influxdb前缀
    private static final String INDICATORS = "indicators_";
    //装备更新最新消息存入influxdb固定时间
    private static final Long TIME = 1688558007051L;

    @Override
    public void handlerMessage(String message, Map<Object, Object> equipmentIndexVOMap) {
        dataExecutor.execute(new Runnable() {
            @Override
            public void run() {
                JSONObject jsonObject = JSONObject.parseObject(message);
                String dataType = jsonObject.getString("dataType");
                String indexAddress = jsonObject.getString("address");
                String traceId = jsonObject.getString("traceId");
                String gatewayId = jsonObject.getString("gatewayId");
                String value = jsonObject.getString("value");
                String key = indexAddress + "_" + gatewayId;

                try {
                    if (equipmentIndexVOMap.get(key) != null) {
                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        EquipmentIndexVO equipmentSpeIndex = (EquipmentIndexVO) equipmentIndexVOMap.get(key);
                        log.info("接收到iot消息: 指标名称:{},地址:{},值:{},网关{}",
                                equipmentSpeIndex.getEquipmentIndexName(),indexAddress,value,gatewayId);

                        Map<String, String> tagsMap = new HashMap<>();
                        Map<String, Object> fieldsMap = new HashMap<>();
                        tagsMap.put("equipmentsIdx", key);
                        tagsMap.put("address", indexAddress);
                        tagsMap.put("gatewayId", gatewayId);
                        tagsMap.put("dataType", dataType);
                        tagsMap.put("isAlarm", String.valueOf(equipmentSpeIndex.getIsAlarm()));
                        tagsMap.put("equipmentSpecificName", equipmentSpeIndex.getEquipmentSpecificName());

                        String valueLabel = valueTranslate(value, equipmentSpeIndex.getValueEnum());
                        fieldsMap.put("traceId", traceId);
                        fieldsMap.put("value", value);
                        fieldsMap.put("valueLabel", valueLabel.equals("") ? value : valueLabel);
                        fieldsMap.put("equipmentIndexName", equipmentSpeIndex.getEquipmentIndexName());
                        fieldsMap.put("unit", equipmentSpeIndex.getUnitName());
                        fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
                        fieldsMap.put("equipmentIndex", JSON.toJSONString(equipmentSpeIndex));

                        influxDbConnection.insert(MEASUREMENT + gatewayId, tagsMap, fieldsMap);
                        influxDbConnection.insert(INDICATORS + gatewayId, tagsMap, fieldsMap, TIME, TimeUnit.MILLISECONDS);
                        if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
                            fieldsMap.putAll(tagsMap);
                            //kafkaProducerService.sendMessageAsync(alarmTopic,JSON.toJSONString(fieldsMap));
                        }
                    }
                } catch (Exception e) {
                    log.error("Iot透传消息解析入库失败" + e.getMessage(), e);
                }
            }
        });

    }

    private String valueTranslate(String value, String enumStr) {
        if (ObjectUtils.isEmpty(enumStr)) {
            return "";
        }
        try {
            JSONArray jsonArray = JSONArray.parseArray(enumStr);
            for (int i = 0; i < jsonArray.size(); i++) {
                JSONObject jsonObject = jsonArray.getJSONObject(i);
                if (jsonObject.get("key").equals(value)) {
                    return jsonObject.getString("label");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "";
    }

    @PostConstruct
    public void iotAsyncExecutor() {
        ThreadPoolTaskExecutor workExecutor = new ThreadPoolTaskExecutor();
        // 设置核心线程数
        int length = Runtime.getRuntime().availableProcessors();
        int size = Math.max(length, 80);
        workExecutor.setCorePoolSize(size * 2);
        log.info("装备服务初始化，系统线程数：{},运行线程数:{}", length, size);
        // 设置最大线程数
        workExecutor.setMaxPoolSize(workExecutor.getCorePoolSize());
        //配置队列大小
        workExecutor.setQueueCapacity(Integer.MAX_VALUE);
        // 设置线程活跃时间（秒）
        workExecutor.setKeepAliveSeconds(60);
        // 设置默认线程名称
        workExecutor.setThreadNamePrefix("装备服务-Iot透传消息消费线程池" + "-");
        // 等待所有任务结束后再关闭线程池
        //当调度器shutdown被调用时，等待当前被调度的任务完成
        workExecutor.setWaitForTasksToCompleteOnShutdown(true);
        //执行初始化
        workExecutor.initialize();
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
        workExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        this.dataExecutor = workExecutor;
    }
}
