jsowell-charger-web/jsowell-thirdparty/src/main/java/com/jsowell/thirdparty/common/NotificationService.java

package com.jsowell.thirdparty.common;

import com.google.common.collect.Lists;
import com.jsowell.common.enums.thirdparty.ThirdPlatformTypeEnum;
import com.jsowell.common.enums.ykc.ReturnCodeEnum;
import com.jsowell.common.exception.BusinessException;
import com.jsowell.common.service.thirdparty.ThirdPartyPlatformApi;
import com.jsowell.common.util.StringUtils;
import com.jsowell.common.util.spring.SpringUtils;
import com.jsowell.pile.dto.nanrui.PushAlarmInfoDTO;
import com.jsowell.pile.vo.ThirdPartySecretInfoVO;
import com.jsowell.thirdparty.dubbo.factory.DynamicThirdPartyPlatformFactory;
import com.jsowell.thirdparty.platform.service.ThirdPartyPlatformService;
import com.jsowell.thirdparty.platform.factory.ThirdPartyPlatformFactory;
import com.jsowell.thirdparty.service.ThirdpartySecretInfoService;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * 主动通知Service - 异步改造版
 */
@Service
public class NotificationService {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private ThirdpartySecretInfoService thirdpartySecretInfoService;

    // 引入线程池
    private ThreadPoolTaskExecutor executor = SpringUtils.getBean("threadPoolTaskExecutor");


    // 判断是否使用old还是new平台
    private boolean isRemotePlatform(String platformType) {
        for (ThirdPlatformTypeEnum item : ThirdPlatformTypeEnum.values()) {
            if (StringUtils.equals(item.getTypeCode(), platformType)) {
                return true; // true: 使用old平台
            }
        }
        return false; // false: 使用new平台
    }

    /**
     * 根据平台类型获取对应的服务实例
     */
    private Object getPlatformService(String platformType) {
        if (isRemotePlatform(platformType)) {
            return ThirdPartyPlatformFactory.getInvokeStrategy(platformType);
        } else {
            return DynamicThirdPartyPlatformFactory.getPlatformService(platformType);
        }
    }

    /**
     * VO对象转换
     */
    private com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO getConversion(ThirdPartySecretInfoVO secretInfoVO) {
        com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO conversion =
                new com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO();
        BeanUtils.copyProperties(secretInfoVO, conversion);
        return conversion;
    }

    /**
     * 充电站信息变化推送
     */
    public String notificationStationInfo(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return "该站点未绑定任何平台";
        }

        StringBuilder result = new StringBuilder();
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            String currentPlatformType = secretInfoVO.getPlatformType();
            if (StringUtils.isNotBlank(platformType) && !platformType.equals(currentPlatformType)) {
                continue;
            }

            try {
                Object service = getPlatformService(currentPlatformType);

                if (service instanceof ThirdPartyPlatformService) {
                    String pushResult = ((ThirdPartyPlatformService) service).notificationStationInfo(stationId);
                    result.append("old平台[").append(secretInfoVO.getPlatformName())
                            .append("]推送结果：").append(pushResult).append("\n");
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
                        try {
                            return apiService.notificationStationInfo(stationId);
                        } catch (Exception e) {
                            return "推送失败：" + e.getMessage();
                        }
                    }, executor).thenAccept(pushResult -> {
                        synchronized (result) {
                            result.append("new平台[").append(secretInfoVO.getPlatformName())
                                    .append("]推送结果：").append(pushResult).append("\n");
                        }
                    });
                    futures.add(future);
                }
            } catch (Exception e) {
                result.append("平台[").append(secretInfoVO.getPlatformName())
                        .append("]处理异常：").append(e.getMessage()).append("\n");
            }
        }

        // 等待所有异步任务完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        return result.toString();
    }

    /**
     * 设备状态变化推送
     */
    public void notificationStationStatus(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String pileConnectorCode = dto.getPileConnectorCode();
        String status = dto.getStatus();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId) || StringUtils.isBlank(pileConnectorCode) || StringUtils.isBlank(status)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());
                com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO conversionVO = getConversion(secretInfoVO);

                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationStationStatus(
                            stationId, pileConnectorCode, status, secretInfoVO);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationStationStatus(stationId, pileConnectorCode, status, conversionVO);
                        } catch (Exception e) {
                            logger.error("new平台[{}]设备状态推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("平台类型:{}, 平台名称:{}, 站点id:{}, 枪口编号:{}, 设备状态变化推送error:{}",
                        secretInfoVO.getPlatformType(), secretInfoVO.getPlatformName(), stationId, pileConnectorCode, e.getMessage());
            }
        }
    }

    /**
     * 设备充电中状态变化推送
     */
    public void notificationConnectorChargeStatus(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String orderCode = dto.getOrderCode();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId) || StringUtils.isBlank(orderCode)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());
                com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO conversionVO = getConversion(secretInfoVO);

                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationConnectorChargeStatus(orderCode, secretInfoVO);
                    ((ThirdPartyPlatformService) service).notificationEquipChargeStatus(orderCode);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationConnectorChargeStatus(orderCode, conversionVO);
                        } catch (Exception e) {
                            logger.error("new平台[{}]充电中状态推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationEquipChargeStatus(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]设备充电状态推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("平台类型:{}, 平台名称:{}, 站点id:{}, 订单编号:{}, 设备充电中状态变化推送error:{}",
                        secretInfoVO.getPlatformType(), secretInfoVO.getPlatformName(), stationId, orderCode, e.getMessage());
            }
        }
    }

    /**
     * 充电订单信息推送
     */
    public void notificationChargeOrderInfo(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String orderCode = dto.getOrderCode();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId) || StringUtils.isBlank(orderCode)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            Object service = null;
            try {
                service = getPlatformService(secretInfoVO.getPlatformType());
            } catch (Exception e) {
                logger.error("获取平台服务失败", e);
                continue;
            }

            com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO conversionVO = getConversion(secretInfoVO);

            try {
                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationChargeOrderInfo(orderCode, secretInfoVO);
                    ((ThirdPartyPlatformService) service).notificationChargeOrderInfo(orderCode);
                    ((ThirdPartyPlatformService) service).notificationStopChargeResult(orderCode);
                    ((ThirdPartyPlatformService) service).notificationPayOrderInfo(orderCode);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationChargeOrderInfo(orderCode, conversionVO);
                        } catch (Exception e) {
                            logger.error("new平台[{}]充电订单信息推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationChargeOrderInfo(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]订单信息推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationStopChargeResult(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]停止充电结果推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationPayOrderInfo(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]充电账单推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("充电订单信息推送异常", e);
            }
        }
    }

    /**
     * 开始充电结果推送
     */
    public void commonPushStartChargeResult(NotificationDTO dto) {
        logger.info("开始调用commonPushStartChargeResult接口");
        String stationId = dto.getStationId();
        String orderCode = dto.getOrderCode();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId) || StringUtils.isBlank(orderCode)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());

                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationStartChargeResult(orderCode);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationStartChargeResult(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]开始充电结果推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("充电订单信息推送error", e);
            }
        }
    }

    /**
     * 站点功率信息推送
     */
    public void notificationStationPowerInfo(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String platformType = dto.getPlatformType();

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());
                List<String> stationIds = Lists.newArrayList(stationId);

                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationPowerInfo(stationIds);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationPowerInfo(stationIds);
                        } catch (Exception e) {
                            logger.error("new平台[{}]站点功率信息推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("站点功率信息推送error", e);
            }
        }
    }

    /**
     * 站点费率变化推送
     */
    public void notificationStationFee(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());
                com.jsowell.common.service.thirdparty.vo.ThirdPartySecretInfoVO conversionVO = getConversion(secretInfoVO);

                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationStationFee(stationId, secretInfoVO);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationStationFee(stationId, conversionVO);
                        } catch (Exception e) {
                            logger.error("new平台[{}]站点费率推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("站点费率变化推送error", e);
            }
        }
    }

    /**
     * 历史充电订单信息推送
     */
    public void notificationChargeOrderInfoHistory(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String orderCode = dto.getOrderCode();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(stationId) || StringUtils.isBlank(orderCode)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());

                if (service instanceof ThirdPartyPlatformService) {
                    ((ThirdPartyPlatformService) service).notificationChargeOrderInfo(orderCode);
                    ((ThirdPartyPlatformService) service).notificationChargeOrderInfoHistory(orderCode);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationChargeOrderInfo(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]历史订单信息推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);

                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationChargeOrderInfoHistory(orderCode);
                        } catch (Exception e) {
                            logger.error("new平台[{}]历史充电订单推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("历史充电订单信息推送 error", e);
            }
        }
    }

    /**
     * 充电设备告警信息推送
     */
    public void notificationAlarmInfo(NotificationDTO dto) {
        String stationId = dto.getStationId();
        String pileConnectorCode = dto.getPileConnectorCode();
        String status = dto.getStatus();
        String platformType = dto.getPlatformType();

        if (StringUtils.isBlank(status) || StringUtils.isBlank(pileConnectorCode)) {
            throw new BusinessException(ReturnCodeEnum.CODE_PARAM_NOT_NULL_ERROR);
        }

        List<ThirdPartySecretInfoVO> secretInfoVOS = thirdpartySecretInfoService.queryStationToPlatformList(stationId);
        if (CollectionUtils.isEmpty(secretInfoVOS)) {
            return;
        }

        for (ThirdPartySecretInfoVO secretInfoVO : secretInfoVOS) {
            if (StringUtils.isNotBlank(platformType) && !StringUtils.equals(platformType, secretInfoVO.getPlatformType())) {
                continue;
            }

            try {
                Object service = getPlatformService(secretInfoVO.getPlatformType());
                PushAlarmInfoDTO pushAlarmInfoDTO = new PushAlarmInfoDTO();
                pushAlarmInfoDTO.setPileConnectorCode(pileConnectorCode);
                pushAlarmInfoDTO.setConnectorStatus(status);

                com.jsowell.common.service.thirdparty.dto.PushAlarmInfoDTO dtoConvert =
                        new com.jsowell.common.service.thirdparty.dto.PushAlarmInfoDTO();
                BeanUtils.copyProperties(pushAlarmInfoDTO, dtoConvert);

                if (service instanceof ThirdPartyPlatformService) {

                    ((ThirdPartyPlatformService) service).notificationAlarmInfo(pushAlarmInfoDTO);
                } else if (service instanceof ThirdPartyPlatformApi) {
                    // 后续新平台异步处理
                    ThirdPartyPlatformApi apiService = (ThirdPartyPlatformApi) service;
                    CompletableFuture.runAsync(() -> {
                        try {
                            apiService.notificationAlarmInfo(dtoConvert);
                        } catch (Exception e) {
                            logger.error("new平台[{}]设备告警信息推送失败", secretInfoVO.getPlatformName(), e);
                        }
                    }, executor);
                }
            } catch (Exception e) {
                logger.error("充电设备告警信息推送error", e);
            }
        }
    }
}