1、实现强制停止RTU远程升级任务功能； 2、业务系统与通信中间件web通信模块优化，使变得更通用与简单。

liurunyu

2024-11-29 2b4fc2b16efdc887b71855b0b4d5b5680453d088

1、实现强制停止RTU远程升级任务功能；
2、业务系统与通信中间件web通信模块优化，使变得更通用与简单。

 pipIrr-platform/pipIrr-global/src/main/java/com/dy/pipIrrGlobal/rtuMw/Web2RtuMw.java

File was renamed from pipIrr-platform/pipIrr-global/src/main/java/com/dy/pipIrrGlobal/rtuMw/ToRtuMwCom.java
@@ -19,33 +19,71 @@
 * @Date: 2024/10/23 11:45
 * @Description
 */
public abstract class ToRtuMwCom {
public abstract class Web2RtuMw {
    /**
     * pro_mw：属性
     * tag从控制器中获取
     * key_mw：url的key
     */
    private static final String pro_mw = "mw";
    private static final String key_mw = "comSendUrl";
    private static final String keyUg_mw = "ugTaskSendUrl";
    private static final String pro_url = "url";

    protected static final String ContextComSend = "/rtuMw/com/send";

    protected static final String ContextRtuLogFile = "/rtuMw/com/rtuLogFile";
    protected static final String ContextRtuLogText = "/rtuMw/com/rtuLogText";

    protected static final String ContextUgTaskSend = "/rtuMw/com/upgradeRtu";
    protected static final String ContextUgForceStop = "/rtuMw/com/ugForceStop";
    /**
     * 得到向通信中间件发送数据的URL
     * @param env
     * @return
     */
    protected String getToMwUrl(Environment env) {
        return env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + key_mw);
    protected String get2MwUrl(Environment env) {
        return env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + pro_url);
    }
    /**
     * 得到向通信中间件发送强制停止升级的命令URL
     * @param env
     * @return
     */
    protected String get2MwRequestUrl(Environment env, String context) {
        return get2MwUrl(env) + context;
    }


    /**
     * 得到向通信中间件发送数据的URL
     * @param env
     * 向通信中间件发送rtu远程升级任务
     * @param restTemplate SpringBoot的RestTemplate
     * @param toMwUrl 到通信中间件的web请求Url
     * @param param 请求参数
     * @return
     */
    protected String getToMwUgUrl(Environment env) {
        return env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + keyUg_mw);
    protected BaseResponse sendRequest2Mw(RestTemplate restTemplate, String toMwUrl, Object param) {
        String url = UriComponentsBuilder.fromUriString(toMwUrl)
                .build()
                .toUriString();
        HttpHeaders headers = new HttpHeaders();
        HttpEntity<?> httpEntity ;
        if(param != null){
            httpEntity = new HttpEntity<>(param, headers);
        }else{
            httpEntity = new HttpEntity<>(headers);
        }
        ResponseEntity<BaseResponse> response = null;
        try {
            // 通过Post方式调用接口
            response = restTemplate.exchange(url, HttpMethod.POST, httpEntity, BaseResponse.class);
        } catch (Exception e) {
            e.printStackTrace();
            return BaseResponseUtils.buildError("后端系统出错，中间件调用异常");
        }
        if(response == null){
            return BaseResponseUtils.buildError("后端系统出错，中间件调用异常");
        }else{
            return response.getBody();
        }
    }

    /**
@@ -74,58 +112,4 @@
        return com ;
    }

    /**
     * 发送命令
     *
     * @return
     */
    protected BaseResponse sendCom2Mw(RestTemplate restTemplate, String comSendUrl, Command com) {
        String url = UriComponentsBuilder.fromUriString(comSendUrl)
                .build()
                .toUriString();
        HttpHeaders headers = new HttpHeaders();
        HttpEntity<Command> httpEntity = new HttpEntity<>(com, headers);
        ResponseEntity<BaseResponse> response = null;
        try {
            // 通过Post方式调用接口
            response = restTemplate.exchange(url, HttpMethod.POST, httpEntity, BaseResponse.class);
        } catch (Exception e) {
            e.printStackTrace();
            return BaseResponseUtils.buildError("后端系统出错，中间件调用异常");
        }
        if(response == null){
            return BaseResponseUtils.buildError("后端系统出错，中间件调用异常");
        }else{
            return response.getBody();
        }
    }


    /**
     * 向通信中间件发送rtu远程升级任务
     * @param restTemplate
     * @param comSendUrl
     * @param vo
     * @return
     */
    protected BaseResponse sendUpgradeTask2Mw(RestTemplate restTemplate, String comSendUrl, UpgradeTaskVo vo) {
        String url = UriComponentsBuilder.fromUriString(comSendUrl)
                .build()
                .toUriString();
        HttpHeaders headers = new HttpHeaders();
        HttpEntity<UpgradeTaskVo> httpEntity = new HttpEntity<>(vo, headers);
        ResponseEntity<BaseResponse> response = null;
        try {
            // 通过Post方式调用接口
            response = restTemplate.exchange(url, HttpMethod.POST, httpEntity, BaseResponse.class);
        } catch (Exception e) {
            e.printStackTrace();
            return BaseResponseUtils.buildError("后端系统出错，中间件调用异常");
        }
        if(response == null){
            return BaseResponseUtils.buildError("后端系统出错，中间件调用异常");
        }else{
            return response.getBody();
        }
    }
}

 pipIrr-platform/pipIrr-global/src/main/resources/application-global.yml

@@ -312,20 +312,14 @@
mw:
    #命令发送地址(中缀是机构tag，其也是数据源后缀名称)
    ym:
        url: "http://127.0.0.1:8070"
        comSendUrl: "http://127.0.0.1:8070/rtuMw/com/send"
        ugTaskSendUrl: "http://127.0.0.1:8070/rtuMw/com/upgradeRtu"
        rtuLogFileUrl: "http://127.0.0.1:8070/rtuMw/com/rtuLogFile"
        rtuLogTextUrl: "http://127.0.0.1:8070/rtuMw/com/rtuLogText"
    sp:
        url: "http://127.0.0.1:8073"
        comSendUrl: "http://127.0.0.1:8073/rtuMw/com/send"
        ugTaskSendUrl: "http://127.0.0.1:8073/rtuMw/com/upgradeRtu"
        rtuLogFileUrl: "http://127.0.0.1:8073/rtuMw/com/rtuLogFile"
        rtuLogTextUrl: "http://127.0.0.1:8073/rtuMw/com/rtuLogText"
    test:
        url: "http://127.0.0.1:8072"
        comSendUrl: "http://127.0.0.1:8072/rtuMw/com/send"
        ugTaskSendUrl: "http://127.0.0.1:8072/rtuMw/com/upgradeRtu"
        rtuLogFileUrl: "http://127.0.0.1:8072/rtuMw/com/rtuLogFile"
        rtuLogTextUrl: "http://127.0.0.1:8072/rtuMw/com/rtuLogText"
    #监测控制模块回调地址
    rtuCallbackUrl_rm: "http://127.0.0.1:8081/remote/comRes/receive"
    #rtu远程升级模块回调地址

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/upgrade/UpgradeManager.java

@@ -88,12 +88,20 @@
     * 因为强制结束升级任务，对一个未升级完成的RTU就会卡死，
     * 所以当强制结束升级任务，代码逻辑并没有强制结果RTU升级过程，如果升级过程也强制停止，那么RTU真会卡死
     */
    public void forceOverUpgradeTask() {
    public String forceOverUpgradeTask() {
        if(this.task != null){
            this.task.countRunningRtuCount();
            if(this.task.curUgRunningRtuTotal > 0){
                return "当前存在升级中的设备，不能停止升级任务" ;
            }else{
            this.stop();
            this.task.forceOver();
        }
        this.task = null ;
                return null ;
            }
        }else{
            return "当前没有升级任务" ;
        }
    }

    /**

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/upgrade/UpgradeTask.java

@@ -52,7 +52,7 @@
    ///////////////////////////////////////////////////
    //以下内部控制用
    @JSONField(serialize = false)
    private int curUgRunningRtuTotal = 0 ;//当前正在升级的RTU个数
    protected int curUgRunningRtuTotal = 0 ;//当前正在升级的RTU个数

    public UpgradeTask() {
        this.curUgRunningRtuTotal = 0 ;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/upgrade/UpgradeUnit.java

@@ -75,10 +75,13 @@
    /**
     * 停止当前升级任务
     * @throws Exception
     * @return null：停止任务成功，否则返回失败原因
     */
    public void overUpgradeTask() throws Exception {
    public String overUpgradeTask() throws Exception {
        if(manager != null ){
            manager.forceOverUpgradeTask() ;
            return manager.forceOverUpgradeTask() ;
        }else{
            return "通信中间件内部错误" ;
        }
    }


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

@@ -1,6 +1,7 @@
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;
@@ -161,6 +162,26 @@
    }

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


    /**
     * 接收web系统发来的命令
     * @param com
     * @return

 pipIrr-platform/pipIrr-mw/pipIrr-mw-simulate-rtu/src/main/java/com/dy/simRtu/tcpClient/downData/DownData.java

@@ -53,7 +53,7 @@
                        if(ServerProperties.argMultiDie){
                            if(count9602 > 10){
                                int random =  CreateRandom.create_between(10, 100) ;
                                if(random > 98){
                                if(random > 90){
                                    UpHeartBeat.upHeartBeat = true ;//使能上行心跳
                                }else{
                                    UpCd9602.upData() ;

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/monitor/MonitorSv.java

@@ -9,7 +9,7 @@
import com.dy.common.webUtil.QueryResultVo;
import com.dy.pipIrrGlobal.daoPr.PrIntakeMapper;
import com.dy.pipIrrGlobal.rtuMw.CodeLocal;
import com.dy.pipIrrGlobal.rtuMw.ToRtuMwCom;
import com.dy.pipIrrGlobal.rtuMw.Web2RtuMw;
import com.dy.pipIrrGlobal.voPr.VoOnLineIntake;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.utils.PojoUtils;
@@ -29,7 +29,7 @@
 */
@Slf4j
@Service
public class MonitorSv extends ToRtuMwCom {
public class MonitorSv extends Web2RtuMw {

    @Autowired
    private PrIntakeMapper prIntakeMapper;
@@ -76,8 +76,8 @@
            //向通信中间件发关命令，查询部分RTU在线情况
            Command com = this.createInnerCommand(CodeLocal.onLinePart);
            com.setParam(rtuAddrs) ;
            String comSendUrl = this.getToMwUrl(this.env) ;
            BaseResponse res = sendCom2Mw(restTemplate, comSendUrl, com) ;
            String rqUrl = this.get2MwRequestUrl(this.env, ContextComSend) ;
            BaseResponse res = sendRequest2Mw(restTemplate, rqUrl, com) ;
            if(res != null){
                if(res.isSuccess()){
                    Command reCom = JSON.parseObject(res.getContent() == null ? null : JSON.toJSONString(res.getContent()), Command.class) ;
@@ -114,8 +114,8 @@
    private QueryResultVo<List<VoOnLineIntake>> selectIntakesOnOrOffLine(QueryVo vo) {
        //向通信中间件发关命令，查询部分RTU在线情况
        Command com = this.createInnerCommand(CodeLocal.onLineAll);
        String comSendUrl = this.getToMwUrl(this.env) ;
        BaseResponse res = sendCom2Mw(restTemplate, comSendUrl, com) ;
        String rqUrl = this.get2MwRequestUrl(this.env, ContextComSend) ;
        BaseResponse res = sendRequest2Mw(restTemplate, rqUrl, com) ;
        if(res != null){
            if(res.isSuccess()){
                Command reCom = JSON.parseObject(res.getContent() == null ? null : JSON.toJSONString(res.getContent()), Command.class) ;

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/rtu/RtuLogCtrl.java

@@ -1,9 +1,9 @@
package com.dy.pipIrrRemote.rtu;

import com.dy.common.aop.SsoAop;
import com.dy.common.multiDataSource.DataSourceContext;
import com.dy.common.webUtil.BaseResponse;
import com.dy.common.webUtil.BaseResponseUtils;
import com.dy.pipIrrGlobal.rtuMw.Web2RtuMw;
import jakarta.servlet.ServletOutputStream;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
@@ -32,11 +32,7 @@
@Slf4j
@RestController
@RequestMapping(path="rtuLog")
public class RtuLogCtrl {

    private static final String pro_mw = "mw";
    private static final String key_mw_file = "rtuLogFileUrl";
    private static final String key_mw_text = "rtuLogTextUrl";
public class RtuLogCtrl extends Web2RtuMw {

    private Environment env ;
    private RestTemplate restTemplate;
@@ -55,10 +51,10 @@
    @GetMapping(path = "file")
    @SsoAop()
    public BaseResponse<List<String>> rtuLogFile(String rtuAddr, HttpServletRequest req, HttpServletResponse rep){
        String mwUrlRtuLogFile = env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + key_mw_file);
        String rqUrl = this.get2MwRequestUrl(this.env, ContextRtuLogFile) ;
        ServletOutputStream out = null ;
        try{
            byte[] bs = this.requestMw4File(rtuAddr, mwUrlRtuLogFile) ;
            byte[] bs = this.requestMw4File(rtuAddr, rqUrl) ;
            if(bs != null && bs.length > 0){
                String fileReName = rtuAddr + ".log" ;
                //URLEncoder.encode可以防止中文乱码
@@ -91,8 +87,9 @@
    @GetMapping(path="text")
    @SsoAop()
    public BaseResponse<List<String>> rtuLogText(String rtuAddr){
        String mwUrlRtuLogText = env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + key_mw_text);
        BaseResponse<List<String>> text = this.requestMw4Text(rtuAddr, mwUrlRtuLogText);
        String rqUrl = this.get2MwRequestUrl(this.env, ContextRtuLogText) ;
        //String mwUrlRtuLogText = env.getProperty(pro_mw + "." + DataSourceContext.get() + "." + key_mw_text);
        BaseResponse<List<String>> text = this.requestMw4Text(rtuAddr, rqUrl);
        if (text != null){
            if (text.getContent().get(0).contains("控制器")){
                return BaseResponseUtils.buildErrorMsg(""+text.getContent().get(0)+"") ;
@@ -101,7 +98,6 @@
        }else {
            return BaseResponseUtils.buildErrorMsg("获取日志文件为null") ;
        }
//        return this.requestMw4Text(rtuAddr, mwUrlRtuLogText) ;
    }


@@ -141,5 +137,4 @@

        return (response==null?null:response.getBody());
    }

}

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/rtuUpgrage/RtuUpgradeCtrl.java

@@ -3,7 +3,6 @@
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.support.ExcelTypeEnum;
import com.dy.common.aop.SsoAop;
import com.dy.common.softUpgrade.state.UpgradeRtu;
import com.dy.common.softUpgrade.state.UpgradeTaskVo;
import com.dy.common.springUtil.SpringContextUtil;
import com.dy.common.webUtil.BaseResponse;
@@ -12,8 +11,7 @@
import com.dy.common.webUtil.ResultCodeMsg;
import com.dy.pipIrrGlobal.pojoUg.UgRtuProgram;
import com.dy.pipIrrGlobal.pojoUg.UgRtuTask;
import com.dy.pipIrrGlobal.rtuMw.ToRtuMwCom;
import com.dy.pipIrrGlobal.voPr.VoDivide;
import com.dy.pipIrrGlobal.rtuMw.Web2RtuMw;
import com.dy.pipIrrGlobal.voUg.VoUgRtuResult;
import com.dy.pipIrrGlobal.voUg.VoWatch;
import io.swagger.v3.oas.annotations.Operation;
@@ -35,7 +33,6 @@

import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;

@@ -48,7 +45,7 @@
@Tag(name = "rtu远程升级任务", description = "rtu远程升级任务相关操作")
@RestController
@RequestMapping(path = "rtuUpgrade")
public class RtuUpgradeCtrl extends ToRtuMwCom {
public class RtuUpgradeCtrl extends Web2RtuMw {
    @Autowired
    private RtuUpgradeSv sv ;

@@ -108,8 +105,8 @@
        vo.rtuAddrList = taskRtuAddrs ;
        vo.callbackWebUrl = ugCallbackUrl_rm ;

        String ugSendUrl = this.getToMwUgUrl(this.env) ;
        BaseResponse res = sendUpgradeTask2Mw(restTemplate, ugSendUrl, vo) ;
        String rqUrl = this.get2MwRequestUrl(this.env, ContextUgTaskSend) ;
        BaseResponse res = sendRequest2Mw(restTemplate, rqUrl, vo) ;
        if(res != null){
            if(res.isSuccess()){
                this.sv.setUpgradeTaskExecuted(id);
@@ -127,6 +124,40 @@
            return BaseResponseUtils.buildErrorMsg("通信中间件返回结果为null") ;
        }
    }
    /**
     * 下发强制停止当前rtu远程升级任务
     * @return 操作结果
     */
    @Operation(summary = "下发强制停止当前rtu远程升级任务", description = "下发强制停止当前rtu远程升级任务")
    @ApiResponses(value = {
            @ApiResponse(
                    responseCode = ResultCodeMsg.RsCode.SUCCESS_CODE,
                    description = "返回操作成功与否数据（BaseResponse.content:Boolean）",
                    content = {@Content(mediaType = MediaType.APPLICATION_JSON_VALUE,
                            schema = @Schema(implementation = String.class))}
            )
    })
    @GetMapping(path = "/forceStop")
    @SsoAop()
    public BaseResponse<String> forceStop() {
        String rqUrl = this.get2MwRequestUrl(this.env, ContextUgForceStop) ;
        BaseResponse res = sendRequest2Mw(restTemplate, rqUrl, null) ;
        if(res != null){
            if(res.isSuccess()){
                return BaseResponseUtils.buildSuccess(true) ;
            }else{
                String msg = res.getContent()==null?null:(String)res.getContent() ;
                if(msg == null){
                    msg = res.getMsg() ;
                }
                log.error("通信中间件执行下发强制停止当前rtu远程升级任务失败：" + msg) ;
                return BaseResponseUtils.buildErrorMsg("通信中间件执行强制停止当前rtu远程升级任务失败：" + msg) ;
            }
        }else{
            log.error("通信中间件返回结果为null") ;
            return BaseResponseUtils.buildErrorMsg("通信中间件返回结果为null") ;
        }
    }

    private void valueFromPo(UpgradeTaskVo vo, UgRtuTask tpo, UgRtuProgram ppo){
        vo.id = "" + tpo.id ;