dacoo-data-sdk-yongfeng/src/main/java/com/cisdi/data/caliper/gateway/CaliperDaemonTask.java

package com.cisdi.data.caliper.gateway;

import com.cisdi.data.common.dto.AlarmDto;
import com.cisdi.data.sdk.consts.ServiceName;
import com.cisdi.data.sdk.service.AlarmService;
import lombok.extern.slf4j.Slf4j;

import java.util.*;

@Slf4j
public class CaliperDaemonTask implements Runnable {

    private CaliperClient caliperClient;
    private final CaliperTaskStartVo startVo;

    public CaliperDaemonTask(CaliperClient caliperClient, CaliperTaskStartVo startVo) {
        this.caliperClient = caliperClient;
        this.startVo = startVo;
    }

    @Override
    public void run() {
        final String[] channelInfo = {null};
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                channelInfo[0] = startChannel();
                if (!startVo.getShouldRun().get()) {
                    timer.cancel();
                }
            }
        }, 0, startVo.getSocketParam().getPullInternal());
        if (caliperClient != null) {
            caliperClient.close();
            caliperClient = null;
        }
        log.info("退出Caliper拉取连接线程 {}", channelInfo[0]);
    }

    private String startChannel() {
        String channelInfo = String.format("channel:%s, active:%s,open:%s,writable:%s", caliperClient.getChannel(),
                caliperClient.getChannel() == null ? "" : caliperClient.getChannel().isActive(),
                caliperClient.getChannel() == null ? "" : caliperClient.getChannel().isOpen(),
                caliperClient.getChannel() == null ? "" : caliperClient.getChannel().isWritable());
        try {
            if (caliperClient != null && caliperClient.getChannel() != null
                    && caliperClient.getChannel().isActive()) {
                if (startVo.getGatewayBase() != null
                        && Boolean.TRUE.equals(startVo.getGatewayBase().getInstanceVo().getLogOpen())) {
                    log.info("当前Caliper拉取连接通道处于连接状态，连接id:{},连接信息:{}",
                            startVo.getGatewayBase().getInstanceVo().getRunId(), channelInfo);
                }

                if (startVo.getCustomParam() != null
                        && Boolean.TRUE.equals(startVo.getCustomParam().getEnableHeartbeat())
                        && Math.abs(System.currentTimeMillis() - caliperClient.getLastHeartbeatTime()) >
                        (long) startVo.getCustomParam().getHeartbeatMaxLostCount() * startVo.getCustomParam().getHeartbeatTimeInMs()) {
                    log.warn("当前Caliper拉取连接通道心跳启用，且心跳超时，触发重连，连接id:{},连接信息:{}",
                            startVo.getGatewayBase().getInstanceVo().getRunId(), channelInfo);
                    //启用了心跳，且心跳时间超时
                    reconnect(channelInfo, true);
                }
            } else {
                assert caliperClient != null;
                if (caliperClient.getChannel() == null) {
                    String errorMsg = String.format("Caliper到%s:%s的连接未建立或已关闭,  %s", caliperClient.getIp(),
                            caliperClient.getPort(), channelInfo);
                    log.warn(errorMsg);
                    sendAlarmToPlatform(errorMsg);
                    reconnect(channelInfo, false);
                } else {
                    String errorMsg = String.format("Caliper到%s:%s的连接未建立或已关闭,  %s", caliperClient.getIp(),
                            caliperClient.getPort(), channelInfo);
                    log.warn(errorMsg);

                    sendAlarmToPlatform(errorMsg);

                    if (startVo.getCustomParam() != null
                            && Boolean.TRUE.equals(startVo.getCustomParam().getEnableHeartbeat())
                            && Math.abs(System.currentTimeMillis() - caliperClient.getLastHeartbeatTime()) >
                            (long) startVo.getCustomParam().getHeartbeatMaxLostCount() * startVo.getCustomParam().getHeartbeatTimeInMs()) {
                        log.warn("当前Caliper拉取连接通道心跳启用，且心跳超时，触发重连，连接id:{},连接信息:{}",
                                startVo.getGatewayBase().getInstanceVo().getRunId(), channelInfo);
                        //启用了心跳，且心跳时间超时
                        reconnect(channelInfo, true);
                    }
                }
            }
        } catch (Exception e) {
            log.warn(e.getMessage(), e);
        }
        return channelInfo;
    }

    private void reconnect(final String copyChannelInfo, boolean forceConnect) throws InterruptedException {
        Thread.sleep(5000);
        Thread thread = new Thread(() -> {
            try {
                log.info("Caliper触发重连强制:{}....{}:{} {}", forceConnect, caliperClient.getIp(), caliperClient.getPort(),
                        copyChannelInfo);
                if (caliperClient.getChannel() == null || forceConnect) {
                    caliperClient.reconnect();
                }
            } catch (Exception e) {
                log.info(e.getMessage(), e);
            }
        });
        thread.start();
    }

    private void sendAlarmToPlatform(String detail) {
        if (startVo.getServiceProvider() != null && startVo.getGatewayBase() != null) {
            AlarmService alarmService = (AlarmService) startVo.getServiceProvider().getByName(ServiceName.Alarm);
            if (alarmService != null) {
                AlarmDto alarmDto = new AlarmDto();

                alarmDto.setAlarmType(AlarmDto.Connect_Fail_Alarm);
                alarmDto.setDetail(detail);
                alarmDto.setDeviceId(startVo.getDeviceId());
                alarmDto.setGatewayId(startVo.getGatewayBase().getInstanceVo().getRunId());
                alarmDto.setGatewayName(startVo.getGatewayBase().getInstanceVo().getName());
                alarmDto.setMsgKey(null);
                alarmDto.setUpper(Boolean.TRUE);
                alarmDto.setTime(new Date());

                alarmService.raiseAlarm(Collections.singletonList(alarmDto));
            }
        }
    }
}