Merge remote-tracking branch 'origin/dev' into dev

jiaosong

2023-11-03 8bf400b67418a4a81185a4a302fdbdfce2cdf018

 server/services/src/main/java/com/doumee/service/business/impl/DeviceSubscribeServiceImpl.java

@@ -6,11 +6,10 @@
import com.doumee.core.exception.BusinessException;
import com.doumee.core.mqtt.config.MqttConfig;
import com.doumee.core.mqtt.service.MqttToolService;
import com.doumee.dao.business.MemberRidesMapper;
import com.doumee.dao.business.MqttLogMapper;
import com.doumee.dao.business.model.Bikes;
import com.doumee.dao.business.model.Locks;
import com.doumee.dao.business.model.MemberRides;
import com.doumee.dao.business.model.MqttLog;
import com.doumee.dao.business.SitesMapper;
import com.doumee.dao.business.model.*;
import com.doumee.service.business.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
@@ -32,6 +31,9 @@
    @Autowired
    @Lazy
    MemberRidesService memberRidesService;
    @Lazy
    @Autowired
    SitesMapper sitesMapper;
    @Autowired
    private MqttLogMapper mqttLogMapper;
    @Autowired
@@ -41,24 +43,21 @@
       log.info("mqtt消息订阅==================="+param);
        String info = Constants.MqttTopic.sub_lockInfo.substring(Constants.MqttTopic.sub_lockInfo.lastIndexOf("/")+1) ;
        String closeLock = Constants.MqttTopic.sub_closeLock.substring(Constants.MqttTopic.sub_closeLock.lastIndexOf("/")+1) ;
        String health = Constants.MqttTopic.sub_health.substring(Constants.MqttTopic.sub_health.lastIndexOf("/")+1) ;
        String brokers = "brokers";
        if(!StringUtils.contains(topic, brokers)){
        if(topic.indexOf(Constants.MqttTopic.topic_index)!=0
                ||topic.split("/").length < 5
                    ||topic.split("/").length < 4
                || (!StringUtils.contains(topic, info)
                    &&!StringUtils.contains(topic, health)
                   &&!StringUtils.contains(topic,closeLock))){
            log.error("mqtt消息订阅===========无效数据========"+param);
            return;
        }

        String[] ss = topic.split("/");
        String siteid =  ss[2];//站点编码
        String lockid =  ss[3];//锁头编码
        if(StringUtils.isBlank(siteid)||StringUtils.isBlank(lockid)){
            //如果锁头编码为空
            log.error("mqtt消息订阅==============无效数据====="+param);
            return;
        }
        MqttLog mqttLog = new MqttLog();
        mqttLog.setMsgId(msgId);
        mqttLog.setTopic(topic);
        int msgCount  =mqttLogMapper.selectCount(new QueryWrapper<MqttLog>().lambda().eq(MqttLog::getMsg, param).eq(MqttLog::getType, Constants.ZERO));
        if(msgCount>0){
            log.error("mqtt消息订阅==============已消费数据====="+param);
@@ -68,8 +67,22 @@
        String logInfo = "";
        int result =0;
        try {
            if(!topic.contains(brokers)){
                String[] ss = topic.split("/");
                String siteid =  ss[2];//站点编码
                if(StringUtils.isBlank(siteid) ){
                    //如果锁头编码为空
                    log.error("mqtt消息订阅==============无效数据====="+topic+param);
                    return;
                }
            if(StringUtils.contains(topic, info)){
                //如果锁头信息上报
                    String lockid =  ss[3];//锁头编码
                    if( StringUtils.isBlank(lockid)){
                        //如果锁头编码为空
                        log.error("mqtt消息订阅==============无效数据====="+topic+param);
                        return;
                    }
                Locks locks  = JSONObject.parseObject(param, Locks.class);
                locks.setSiteId(siteid);
                locks.setCode(lockid);
@@ -77,18 +90,64 @@
                result = memberRidesService.mqttLockInfoEvent(locks);
                logInfo = "mqtt消息订阅锁头信息";
                log.info("mqtt消息订阅=========锁信息==========成功");
            }
            if(StringUtils.contains(topic, closeLock)){
                }else if(StringUtils.contains(topic, closeLock)){
                //如果还车上报
                    String lockid =  ss[3];//锁头编码
                    if( StringUtils.isBlank(lockid)){
                        //如果锁头编码为空
                        log.error("mqtt消息订阅==============无效数据====="+topic+param);
                        return;
                    }
                JSONObject pjson  = JSONObject.parseObject(param);
                MemberRides bikes = new MemberRides();
                bikes.setBikeCode(pjson.getString("bikeCode"));
                bikes.setBackLockId( lockid);
                bikes.setBackSiteId( siteid);
                bikes.setBackCommondId(logId);
                    bikes.setStatus(pjson.getInteger("status"));
                result = memberRidesService.mqttCloseBikeEvent(bikes);
                logInfo = "mqtt消息订阅还车消息";
                log.info("mqtt消息订阅=========还车==========成功");
                }else if(StringUtils.contains(topic, health)){
                    //心跳消息=
                    Sites site = new Sites();
                    site.setId(siteid);
                    site.setOnline(Constants.ZERO);
                    site.setLastLinkDate(new Date());
                    sitesMapper.updateById(site);//更新站点状态
                    logInfo = "mqtt消息订阅心跳消息";
                    log.info("mqtt消息订阅=========心跳消息==========成功");

                }
            }else {
                //如果站点上下线消息
                JSONObject pjson = JSONObject.parseObject(param);
                String clientId = String.valueOf(pjson.get("clientid"));
                if(clientId.contains("doumee")){
                    log.error("mqtt消息订阅==============无效订阅状态====="+topic+param);
                    return;
                }
                Sites site = sitesMapper.selectById(clientId);
                if(site == null){
                    log.error("mqtt消息订阅==============无效订阅状态====="+topic+param);
                    return;
                }
                if (topic.endsWith("disconnected")) {
                    if(Constants.formatIntegerNum(site.getOnline()) == Constants.ONE){
                        log.error("mqtt消息订阅==============已订阅状态掉线====="+param);
                        return;
                    }
                    site.setOnline(Constants.ONE);
                } else {
                    if(Constants.formatIntegerNum(site.getOnline()) == Constants.ZERO){
                        log.error("mqtt消息订阅==============已订阅状态上线====="+param);
                        return;
                    }
                    site.setOnline(Constants.ZERO);
                }
                site.setLastLinkDate(new Date());
                sitesMapper.updateById(site);//更新站点状态
                log.info("mqtt消息订阅=========站点上下线==========成功");
            }
        }catch (BusinessException e){
            e.printStackTrace();