| 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; | 
| import com.doumee.core.mqtt.service.MqttToolService; | 
| 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.service.business.*; | 
| import lombok.extern.slf4j.Slf4j; | 
| import org.apache.commons.lang3.StringUtils; | 
| import org.springframework.beans.factory.annotation.Autowired; | 
| import org.springframework.context.annotation.Lazy; | 
| import org.springframework.stereotype.Service; | 
|   | 
| import javax.annotation.PostConstruct; | 
| import java.util.Date; | 
|   | 
| /** | 
|  *  与硬件对接服务 | 
|  * @author 江蹄蹄 | 
|  * @date 2023/10/09 18:06 | 
|  */ | 
| @Service | 
| @Slf4j | 
| public class DeviceSubscribeServiceImpl implements DeviceSubcribeService { | 
|     @Autowired | 
|     @Lazy | 
|     MemberRidesService memberRidesService; | 
|     @Autowired | 
|     private MqttLogMapper mqttLogMapper; | 
|     @Autowired | 
|     private MqttConfig mqttConfig; | 
|     @Override | 
|     public void listener(String msgId,String param,String topic) { | 
|        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) ; | 
|         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[] 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(); | 
|         String logInfo = ""; | 
|         int result =0; | 
|         try { | 
|             if(StringUtils.contains(topic, info)){ | 
|                 //如果锁头信息上报 | 
|                 Locks locks  = JSONObject.parseObject(param, Locks.class); | 
|                 locks.setSiteId(siteid); | 
|                 locks.setCode(lockid); | 
|                 locks.setInfo(logId); | 
|                 result = memberRidesService.mqttLockInfoEvent(locks); | 
|                 logInfo = "mqtt消息订阅锁头信息"; | 
|                 log.info("mqtt消息订阅=========锁信息==========成功"); | 
|             } | 
|             if(StringUtils.contains(topic, closeLock)){ | 
|                 //如果还车上报 | 
|                 JSONObject pjson  = JSONObject.parseObject(param); | 
|                 MemberRides bikes = new MemberRides(); | 
|                 bikes.setBikeCode(pjson.getString("bikeCode")); | 
|                 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){ | 
|             e.printStackTrace(); | 
|             logInfo = "mqtt消息订阅异常==="+e.getMessage(); | 
|             result =1; | 
|         } | 
|         createSubLog(topic,msgId,logId,result,param,logInfo); | 
|     } | 
|     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.setResult(result); | 
|         log.setTopic(topic); | 
|         log.setClientid(mqttConfig.getClientid()); | 
|         log.setHostInfo(mqttConfig.getHost()); | 
|         log.setInfo(JSONObject.toJSONString(mqttConfig)); | 
|         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) { | 
|         String[] ss = topic.split("/"); | 
|         if(ss.length>2){ | 
|             return  ss[ss.length-2]; | 
|         } | 
|         return null; | 
| //        topic = topic.substring(0,topic.lastIndexOf("/")); | 
| //        String id =  topic.substring( topic.lastIndexOf("/")+1); | 
| //        return  id; | 
|     } | 
| } |