| | |
| | | package com.doumee.service.business.impl; |
| | | |
| | | import com.alibaba.fastjson.JSONObject; |
| | | import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; |
| | | import com.doumee.core.constants.Constants; |
| | | import com.doumee.core.exception.BusinessException; |
| | | import com.doumee.core.mqtt.config.MqttConfig; |
| | |
| | | @Autowired |
| | | private MqttConfig mqttConfig; |
| | | @Override |
| | | public void listener(String param,String topic) { |
| | | public void listener(String msgId,String param,String topic) { |
| | | log.info("mqtt消息订阅==================="+param); |
| | | String info = Constants.MqttTopic.lockInfo.substring(Constants.MqttTopic.lockInfo.lastIndexOf("/")+1) ; |
| | | String closeLock = Constants.MqttTopic.lockInfo.substring(Constants.MqttTopic.closeLock.lastIndexOf("/")+1) ; |
| | | 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) ; |
| | | if(topic.indexOf(Constants.MqttTopic.topic_index)!=0 |
| | | ||topic.split("/").length < 5 |
| | | || (!StringUtils.contains(topic, info) |
| | | &&!StringUtils.contains(topic,closeLock))){ |
| | | log.error("mqtt消息订阅===========无效数据========"+param); |
| | | return; |
| | | } |
| | | String lockid = getLockIdFromTopic(topic); |
| | | if(StringUtils.isBlank(lockid)){ |
| | | |
| | | 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); |
| | | if(mqttLogMapper.selectCount(new QueryWrapper<MqttLog>().lambda().eq(MqttLog::getMsgId, msgId)) >0){ |
| | | log.error("mqtt消息订阅==============已消费数据====="+param); |
| | | return; |
| | | } |
| | | String logId =Constants.getUUID(); |
| | |
| | | if(StringUtils.contains(topic, info)){ |
| | | //如果锁头信息上报 |
| | | Locks locks = JSONObject.parseObject(param, Locks.class); |
| | | locks.setId(lockid); |
| | | locks.setSiteId(siteid); |
| | | locks.setCode(lockid); |
| | | locks.setInfo(logId); |
| | | result = memberRidesService.mqttLockInfoEvent(locks); |
| | | logInfo = "mqtt消息订阅锁头信息"; |
| | |
| | | JSONObject pjson = JSONObject.parseObject(param); |
| | | MemberRides bikes = new MemberRides(); |
| | | bikes.setBikeCode(pjson.getString("bikeCode")); |
| | | bikes.setBackLockId( pjson.getString("lockId")); |
| | | bikes.setBackLockId( pjson.getString("siteId")); |
| | | bikes.setBackLockId( lockid); |
| | | bikes.setBackSiteId( siteid); |
| | | bikes.setBackCommondId(logId); |
| | | result = memberRidesService.mqttCloseBikeEvent(bikes); |
| | | logInfo = "mqtt消息订阅还车消息"; |
| | | log.info("mqtt消息订阅=========还车==========成功"); |
| | | } |
| | | }catch (BusinessException e){ |
| | | e.printStackTrace(); |
| | | logInfo = "mqtt消息订阅错误==="+e.getMessage(); |
| | | result =1; |
| | | }catch (Exception e){ |
| | | logInfo = "mqtt消息订阅异常==="; |
| | | e.printStackTrace(); |
| | | logInfo = "mqtt消息订阅异常==="+e.getMessage(); |
| | | result =1; |
| | | } |
| | | createSubLog(topic,logId,result,param,logInfo); |
| | | createSubLog(topic,msgId,logId,result,param,logInfo); |
| | | } |
| | | private void createSubLog(String topic, String logId, int result,String param,String info) { |
| | | private void createSubLog(String topic, String msgId,String logId, int result,String param,String info) { |
| | | MqttLog log = new MqttLog(); |
| | | log.setId(logId); |
| | | log.setCreateDate(new Date()); |
| | |
| | | log.setType(Constants.ZERO); |
| | | log.setMsg(param); |
| | | log.setInfo(info); |
| | | log.setMsgId(msgId); |
| | | mqttLogMapper.insert(log); |
| | | } |
| | | private String getLockIdFromTopic(String topic) { |