package com.fjhx.iot.listener;

import cn.hutool.core.io.IoUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONReader;
import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
import com.fjhx.common.constant.SourceConstant;
import com.fjhx.iot.constants.RedisConstant;
import com.fjhx.iot.entity.tda.dto.EbcDto;
import com.fjhx.iot.entity.tda.dto.MessageBody;
import com.fjhx.iot.entity.tda.po.TdaConfig;
import com.fjhx.iot.entity.tda.po.TdaProduct;
import com.fjhx.iot.service.tda.TdaConfigService;
import com.fjhx.iot.service.tda.TdaProductService;
import com.fjhx.iot.utils.amqp.AmqpClient;
import com.fjhx.iot.utils.amqp.AmqpClientOptions;
import com.fjhx.iot.utils.amqp.AmqpConstant;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.framework.mybatis.holder.TenantHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * iot数据推送监听器
 *
 * @author zlj
 */
@Slf4j
@Component
public class DtaListener {

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private TdaConfigService tdaConfigService;

    @Autowired
    private TdaProductService tdaProductService;

    public static final Map<Long, MessageConsumer> listenerMap = new ConcurrentHashMap<>();

//    @PostConstruct
    public void listener() {

        TenantHolder.setIgnore(true);
//        DynamicDataSourceContextHolder.push(SourceConstant.IOT);

        List<TdaConfig> list = tdaConfigService.list();

        for (TdaConfig tdaConfig : list) {
            MessageConsumer consumer = getMessageConsumer(tdaConfig);
            listenerMap.put(tdaConfig.getId(), consumer);
        }

//        DynamicDataSourceContextHolder.clear();
        TenantHolder.clear();

    }

    public MessageConsumer getMessageConsumer(TdaConfig tdaConfig) {

        MessageConsumer consumer = null;
        try {

            AmqpClientOptions options = AmqpClientOptions.builder()
                    .host(tdaConfig.getEndPoint())
                    .port(AmqpConstant.PORT)
                    .accessKey(tdaConfig.getAccessKey())
                    .accessCode(tdaConfig.getAccessCode())
                    .queuePrefetch(1000)
                    .build();

            AmqpClient amqpClient = new AmqpClient(options);
            amqpClient.initialize();
            consumer = amqpClient.newConsumer(AmqpConstant.DEFAULT_QUEUE);
            // 此处进行消息处理。如果处理比较耗时，最好进行开启新的线程处理，否则可能造成心跳超时链接断开。
            consumer.setMessageListener(this::handleData);
            return consumer;
        } catch (Exception e) {
            IoUtil.close(consumer);
            log.error("链接失败，tdaConfigId:{}", tdaConfig.getId(), e);
            throw new ServiceException("连接iot失败");
        }
    }

    /**
     * 处理推送数据
     * <p>
     * remarks: 添加成功必须手动 message.acknowledge();
     * </p>
     *
     * @param message 推送消息
     */
    public void handleData(Message message) {
        try {

            String messageBodyStr = message.getBody(String.class);
            MessageBody messageBody = JSONObject.parseObject(messageBodyStr, MessageBody.class, JSONReader.Feature.SupportSmartMatch);

            if (messageBody == null) {
                log.error(" messageBody == null ");
                return;
            }

            // 事件类型
            String event = messageBody.getEvent();

            switch (event) {
                case "report":
                    reportHandel(messageBody);
                    break;
                default:
                    log.error("未知事件类型：{}\r\n messageBody：{}", event, messageBodyStr);
                    return;
            }

            message.acknowledge();

        } catch (JMSException e) {
            log.error("iot监听发生未知异常", e);
        }

    }

    /**
     * 处理报告数据
     */
    private void reportHandel(MessageBody messageBody) {

        MessageBody.NotifyData notifyData = messageBody.getNotifyData();

        if (notifyData == null) {
            log.error(" notifyData == null , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        MessageBody.Header header = notifyData.getHeader();
        if (header == null) {
            log.error(" header == null , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        MessageBody.Body body = notifyData.getBody();
        if (body == null) {
            log.error(" body == null , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        List<MessageBody.Service> services = body.getServices();
        if (services == null || services.size() == 0) {
            log.error(" services == null || services.size() == 0 , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        // 数据列表
        MessageBody.Service service = services.get(0);

        // 产品id
        String productId = header.getProductId();
        if (StrUtil.isBlank(productId)) {
            log.error(" StrUtil.isBlank(productId) , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        // 设备id
        String deviceId = header.getDeviceId();

        // 数据推送时间
        Date eventTime = messageBody.getEventTime();

        // 实际生产数据
        JSONObject properties = service.getProperties();

        TenantHolder.setIgnore(true);
//        DynamicDataSourceContextHolder.push(SourceConstant.IOT);

        TdaProduct tdaProduct = tdaProductService.getOne(q -> q.eq(TdaProduct::getProductId, productId));
        if (tdaProduct == null) {
            log.error(" tdaProduct == null , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        String deviceType = tdaProduct.getDeviceType();
        if (StrUtil.isBlank(deviceType)) {
            log.error(" StrUtil.isBlank(deviceType) , messageBody = {} ", JSON.toJSONString(messageBody));
            return;
        }

        JSONObject deviceData = properties.getJSONObject("DeviceData");
        switch (deviceType) {
            case "ebc":
                doEbc(productId, deviceId, eventTime, deviceData.toJavaObject(EbcDto.class));
                break;

            // 默认上位机
            default:

                // 设备编号
                String equipmentNo = deviceData.getString("equipmentNo");

                if (equipmentNo != null) {
                    deviceData.put("iot_productId", productId);
                    deviceData.put("iot_deviceId", deviceId);
                    redisCache.setCacheObject(RedisConstant.DEVICE_COMPUTER_PREFIX + ":" + equipmentNo, deviceData);
                }
        }

//        DynamicDataSourceContextHolder.clear();
        TenantHolder.clear();

    }

    /**
     * 经编机
     *
     * @param productId 产品id
     * @param deviceId  设备id
     * @param eventTime 推送时间
     * @param ebcDto    推送数据
     */
    private void doEbc(String productId, String deviceId, Date eventTime, EbcDto ebcDto) {

        EbcDto oldEbcDto = redisCache.getCacheObject(RedisConstant.DEVICE_PREFIX + deviceId);

        if (oldEbcDto != null && eventTime.before(oldEbcDto.getEventTime())) {
            return;
        }

        ebcDto.setDeviceId(deviceId);
        ebcDto.setProductId(productId);
        ebcDto.setEventTime(eventTime);
        redisCache.setCacheObject(RedisConstant.DEVICE_PREFIX + deviceId, ebcDto);

    }

}
