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.MemberRidesMapper; import com.doumee.dao.business.MqttLogMapper; import com.doumee.dao.business.SitesMapper; import com.doumee.dao.business.model.*; import com.doumee.service.business.*; import io.swagger.models.auth.In; 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; @Lazy @Autowired SitesMapper sitesMapper; @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) ; 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 < 4 || (!StringUtils.contains(topic, info) &&!StringUtils.contains(topic, health) &&!StringUtils.contains(topic,closeLock))){ log.error("mqtt消息订阅===========无效数据========"+param); return; } } MqttLog mqttLog = new MqttLog(); mqttLog.setMsgId(msgId); mqttLog.setTopic(topic); int msgCount =mqttLogMapper.selectCount(new QueryWrapper().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(!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)){ //如果锁头信息上报 Integer lockid = Constants.formatIntegerFromStr(ss[3]) ;//锁头编码 if( lockid == null){ //如果锁头编码为空 log.error("mqtt消息订阅==============无效数据====="+topic+param); return; } 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消息订阅=========锁信息==========成功"); }else if(StringUtils.contains(topic, closeLock)){ //如果还车上报 Integer lockid = Constants.formatIntegerFromStr(ss[3]);//锁头编码 if( lockid == null){ //如果锁头编码为空 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(); 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; } }