|  |  |  | 
|---|
|  |  |  | 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); | 
|---|
|  |  |  | int msgCount  =mqttLogMapper.selectCount(new QueryWrapper<MqttLog>().lambda().eq(MqttLog::getMsg, param).eq(MqttLog::getType, Constants.ZERO)); | 
|---|
|  |  |  | if(msgCount>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.setIsdeleted(Constants.ZERO); | 
|---|
|  |  |  | log.setMsgId(msgId); | 
|---|
|  |  |  | mqttLogMapper.insert(log); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | private String getLockIdFromTopic(String topic) { | 
|---|