完善代码

liurunyu

2025-08-18 9440721414f31d1c85094be1a035d9a90f9c4d5a

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/web/com/CommandCtrl.java

@@ -1,18 +1,37 @@
package com.dy.rtuMw.web.com;

import com.dy.common.softUpgrade.state.UpgradeTaskVo;
import com.dy.common.webUtil.ResultCodeMsg;
import com.dy.rtuMw.resource.ResourceUnit;
import com.dy.rtuMw.server.ServerProperties;
import com.dy.rtuMw.server.forTcp.TcpSessionCache;
import com.dy.rtuMw.server.local.CommandInnerDeaLer;
import com.dy.rtuMw.server.local.ReturnCommand;
import com.dy.rtuMw.server.tasks.RtuDownTask;
import com.dy.rtuMw.server.mqtt.DevStatusDealer;
import com.dy.rtuMw.server.mqtt.MqttManager;
import com.dy.rtuMw.server.msCenter.MsCenterUnit;
import com.dy.rtuMw.server.tasks.WebDownCom4MqttTask;
import com.dy.rtuMw.server.tasks.WebDownComTask;
import com.dy.common.mw.core.CoreUnit;
import com.dy.common.mw.protocol.Command;
import com.dy.common.mw.protocol.CommandType;
import com.dy.common.webUtil.BaseResponse;
import com.dy.common.webUtil.BaseResponseUtils;
import com.dy.rtuMw.server.tasks.WebDownTransparentComTask;
import com.dy.rtuMw.server.upgrade.UpgradeUnit;
import jakarta.servlet.ServletOutputStream;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;

import java.io.*;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;


/**
@@ -32,6 +51,190 @@
        return BaseResponseUtils.buildSuccess("ok");
    }

    /**
     * 下载控制器（RTU）上下行数据的log日志文件
     * @param rtuAddr
     * @param req
     * @param rep
     */
    @GetMapping("/rtuLogFile")
    public void rtuLogFile(String rtuAddr, HttpServletRequest req, HttpServletResponse rep){
        logFile(rtuAddr, req, rep) ;
    }

    /**
     * 下载控制器（RTU）上下行数据的log日志文件
     * @param devId
     * @param req
     * @param rep
     */
    @GetMapping("/mqttDevLogFile")
    public void mqttDevLogFile(String devId, HttpServletRequest req, HttpServletResponse rep){
        logFile(devId, req, rep) ;
    }
    private void logFile(String fileName, HttpServletRequest req, HttpServletResponse rep){
        File logFile = ResourceUnit.getInstance().getLogFile(fileName + ".log") ;
        if(logFile != null && logFile.exists()){
            //在Spring Boot中，application/octet-stream;charset=UTF-8通常表示响应的内容是字节流，
            //并且字符集是UTF-8。对于这种类型的响应，Spring Boot默认使用ByteArrayHttpMessageConverter来处理，
            //因为它可以处理所有application/octet-stream类型的响应。
            //然而，ByteArrayHttpMessageConverter并不直接处理字符集（charset）。
            //字符集通常用于文本内容，而application/octet-stream通常用于二进制内容，因此在这种情况下指定字符集可能是不合适的。
            //不过，如果你确实需要处理带有特定字符集的application/octet-stream响应，你可能需要自定义HttpMessageConverter。
            rep.addHeader("content-type", "application/octet-stream;charset=UTF-8");
            rep.addHeader("Content-Disposition", "attachment;fileName=" + (fileName + ".log")) ;
            ServletOutputStream out = null;
            FileInputStream in = null ;
            try {
                out = rep.getOutputStream() ;
            } catch (Exception ee) {
                out = null ;
            }finally{
                if(out != null){
                    byte[] bs = new byte[1024] ;
                    int len = -1 ;
                    try {
                        in = new FileInputStream(logFile);
                        len = in.read(bs) ;
                        while(len != -1){
                            out.write(bs, 0, len);
                            len = in.read(bs) ;
                        }
                    } catch (Exception eee) {
                    } finally {
                        if(out != null){
                            try{
                                out.flush();
                                out.close();
                            }catch(Exception e){
                            }finally{
                                if(in != null){
                                    try{
                                        in.close();
                                    }catch(Exception e){
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
    /**
     * 下载控制器（RTU）上下行数据的log日志文件
     * @param rtuAddr
     */
    @GetMapping("/rtuLogText")
    public BaseResponse<List<String>> rtuLogText(String rtuAddr){
        return logText(rtuAddr, true) ;
    }
    /**
     * 下载控制器（RTU）上下行数据的log日志文件
     * @param devId
     */
    @GetMapping("/mqttDevLogText")
    public BaseResponse<List<String>> mqttDevLogText(String devId){
        return logText(devId, false) ;
    }
    /**
     * 下载控制器（RTU）上下行数据的log日志文件
     * @param fileName
     * @param reverseOrder
     */
    private BaseResponse<List<String>> logText(String fileName, boolean reverseOrder){
        List<String> list ;
        File logFile = ResourceUnit.getInstance().getLogFile(fileName + ".log") ;
        if(logFile != null && logFile.exists()){
            BufferedReader reader = null ;
            try {
                reader = new BufferedReader(new FileReader(logFile)) ;
                Stream<String> linesStream = reader.lines() ;
                if(reverseOrder){
                    list = linesStream.sorted(Comparator.reverseOrder()).collect(Collectors.toList()) ;//倒序
                }else{
                    list = linesStream.toList() ; //按原来顺序
                }
                return BaseResponseUtils.buildSuccess(list);
            } catch (Exception e) {
                list = new ArrayList() ;
                list.add("读取控制器（" + fileName + "）的日志文件异常：" + (e.getMessage() == null?"":("，" + e.getMessage()))) ;
                return BaseResponseUtils.buildSuccess(list);
            }finally{
                if(reader != null){
                    try{
                        reader.close();
                    }catch(Exception e){
                    }
                }
            }
        }else{
            list = new ArrayList() ;
            list.add("未得到控制器（" + fileName + "）的日志文件") ;
            return BaseResponseUtils.buildSuccess(list);
        }
    }


    /**
     * 注册消息接收者
     * @param msReceiverWebUrl 消息接收者web http post url
     * @return
     */
    @GetMapping("/registerMsReceiverWebUrl")
    public BaseResponse<Boolean> registerMsReceiverWebUrl(String msReceiverWebUrl){
        try{
            MsCenterUnit.getInstance().registerMsReceiver(msReceiverWebUrl);
        }catch (Exception e){
            log.error("注册消息接收者时发生异常", e);
            return BaseResponseUtils.buildError("注册消息接收者时发生异常" + (e.getMessage() == null?"":("：" + e.getMessage())));
        }
        return BaseResponseUtils.buildSuccess(true);
    }

    /**
     * 接收web系统发来升级任务
     * @param vo
     * @return
     */
    @PostMapping(path = "upgradeRtu", consumes = MediaType.APPLICATION_JSON_VALUE)
    public BaseResponse<Boolean> upgradeRtu(@RequestBody UpgradeTaskVo vo) {
        log.info("收到RTU升级任务：\n" + vo.toString()) ;
        try{
            UpgradeUnit.getInstance().setUpgradeTask(vo);
        }catch (Exception e){
            log.error("设置RTU升级任务时发生异常", e);
            return BaseResponseUtils.buildError("设置RTU升级任务时发生异常" + (e.getMessage() == null?"":("：" + e.getMessage())));
        }
        return BaseResponseUtils.buildSuccess(true);
    }

    /**
     * 接收web系统发来强制结束升级任务
     * @return
     */
    @GetMapping(path = "ugForceOver")
    public BaseResponse<String> ugForceOver() {
        log.info("收到停止RTU升级任务命令") ;
        try{
            String mes = UpgradeUnit.getInstance().forceOverUpgradeTask();
            if(mes == null){
                mes = "停止升级任务成功" ;
                return BaseResponseUtils.buildResult(ResultCodeMsg.RsCode.SUCCESS_CODE, mes, mes);
            }else{
                return BaseResponseUtils.buildResult(ResultCodeMsg.RsCode.FAIL_CODE, mes, mes);
            }
        }catch (Exception e){
            log.error("停止RTU升级任务时发生异常", e);
            return BaseResponseUtils.buildError("停止RTU升级任务时发生异常" + (e.getMessage() == null?"":("：" + e.getMessage())));
        }
    }

    /**
     * 接收web系统发来的命令
     * @param com
     * @return
     */
    @PostMapping(path = "send", consumes = MediaType.APPLICATION_JSON_VALUE)
    public BaseResponse<Command> send(@RequestBody Command com) {
        log.info("收到web系统发来的命令：\n" + com.toString()) ;
@@ -54,28 +257,28 @@
            //通信中间件内部命令，例如查询监控中间件时钟，查询RTU在线情况等
            try{
                Command reCom = new CommandInnerDeaLer().deal(com) ;
                /*
                boolean error = false ;
                if(reCom.param != null && reCom.param != null){
                    CommandBackParam cbp = (CommandBackParam)reCom.param ;
                    if(cbp.getSuccess() != null && !cbp.getSuccess().booleanValue()){
                        error = true ;
                        return BaseResponseUtils.buildError(ReturnCommand.errored(cbp.getMessage(), com.getId(), null));
                    }
                }
                if(!error){
                    return BaseResponseUtils.buildError(ReturnCommand.errored(cbp.getMessage(), com.getId(), null));
                }
                */
                return BaseResponseUtils.buildSuccess(reCom);
            }catch(Exception e){
                return BaseResponseUtils.buildError(ReturnCommand.errored("处理内部命令出错" + (e.getMessage() == null?"":("，" + e.getMessage())), com.getId(), com.getCode()) );
            }

        }else if(commandType.equals(CommandType.outerCommand)){
            //发向RTU的外部命令，异步处理，web端jroups成员同步得到命令处理结果，但构造命令及下发命令和命令结果接收要异步得到
            //发向RTU的外部命令
            try{
                return this.dealOuterCommand(com) ;
            }catch(Exception e){
                return BaseResponseUtils.buildError(ReturnCommand.errored("处理发向RTU的外部命令出错" + (e.getMessage() == null?"":("，" + e.getMessage())), com.getId(), com.getCode()) );
            }
        }else if(commandType.equals(CommandType.outerTransCommand)){
            //发向RTU的外部透传命令
            try{
                return this.dealOutTransparentCommand(com) ;
            }catch(Exception e){
                return BaseResponseUtils.buildError(ReturnCommand.errored("处理发向RTU的外部透传命令出错" + (e.getMessage() == null?"":("，" + e.getMessage())), com.getId(), com.getCode()) );
            }
        }else if(commandType.equals(CommandType.mqttCommand)){
            //发向MQTT的外部命令
            try{
                return this.dealMqttCommand(com) ;
            }catch(Exception e){
                return BaseResponseUtils.buildError(ReturnCommand.errored("处理发向RTU的外部命令出错" + (e.getMessage() == null?"":("，" + e.getMessage())), com.getId(), com.getCode()) );
            }
@@ -85,6 +288,20 @@
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，收到命令的命令功能码为空！", com.getId(), com.getCode()));
        }
    }

    /**
     * 本地调用，例如任务树中某个任务下发的命令（如清空命令）
     * @param com
     * @return
     */
    public BaseResponse<Command> sendOutComFromLocal(Command com) {
        try{
            return this.dealOuterCommand(com) ;
        }catch(Exception e){
            return BaseResponseUtils.buildError(ReturnCommand.errored("处理发向RTU的外部命令出错" + (e.getMessage() == null?"":("，" + e.getMessage())), com.getId(), com.getCode()) );
        }
    }


    /**
     * 处理发向RTU的外部命令
@@ -106,7 +323,7 @@
        }

        //生成异步任务
        RtuDownTask task = new RtuDownTask() ;
        WebDownComTask task = new WebDownComTask() ;
        task.data = command ;
        try{
            log.info("构造下发远程命令" + command.getCode() + "的核心任务，并放入任务队列中");
@@ -119,4 +336,104 @@
        return BaseResponseUtils.buildSuccess(ReturnCommand.successed("命令已接受，即将构造并下发命令。", command.getId(), command.getCode()));
    }



    /**
     * 处理发向RTU的外部命令
     * @return 结果
     */
    private BaseResponse<Command> dealOutTransparentCommand(Command command){
        String rtuAddr = command.getRtuAddr() ;
        if(rtuAddr == null || rtuAddr.trim().equals("")){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，RTU地址为空！", command.getId(), command.getCode())) ;
        }
        rtuAddr = rtuAddr.trim() ;
        Boolean onLine = TcpSessionCache.isConnect(rtuAddr);
        if(onLine == null){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，RTU（地址=" + rtuAddr + "）未上线！", command.getId(), command.getCode())) ;
        }else if(!onLine.booleanValue()){
            if(!ServerProperties.isLowPower){
                return BaseResponseUtils.buildError(ReturnCommand.errored("出错，RTU（地址=" + rtuAddr + "）离线！", command.getId(), command.getCode())) ;
            }
        }

        //生成异步任务
        WebDownTransparentComTask task = new WebDownTransparentComTask() ;
        task.data = command ;
        try{
            log.info("构造下发远程透传命令" + command.getCode() + "的核心任务，并放入任务队列中");
            CoreUnit.getInstance().pushCoreTask(task);
        }catch(Exception e){
            log.error(e.getMessage(), e);
            return BaseResponseUtils.buildError(ReturnCommand.errored("透传命令处理失败" + e.getMessage(), command.getId(), command.getCode())) ;
        }

        return BaseResponseUtils.buildSuccess(ReturnCommand.successed("透传命令已接受，即将构造并下发命令。", command.getId(), command.getCode()));
    }

    /**
     * 处理发向MQTT的外部命令
     * @return 结果
     */
    private BaseResponse<Command> dealMqttCommand(Command command){
        String mqttDevId = command.getRtuAddr() ;//FBox设备号
        if(mqttDevId == null || mqttDevId.trim().equals("")){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，FBox设备ID为空！", command.getId(), command.getCode())) ;
        }
        if(!ServerProperties.mqttUnitEnable.booleanValue()){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，MQTT连接模块配置未启动！", command.getId(), command.getCode())) ;
        }
        if(MqttManager.getInstance().poolIsClose()){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，MQTT连接池未创建成功！", command.getId(), command.getCode())) ;
        }
        if(!DevStatusDealer.oneOnLine(mqttDevId)){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，FBox设备未在线！", command.getId(), command.getCode())) ;
        }

        //生成异步任务
        WebDownCom4MqttTask task = new WebDownCom4MqttTask() ;
        task.data = command ;
        try{
            log.info("构造下发MQTT命令" + command.getCode() + "的核心任务，并放入任务队列中");
            CoreUnit.getInstance().pushCoreTask(task);
        }catch(Exception e){
            log.error(e.getMessage(), e);
            return BaseResponseUtils.buildError(ReturnCommand.successed("MQTT命令处理失败" + e.getMessage(), command.getId(), command.getCode())) ;
        }
        return BaseResponseUtils.buildSuccess(ReturnCommand.successed("MQTT命令已接受，即将构造并下发命令。", command.getId(), command.getCode()));
    }


    /**
     * 处理发向MQTT的内部部命令
     * @return 结果
     */
    private BaseResponse<Command> dealMqttInnerCommand(Command command){
        String mqttDevId = command.getRtuAddr() ;//FBox设备号
        if(mqttDevId == null || mqttDevId.trim().equals("")){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，FBox设备ID为空！", command.getId(), command.getCode())) ;
        }
        if(!ServerProperties.mqttUnitEnable.booleanValue()){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，MQTT连接模块配置未启动！", command.getId(), command.getCode())) ;
        }
        if(MqttManager.getInstance().poolIsClose()){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，MQTT连接池未创建成功！", command.getId(), command.getCode())) ;
        }
        if(!DevStatusDealer.oneOnLine(mqttDevId)){
            return BaseResponseUtils.buildError(ReturnCommand.errored("出错，FBox设备未在线！", command.getId(), command.getCode())) ;
        }

        //生成异步任务
        WebDownCom4MqttTask task = new WebDownCom4MqttTask() ;
        task.data = command ;
        try{
            log.info("构造下发MQTT命令" + command.getCode() + "的核心任务，并放入任务队列中");
            CoreUnit.getInstance().pushCoreTask(task);
        }catch(Exception e){
            log.error(e.getMessage(), e);
            return BaseResponseUtils.buildError(ReturnCommand.successed("MQTT命令处理失败" + e.getMessage(), command.getId(), command.getCode())) ;
        }
        return BaseResponseUtils.buildSuccess(ReturnCommand.successed("MQTT命令已接受，即将构造并下发命令。", command.getId(), command.getCode()));
    }

}