1、通信中间件重构MQTT相关下行命令的逻辑； 2、remote模块完善MQTT相关下行命令解除报警、搅拌启停、注肥启停、灌溉启停功能类；

liurunyu

昨天 0c870ae5b2af348e19b10cc1a99f103f95c5a1cc

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/mqtt/MqttPubMsgNode.java

@@ -20,12 +20,16 @@
    public MqttPubMsg result ;//下行命令
    public Long cachTime ;//缓存时刻
    public boolean onceReceivedResult ;//已经收到命令应答
    public long lastSendTime = 0 ;//上次发送时间
    public int sendTimes = 0 ;//发送次数


    public MqttPubMsgNode(MqttPubMsg result){
        this.result = result ;
        this.cachTime = System.currentTimeMillis() ;
        this.onceReceivedResult = false ;
        this.lastSendTime = 0L ;
        this.sendTimes = 0 ;
    }

    /**
@@ -36,9 +40,13 @@
    public boolean dealSelf(Long now){
        if(this.onceReceivedResult){
            //已经收到命令结果
            //记录状态
            //RtuStatusDealer.commandSuccess(this.result.rtuAddr, this.result.downCode, this.result.downCodeName);
            return true ;
        }
        if(this.sendTimes >= (1 + MqttUnit.confVo.reSendTimesByNoResult)){
            return this.decideRemoveNodeFromCach(now, null) ;
        }
        if(this.lastSendTime != 0 && now - this.lastSendTime >= MqttUnit.confVo.sendInterval){
            return this.decideRemoveNodeFromCach(now, null) ;
        }
        boolean noConnect2MqSv = false ;
        MqttManager mqttManager = MqttManager.getInstance() ;
@@ -47,7 +55,7 @@
        noConnect2MqSv = mqttManager.poolIsClose() ;
        if(noConnect2MqSv){
            //未曾连接MQTT服务器
            return this.decideRemoveNodeFromCach(now) ;
            return this.decideRemoveNodeFromCach(now, null) ;
        }else{
            try {
                //如果网络不好或断网，此处用时较长
@@ -61,16 +69,18 @@
        }
        if(noConnect2MqSv){
            //未曾连接MQTT服务器
            return this.decideRemoveNodeFromCach(now) ;
            return this.decideRemoveNodeFromCach(now, null) ;
        }else{
            if(mqttClient != null && mqttClient.isConnected()){
                try {
                    mqttManager.publishMsg(mqttClient, this.result.topic.longName(), this.result.msg);
                    this.sendTimes ++ ;
                    this.lastSendTime = System.currentTimeMillis() ;
                    DevStatusDealer.afterSendPubMessage(this.result.deviceId);
                    RtuLogDealer.log4Mqtt(this.result.deviceId, "发布消息    主题：" + this.result.topic + "   消息：" + this.result.msg);
                    log.info("发布MQTT消息（主题=" + this.result.topic + "）" + this.result.msg);
                    RtuLogDealer.log4Mqtt(this.result.deviceId, "发布消息    主题：" + this.result.topic.longName() + "   消息：" + this.result.msg);
                    log.info("发布MQTT消息（主题=" + this.result.topic.longName() + "）" + this.result.msg);
                }catch (Exception e){
                    log.error("MQTT发布消息失败（主题=" + this.result.topic + "）" , e);
                    log.error("MQTT发布消息失败（主题=" + this.result.topic.longName() + "）" , e);
                }finally {
                    mqttManager.pushMqttClient(mqttClient);
                }
@@ -81,18 +91,18 @@
                }
            }else{
                //未曾连接MQTT服务器
                return this.decideRemoveNodeFromCach(now) ;
                return this.decideRemoveNodeFromCach(now, false) ;
            }
        }
    }

    private boolean decideRemoveNodeFromCach(Long now){
        if(!this.result.isCacheForOffLine){
    private boolean decideRemoveNodeFromCach(Long now, Boolean isOffLine){
        if(isOffLine != null && isOffLine.booleanValue() && !this.result.isCacheForOffLine){
            //不在线命令不缓存
            return true ;
        }else{
            //不在线命令缓存
            if(now - this.cachTime >= ServerProperties.offLineCacheTimeout){
            if(now - this.cachTime >= MqttUnit.confVo.comCacheTimeout){
                //缓存超时
                return true ;
            }