111
k94314517
2023-10-28 2fce95a6a35c86c1939f16afe60fc06bbb8315e2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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;
    }
}