|
|
@@ -10,6 +10,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
|
|
10
|
10
|
import org.springframework.stereotype.Component;
|
|
11
|
11
|
|
|
12
|
12
|
import java.text.SimpleDateFormat;
|
|
|
13
|
+import java.util.Collections;
|
|
13
|
14
|
import java.util.Date;
|
|
14
|
15
|
import java.util.List;
|
|
15
|
16
|
import java.util.concurrent.ExecutorService;
|
|
|
@@ -55,15 +56,17 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
|
|
55
|
56
|
String timestamp = weather.getTimestamp();
|
|
56
|
57
|
String fleetId = weather.getFleet_id();
|
|
57
|
58
|
String controllerId = weather.getController_id();
|
|
|
59
|
+
|
|
58
|
60
|
List<topics> topics = weather.getTopics();
|
|
|
61
|
+ if (topics == null) topics = Collections.emptyList();
|
|
59
|
62
|
List<topics> cmdtopics = weather.getCmd_topics();
|
|
|
63
|
+ if (cmdtopics == null) cmdtopics = Collections.emptyList();
|
|
60
|
64
|
topics faultprot = weather.getFault_prot();
|
|
61
|
65
|
|
|
62
|
66
|
Integer controllercountcount = 0;
|
|
63
|
67
|
for (topics topicsMap : topics) {
|
|
64
|
68
|
Integer count = sysControllerService.selectsyscontrollercount(topicsMap.getPath());
|
|
65
|
69
|
if (count <= 0) {
|
|
66
|
|
- stringRedisTemplate.persist(controllerId);
|
|
67
|
70
|
stringRedisTemplate.opsForHash().put(controllerId + ":" + topicsMap.getName(), "path", topicsMap.getPath());
|
|
68
|
71
|
sysControllerService.insertsyscontroller(controllerId, timestamp, fleetId, topicsMap.getName(), topicsMap.getPath(), topicsMap.getPath().split("/")[1]);
|
|
69
|
72
|
controllercountcount++;
|
|
|
@@ -74,27 +77,28 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
|
|
74
|
77
|
}
|
|
75
|
78
|
}
|
|
76
|
79
|
|
|
77
|
|
- Integer controllercountcmdcount = 0;
|
|
78
|
80
|
for (topics cmdtopicsMap : cmdtopics) {
|
|
79
|
81
|
Integer count = sysControllerService.selectsyscontrollercountcmd(cmdtopicsMap.getPath());
|
|
80
|
82
|
if (count <= 0) {
|
|
81
|
83
|
stringRedisTemplate.opsForHash().put(controllerId + "_cmd:" + cmdtopicsMap.getName(), "path", cmdtopicsMap.getPath());
|
|
82
|
|
- stringRedisTemplate.persist(controllerId);
|
|
83
|
84
|
sysControllerService.insertsyscontrollercmd(controllerId, timestamp, fleetId, cmdtopicsMap.getName(), cmdtopicsMap.getPath());
|
|
84
|
|
- controllercountcmdcount++;
|
|
85
|
85
|
}
|
|
86
|
86
|
}
|
|
87
|
87
|
|
|
88
|
|
- Integer count = sysControllerService.selectsyscontrollercountfault(faultprot.getPath());
|
|
89
|
|
- if (count <= 0) {
|
|
90
|
|
- stringRedisTemplate.opsForHash().put(controllerId + "_fault:" + faultprot.getName(), "path", faultprot.getPath());
|
|
91
|
|
- sysControllerService.insertsyscontrollerfault(controllerId, timestamp, fleetId, faultprot.getName(), faultprot.getPath());
|
|
92
|
|
- stringRedisTemplate.persist(controllerId);
|
|
|
88
|
+ if (faultprot != null) {
|
|
|
89
|
+ Integer count = sysControllerService.selectsyscontrollercountfault(faultprot.getPath());
|
|
|
90
|
+ if (count <= 0) {
|
|
|
91
|
+ stringRedisTemplate.opsForHash().put(controllerId + "_fault:" + faultprot.getName(), "path", faultprot.getPath());
|
|
|
92
|
+ sysControllerService.insertsyscontrollerfault(controllerId, timestamp, fleetId, faultprot.getName(), faultprot.getPath());
|
|
|
93
|
+ }
|
|
93
|
94
|
}
|
|
94
|
95
|
|
|
|
96
|
+ // 统一 persist controllerId,确保其不会被过期删除
|
|
|
97
|
+ stringRedisTemplate.persist(controllerId);
|
|
|
98
|
+
|
|
95
|
99
|
if (controllercountcount > 0) {
|
|
96
|
|
- messageListenerService2.destroy();
|
|
97
|
|
- messageListenerService2.initMqttConnection();
|
|
|
100
|
+ // TODO: 后续建议改为 Spring Event 解耦,避免直接调用其他 Bean 的方法
|
|
|
101
|
+ messageListenerService2.refreshMqttSubscription();
|
|
98
|
102
|
}
|
|
99
|
103
|
}
|
|
100
|
104
|
}
|