钟来 / Luhui · 提交

转到一个项目

作者 钟来
提交 aefbf104bfe295ed5a9ccedd141e01d2a6c59f6f 1 个父辈 1063e307

新版x6开发

内嵌 并排对比
正在显示 17 个修改的文件 包含 199 行增加 和 128 行删除
1 -package com.zhonglai.luhui.mqtt.comm.config;  
2 -  
3 -import org.springframework.beans.factory.annotation.Value;  
4 -import org.springframework.stereotype.Component;  
5 -  
6 -@Component  
7 -public class RedisKeyMqttUser {  
8 - public static String ROLEID;  
9 - public static String USERNAME;  
10 -  
11 - @Value("${mqtt.redis.key.roleid}")  
12 - public void setROLEID(String roleid) {  
13 - ROLEID = roleid;  
14 - }  
15 -  
16 - @Value("${mqtt.redis.key.username}")  
17 - public void setUSERNAME(String username) {  
18 - USERNAME = username;  
19 - }  
20 -}  
@@ -49,9 +49,15 @@ public class BaseDao { @@ -49,9 +49,15 @@ public class BaseDao {
49 List<Object> valueList = new ArrayList<Object>(); 49 List<Object> valueList = new ArrayList<Object>();
50 for(Field field:fields) 50 for(Field field:fields)
51 {// 51 {//
  52 + PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
  53 + if(null !=publicSQLConfig && !publicSQLConfig.isSelect())
  54 + {
  55 + continue;
  56 + }
52 Method method; 57 Method method;
53 try { 58 try {
54 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName())); 59 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName()));
  60 +
55 Object value = method.invoke(object); 61 Object value = method.invoke(object);
56 if(null != value) 62 if(null != value)
57 { 63 {
@@ -164,6 +170,11 @@ public class BaseDao { @@ -164,6 +170,11 @@ public class BaseDao {
164 String values = "("; 170 String values = "(";
165 for(Field field:fields) 171 for(Field field:fields)
166 {// 172 {//
  173 + PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
  174 + if(null !=publicSQLConfig && !publicSQLConfig.isSelect())
  175 + {
  176 + continue;
  177 + }
167 Method method; 178 Method method;
168 try { 179 try {
169 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName())); 180 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName()));
@@ -232,9 +243,14 @@ public class BaseDao { @@ -232,9 +243,14 @@ public class BaseDao {
232 int j = 0; 243 int j = 0;
233 for(int i=0;i<fields.length;i++) 244 for(int i=0;i<fields.length;i++)
234 { 245 {
  246 +
235 Field field = fields[i]; 247 Field field = fields[i];
236 try { 248 try {
237 - 249 + PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
  250 + if(null !=publicSQLConfig && !publicSQLConfig.isSelect())
  251 + {
  252 + continue;
  253 + }
238 Method method = null; 254 Method method = null;
239 try { 255 try {
240 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName())); 256 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName()));
@@ -307,6 +323,11 @@ public class BaseDao { @@ -307,6 +323,11 @@ public class BaseDao {
307 String idName = "id"; 323 String idName = "id";
308 for(Field field:fields) 324 for(Field field:fields)
309 { 325 {
  326 + PublicSQLConfig publicSQLConfig1 = field.getAnnotation(PublicSQLConfig.class);
  327 + if(null !=publicSQLConfig1 && !publicSQLConfig1.isSelect())
  328 + {
  329 + continue;
  330 + }
310 PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class); 331 PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
311 if(null != publicSQLConfig && publicSQLConfig.isPrimarykey()) 332 if(null != publicSQLConfig && publicSQLConfig.isPrimarykey())
312 { 333 {
@@ -401,6 +422,11 @@ public class BaseDao { @@ -401,6 +422,11 @@ public class BaseDao {
401 Field[] fields = clas.getDeclaredFields(); 422 Field[] fields = clas.getDeclaredFields();
402 for(Field field:fields) 423 for(Field field:fields)
403 { 424 {
  425 + PublicSQLConfig publicSQLConfig1 = field.getAnnotation(PublicSQLConfig.class);
  426 + if(null !=publicSQLConfig1 && !publicSQLConfig1.isSelect())
  427 + {
  428 + continue;
  429 + }
404 PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class); 430 PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
405 if(null != publicSQLConfig && publicSQLConfig.isPrimarykey()) 431 if(null != publicSQLConfig && publicSQLConfig.isPrimarykey())
406 { 432 {
@@ -712,6 +738,11 @@ public class BaseDao { @@ -712,6 +738,11 @@ public class BaseDao {
712 for(int i=0;i<fields.length;i++) 738 for(int i=0;i<fields.length;i++)
713 { 739 {
714 Field field = fields[i]; 740 Field field = fields[i];
  741 + PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
  742 + if(null !=publicSQLConfig && !publicSQLConfig.isSelect())
  743 + {
  744 + continue;
  745 + }
715 try { 746 try {
716 Method method; 747 Method method;
717 method = object.getClass().getMethod("get"+ com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName())); 748 method = object.getClass().getMethod("get"+ com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName()));
@@ -783,6 +814,11 @@ public class BaseDao { @@ -783,6 +814,11 @@ public class BaseDao {
783 {// 814 {//
784 Method method; 815 Method method;
785 try { 816 try {
  817 + PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
  818 + if(null !=publicSQLConfig && !publicSQLConfig.isSelect())
  819 + {
  820 + continue;
  821 + }
786 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName())); 822 method = object.getClass().getMethod("get"+com.zhonglai.luhui.mqtt.comm.util.StringUtils.getName(field.getName()));
787 Object value = method.invoke(object); 823 Object value = method.invoke(object);
788 if(null != value) 824 if(null != value)
@@ -845,6 +881,11 @@ public class BaseDao { @@ -845,6 +881,11 @@ public class BaseDao {
845 sb.append("("); 881 sb.append("(");
846 for(Field field:fields) 882 for(Field field:fields)
847 { 883 {
  884 + PublicSQLConfig publicSQLConfig = field.getAnnotation(PublicSQLConfig.class);
  885 + if(null !=publicSQLConfig && !publicSQLConfig.isSelect())
  886 + {
  887 + continue;
  888 + }
848 if(!"".equals(update) ) 889 if(!"".equals(update) )
849 { 890 {
850 sb.append(","); 891 sb.append(",");
@@ -960,7 +1001,6 @@ public class BaseDao { @@ -960,7 +1001,6 @@ public class BaseDao {
960 tableNmae = (String) tObject; 1001 tableNmae = (String) tObject;
961 } 1002 }
962 } catch (NoSuchMethodException e) { 1003 } catch (NoSuchMethodException e) {
963 - System.out.println("未找到getTableName");  
964 } catch (IllegalAccessException e) { 1004 } catch (IllegalAccessException e) {
965 e.printStackTrace(); 1005 e.printStackTrace();
966 } catch (InvocationTargetException e) { 1006 } catch (InvocationTargetException e) {
@@ -8,7 +8,7 @@ import com.ruoyi.system.domain.IotThingsModel; @@ -8,7 +8,7 @@ import com.ruoyi.system.domain.IotThingsModel;
8 */ 8 */
9 public interface ThingsModelBase<T> { 9 public interface ThingsModelBase<T> {
10 void conversionThingsModel(IotThingsModel thingsModel); 10 void conversionThingsModel(IotThingsModel thingsModel);
11 - void addValue(T t); 11 + void addValue(Object t);
12 // @JSONField(serialize=false) 12 // @JSONField(serialize=false)
13 String getView(); 13 String getView();
14 String getSaveView(); 14 String getSaveView();
@@ -11,8 +11,8 @@ public class ArrayModelOutput extends ThingsModelItemBase<JSONArray> @@ -11,8 +11,8 @@ public class ArrayModelOutput extends ThingsModelItemBase<JSONArray>
11 private String arrayType; 11 private String arrayType;
12 12
13 @Override 13 @Override
14 - public void addValue(JSONArray jsonArray) {  
15 - setValue(jsonArray); 14 + public void addValue(Object jsonArray) {
  15 + setValue((JSONArray) jsonArray);
16 } 16 }
17 17
18 @Override 18 @Override
@@ -10,8 +10,8 @@ public class BoolModelOutput extends ThingsModelItemBase<Boolean> @@ -10,8 +10,8 @@ public class BoolModelOutput extends ThingsModelItemBase<Boolean>
10 private String trueText; 10 private String trueText;
11 11
12 @Override 12 @Override
13 - public void addValue(Boolean bl) {  
14 - setValue( bl); 13 + public void addValue(Object bl) {
  14 + setValue((Boolean) bl);
15 } 15 }
16 16
17 @Override 17 @Override
@@ -14,8 +14,11 @@ public class DecimalModelOutput extends ThingsModelItemBase<BigDecimal> @@ -14,8 +14,11 @@ public class DecimalModelOutput extends ThingsModelItemBase<BigDecimal>
14 private String unit; 14 private String unit;
15 15
16 @Override 16 @Override
17 - public void addValue(BigDecimal object) {  
18 - setValue(object); 17 + public void addValue(Object object) {
  18 + if(null != object)
  19 + {
  20 + setValue(new BigDecimal(object.toString()));
  21 + }
19 } 22 }
20 23
21 @Override 24 @Override
@@ -17,8 +17,8 @@ public class IntegerModelOutput extends ThingsModelItemBase<Integer> @@ -17,8 +17,8 @@ public class IntegerModelOutput extends ThingsModelItemBase<Integer>
17 private Integer acy; //精度 17 private Integer acy; //精度
18 18
19 @Override 19 @Override
20 - public void addValue(Integer object) {  
21 - setValue(object); 20 + public void addValue(Object object) {
  21 + setValue((Integer) object);
22 } 22 }
23 23
24 @Override 24 @Override
@@ -36,6 +36,10 @@ public class IntegerModelOutput extends ThingsModelItemBase<Integer> @@ -36,6 +36,10 @@ public class IntegerModelOutput extends ThingsModelItemBase<Integer>
36 { 36 {
37 return null; 37 return null;
38 } 38 }
  39 + if(null == acy)
  40 + {
  41 + acy = 1;
  42 + }
39 BigDecimal bigDecimal = new BigDecimal(getValue().toString()); 43 BigDecimal bigDecimal = new BigDecimal(getValue().toString());
40 return bigDecimal.divide(new BigDecimal(acy),acy.toString().length()-1, RoundingMode.HALF_UP).toString(); 44 return bigDecimal.divide(new BigDecimal(acy),acy.toString().length()-1, RoundingMode.HALF_UP).toString();
41 } 45 }
@@ -8,6 +8,7 @@ import com.zhonglai.luhui.mqtt.dto.MessageCode; @@ -8,6 +8,7 @@ import com.zhonglai.luhui.mqtt.dto.MessageCode;
8 import com.zhonglai.luhui.mqtt.service.db.DeviceService; 8 import com.zhonglai.luhui.mqtt.service.db.DeviceService;
9 import org.apache.rocketmq.common.message.MessageExt; 9 import org.apache.rocketmq.common.message.MessageExt;
10 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; 10 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  11 +import org.apache.rocketmq.spring.annotation.SelectorType;
11 import org.apache.rocketmq.spring.core.RocketMQReplyListener; 12 import org.apache.rocketmq.spring.core.RocketMQReplyListener;
12 import org.slf4j.Logger; 13 import org.slf4j.Logger;
13 import org.slf4j.LoggerFactory; 14 import org.slf4j.LoggerFactory;
@@ -16,7 +17,7 @@ import org.springframework.stereotype.Service; @@ -16,7 +17,7 @@ import org.springframework.stereotype.Service;
16 17
17 18
18 @Service 19 @Service
19 -@RocketMQMessageListener(consumerGroup = "deviceCommand", topic = "lh-mqtt-service-deviceCommand") 20 +@RocketMQMessageListener(consumerGroup = "deviceCommand", topic = "${rocketmq.send-topic:}",selectorType = SelectorType.TAG,selectorExpression = "${rocketmq.send-tag:}")
20 public class RocketMqService implements RocketMQReplyListener<MessageExt, Message> { 21 public class RocketMqService implements RocketMQReplyListener<MessageExt, Message> {
21 private static final Logger log = LoggerFactory.getLogger(MqttCallback.class); 22 private static final Logger log = LoggerFactory.getLogger(MqttCallback.class);
22 23
@@ -29,6 +30,7 @@ public class RocketMqService implements RocketMQReplyListener<MessageExt, Messag @@ -29,6 +30,7 @@ public class RocketMqService implements RocketMQReplyListener<MessageExt, Messag
29 // String clint = MessageUtil.getReplyToClient(messageExt); 30 // String clint = MessageUtil.getReplyToClient(messageExt);
30 31
31 String str = new String(messageExt.getBody()); 32 String str = new String(messageExt.getBody());
  33 + log.info("消息body{}",str);
32 DeviceCommandApi deviceCommandApi = JSON.parseObject(str, DeviceCommandApi.class); 34 DeviceCommandApi deviceCommandApi = JSON.parseObject(str, DeviceCommandApi.class);
33 try { 35 try {
34 return deviceCommandApi.invokeApi(deviceService); 36 return deviceCommandApi.invokeApi(deviceService);
@@ -88,7 +88,6 @@ public class BusinessDataUpdateService { @@ -88,7 +88,6 @@ public class BusinessDataUpdateService {
88 }else{ 88 }else{
89 iotDevice.setListen_service_ip(SysParameter.service_ip+":"+port+contextPath); 89 iotDevice.setListen_service_ip(SysParameter.service_ip+":"+port+contextPath);
90 } 90 }
91 - logger.info("更新网关数据{}",iotDevice);  
92 if(null== iotDevice.getStatus() || 1 == iotDevice.getStatus() || 4==iotDevice.getStatus()) 91 if(null== iotDevice.getStatus() || 1 == iotDevice.getStatus() || 4==iotDevice.getStatus())
93 { 92 {
94 iotDevice.setStatus(3); 93 iotDevice.setStatus(3);
@@ -99,8 +98,11 @@ public class BusinessDataUpdateService { @@ -99,8 +98,11 @@ public class BusinessDataUpdateService {
99 serverDto.setIotDevice(iotDevice); 98 serverDto.setIotDevice(iotDevice);
100 }else{ //终端 99 }else{ //终端
101 IotTerminal iotTerminal = translateTerminal(type,key,olddevice,jsData,isOperLog,serverDto); 100 IotTerminal iotTerminal = translateTerminal(type,key,olddevice,jsData,isOperLog,serverDto);
102 - logger.info("更新终端数据{}",iotTerminal);  
103 iotTerminal.setData_update_time(DateUtils.getNowTimeMilly()); 101 iotTerminal.setData_update_time(DateUtils.getNowTimeMilly());
  102 + if(null== iotTerminal.getOnline() || 1 == iotTerminal.getOnline() || 4==iotTerminal.getOnline())
  103 + {
  104 + iotTerminal.setOnline(3);
  105 + }
104 serverDto.getIotTerminalList().add(iotTerminal); 106 serverDto.getIotTerminalList().add(iotTerminal);
105 } 107 }
106 } 108 }
@@ -18,7 +18,6 @@ import org.springframework.beans.factory.annotation.Autowired; @@ -18,7 +18,6 @@ import org.springframework.beans.factory.annotation.Autowired;
18 import org.springframework.beans.factory.annotation.Value; 18 import org.springframework.beans.factory.annotation.Value;
19 import org.springframework.stereotype.Service; 19 import org.springframework.stereotype.Service;
20 20
21 -import javax.annotation.PostConstruct;  
22 import java.util.Map; 21 import java.util.Map;
23 import java.util.concurrent.TimeUnit; 22 import java.util.concurrent.TimeUnit;
24 23
@@ -31,42 +30,43 @@ public class ClienNoticeService { @@ -31,42 +30,43 @@ public class ClienNoticeService {
31 30
32 @Autowired 31 @Autowired
33 private TerminalService terminalService; 32 private TerminalService terminalService;
34 -  
35 - private ExpiringMap<String, ClienConnection> clienConnectionMap;  
36 -  
37 @Value("${mqtt.client.operationTime}") 33 @Value("${mqtt.client.operationTime}")
38 private long operationTime; //客户端操作时间 34 private long operationTime; //客户端操作时间
39 35
  36 + // maxSize: 设置最大值,添加第11个entry时，会导致第1个立马过期(即使没到过期时间)
  37 + // expiration：设置每个key有效时间10s, 如果key不设置过期时间，key永久有效。
  38 + // variableExpiration: 允许更新过期时间值,如果不设置variableExpiration，不允许后面更改过期时间,一旦执行更改过期时间操作会抛异常UnsupportedOperationException
  39 + // policy:
  40 + // CREATED: 只在put和replace方法清零过期时间
  41 + // ACCESSED: 在CREATED策略基础上增加, 在还没过期时get方法清零过期时间。
  42 + // 清零过期时间也就是重置过期时间，重新计算过期时间.
  43 + private static ExpiringMap<String, ClienConnection> clienConnectionMap = ExpiringMap.builder().maxSize(20000).expiration(5, TimeUnit.SECONDS)
  44 + .asyncExpirationListener(new ExpirationListener<String, ClienConnection>() {
  45 + @Override
  46 + public void expired(String s, ClienConnection clienConnection) {
  47 + log.info("{} 通道消失了>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>",s);
  48 + clienConnection.close();
  49 + }
  50 + })
  51 + .expirationPolicy(ExpirationPolicy.CREATED).build();
  52 +
40 @Value("#{${mqtt.top_return_map}}") 53 @Value("#{${mqtt.top_return_map}}")
41 private Map<String,String> top_return_map; //topic返回的对应关系 54 private Map<String,String> top_return_map; //topic返回的对应关系
42 55
43 - @PostConstruct  
44 - public void init()  
45 - {  
46 - // maxSize: 设置最大值,添加第11个entry时，会导致第1个立马过期(即使没到过期时间)  
47 - // expiration：设置每个key有效时间10s, 如果key不设置过期时间，key永久有效。  
48 - // variableExpiration: 允许更新过期时间值,如果不设置variableExpiration，不允许后面更改过期时间,一旦执行更改过期时间操作会抛异常UnsupportedOperationException  
49 - // policy:  
50 - // CREATED: 只在put和replace方法清零过期时间  
51 - // ACCESSED: 在CREATED策略基础上增加, 在还没过期时get方法清零过期时间。  
52 - // 清零过期时间也就是重置过期时间，重新计算过期时间.  
53 - clienConnectionMap = ExpiringMap.builder().maxSize(20000).expiration(operationTime, TimeUnit.SECONDS)  
54 - .asyncExpirationListener((ExpirationListener<String, ClienConnection>) (s, clienConnection) -> clienConnection.close())  
55 - .expirationPolicy(ExpirationPolicy.CREATED).build();  
56 - }  
57 -  
58 public Message sendMessage(Topic topic, MqttMessage mqttMessage) throws MqttException, InterruptedException { 56 public Message sendMessage(Topic topic, MqttMessage mqttMessage) throws MqttException, InterruptedException {
59 //设置通知渠道 57 //设置通知渠道
60 ClienConnection clienConnection = new ClienConnectionImpl(); 58 ClienConnection clienConnection = new ClienConnectionImpl();
61 - log.info("{} {} {} {}",topic.generateClienKey(),topic.getTopicType(),top_return_map,clienConnection);  
62 - clienConnectionMap.put(topic.generateClienKey().replace(topic.getTopicType(),top_return_map.get(topic.getTopicType())),clienConnection);  
63 - 59 + String key = topic.generateClienKey().replace(topic.getTopicType(),top_return_map.get(topic.getTopicType()));
  60 + log.info("设置通知渠道 {} {}",key,clienConnection);
  61 + clienConnectionMap.put(key,clienConnection);
64 sendMessage(topic.generateSendMessageTopic(),mqttMessage); 62 sendMessage(topic.generateSendMessageTopic(),mqttMessage);
65 synchronized(clienConnection) 63 synchronized(clienConnection)
66 { 64 {
67 log.info("{}等待通知",topic.getClientid()); 65 log.info("{}等待通知",topic.getClientid());
68 clienConnection.wait(operationTime*1000+3000l); 66 clienConnection.wait(operationTime*1000+3000l);
69 } 67 }
  68 + //清楚通道
  69 + clienConnectionMap.remove(key);
70 log.info("{}收到通知{}",topic.getClientid(),clienConnection.getReplyMessage().getMessage()); 70 log.info("{}收到通知{}",topic.getClientid(),clienConnection.getReplyMessage().getMessage());
71 Message message = clienConnection.getReplyMessage(); 71 Message message = clienConnection.getReplyMessage();
72 log.info("{}返回通知{}",topic.getClientid(),message); 72 log.info("{}返回通知{}",topic.getClientid(),message);
@@ -119,19 +119,21 @@ public class DataModeAnalysisService { @@ -119,19 +119,21 @@ public class DataModeAnalysisService {
119 serverDto.getLogDeviceOperationList().add(dviceLogService.newLogDeviceOperation(id,thingsModelBase.getSaveView(),null,controlModel+thingsModelItemBase.getName()+"为"+thingsModelBase.getView(),jsData.toString())); 119 serverDto.getLogDeviceOperationList().add(dviceLogService.newLogDeviceOperation(id,thingsModelBase.getSaveView(),null,controlModel+thingsModelItemBase.getName()+"为"+thingsModelBase.getView(),jsData.toString()));
120 } 120 }
121 121
122 -  
123 - if(null != thingsModel && null != thingsModel.getType() && 2==thingsModel.getType())  
124 - {  
125 - config.put(key,thingsModelBase);  
126 - data.put(key,thingsModelBase);  
127 - }else 122 + switch (thingsModel.getIs_config())
128 { 123 {
129 - if(null != thingsModel && null !=thingsModel.getIs_config() && 1==thingsModel.getIs_config())  
130 - { 124 + case 0:
  125 + data.put(key,thingsModelBase);
  126 + break;
  127 + case 1:
131 config.put(key,thingsModelBase); 128 config.put(key,thingsModelBase);
132 - }else{ 129 + break;
  130 + case 2:
  131 + config.put(key,thingsModelBase);
  132 + data.put(key,thingsModelBase);
  133 + break;
  134 + default:
133 data.put(key,thingsModelBase); 135 data.put(key,thingsModelBase);
134 - } 136 + break;
135 } 137 }
136 } 138 }
137 SaveDataDto saveDataDto = new SaveDataDto(); 139 SaveDataDto saveDataDto = new SaveDataDto();
@@ -49,8 +49,12 @@ public class TerminalService { @@ -49,8 +49,12 @@ public class TerminalService {
49 private String broker; 49 private String broker;
50 @Value("${mqtt.clientId}") 50 @Value("${mqtt.clientId}")
51 private String clientId; 51 private String clientId;
52 - @Value("${mqtt.topics}")  
53 - private String topics; 52 + @Value("${mqtt.roleid}")
  53 + private String roleid;
  54 + @Value("#{'${mqtt.mqtt_usernames}'.split(',')}")
  55 + private List<String> mqtt_usernames;
  56 + @Value("#{'${mqtt.topics}'.split(',')}")
  57 + private List<String> topics;
54 @Value("${mqtt.username}") 58 @Value("${mqtt.username}")
55 private String username; 59 private String username;
56 @Value("${mqtt.password}") 60 @Value("${mqtt.password}")
@@ -80,7 +84,34 @@ public class TerminalService { @@ -80,7 +84,34 @@ public class TerminalService {
80 } 84 }
81 85
82 public void subscribe() throws MqttException { 86 public void subscribe() throws MqttException {
83 - mqttclient.subscribe(topics.split(",")); 87 + List<String> ts = getCompletionTopics();
  88 + mqttclient.subscribe(ts.toArray(new String[ts.size()]));
  89 + }
  90 +
  91 + public List<String> getCompletionTopics()
  92 + {
  93 + List<String> ts = new ArrayList<>();
  94 + for(String mqtt_username:mqtt_usernames)
  95 + {
  96 + StringBuffer topic = new StringBuffer();
  97 + topic.append("/");
  98 + topic.append(roleid);
  99 + topic.append("/");
  100 + topic.append(mqtt_username);
  101 + topic.append("/");
  102 + topic.append("+");
  103 + topic.append("/");
  104 + topic.append("+");
  105 + for(String tc:topics)
  106 + {
  107 + StringBuffer t = new StringBuffer(topic);
  108 + t.append("/");
  109 + t.append(tc);
  110 +
  111 + ts.add(t.toString());
  112 + }
  113 + }
  114 + return ts;
84 } 115 }
85 116
86 @PostConstruct 117 @PostConstruct
@@ -103,7 +134,8 @@ public class TerminalService { @@ -103,7 +134,8 @@ public class TerminalService {
103 { 134 {
104 List<String> roleids=new ArrayList<>(); 135 List<String> roleids=new ArrayList<>();
105 List<String> usernames= new ArrayList<>(); 136 List<String> usernames= new ArrayList<>();
106 - for(String topicstr:topics.split(",")) 137 + List<String> ts = getCompletionTopics();
  138 + for(String topicstr:ts)
107 { 139 {
108 Topic topic = new Topic(topicstr); 140 Topic topic = new Topic(topicstr);
109 if(null != topic) 141 if(null != topic)
@@ -41,10 +41,12 @@ public class CacheServiceImpl implements CacheService { @@ -41,10 +41,12 @@ public class CacheServiceImpl implements CacheService {
41 IotDevice iotDevice = serverDto.getIotDevice(); 41 IotDevice iotDevice = serverDto.getIotDevice();
42 if(null == iotDevice) 42 if(null == iotDevice)
43 { 43 {
44 - return false; 44 + iotDevice = new IotDevice();
  45 + iotDevice.setClient_id(topic.getClientid());
  46 + iotDevice.setStatus(3);
45 } 47 }
46 setIotDeviceToRedis(iotDevice); 48 setIotDeviceToRedis(iotDevice);
47 - log.info("{} 缓存更新 {}",topic,iotDevice); 49 + log.info("缓存更新 {} ",topic);
48 50
49 List<IotTerminal> list = serverDto.getIotTerminalList(); 51 List<IotTerminal> list = serverDto.getIotTerminalList();
50 if(null != list && list.size() != 0 ) 52 if(null != list && list.size() != 0 )
@@ -54,7 +56,7 @@ public class CacheServiceImpl implements CacheService { @@ -54,7 +56,7 @@ public class CacheServiceImpl implements CacheService {
54 setIotTerminalToRedis(iotTerminal); 56 setIotTerminalToRedis(iotTerminal);
55 } 57 }
56 } 58 }
57 - log.info("{} 缓存更新 {}",topic,list); 59 + log.info("缓存更新 {} ",topic);
58 return true; 60 return true;
59 } 61 }
60 62
@@ -81,7 +83,7 @@ public class CacheServiceImpl implements CacheService { @@ -81,7 +83,7 @@ public class CacheServiceImpl implements CacheService {
81 if(null == iotTerminal) 83 if(null == iotTerminal)
82 { 84 {
83 String imei = id.split("_")[0]; 85 String imei = id.split("_")[0];
84 - IotDevice iotDevice = getIotDevice(id); 86 + IotDevice iotDevice = getIotDevice(imei);
85 if(null == iotDevice) 87 if(null == iotDevice)
86 { 88 {
87 return null; 89 return null;
@@ -25,12 +25,15 @@ public class DataPersistenceServiceImpl extends DataPersistenceService { @@ -25,12 +25,15 @@ public class DataPersistenceServiceImpl extends DataPersistenceService {
25 private DeviceLogService dviceLogService; 25 private DeviceLogService dviceLogService;
26 @Override 26 @Override
27 public void persistence(Topic topic, ServerDto serverDto) { 27 public void persistence(Topic topic, ServerDto serverDto) {
  28 + log.info("更新数据{}",topic);
28 IotDevice iotDevice = serverDto.getIotDevice(); 29 IotDevice iotDevice = serverDto.getIotDevice();
29 if(null == iotDevice) 30 if(null == iotDevice)
30 { 31 {
31 - return ; 32 + iotDevice = new IotDevice();
  33 + iotDevice.setClient_id(topic.getClientid());
  34 + iotDevice.setStatus(3);
32 } 35 }
33 - log.info("更新网关数据{}",iotDevice); 36 +
34 baseDao.update(iotDevice); 37 baseDao.update(iotDevice);
35 38
36 List<IotTerminal> list = serverDto.getIotTerminalList(); 39 List<IotTerminal> list = serverDto.getIotTerminalList();
@@ -38,7 +41,6 @@ public class DataPersistenceServiceImpl extends DataPersistenceService { @@ -38,7 +41,6 @@ public class DataPersistenceServiceImpl extends DataPersistenceService {
38 { 41 {
39 for (IotTerminal iotTerminal:list) 42 for (IotTerminal iotTerminal:list)
40 { 43 {
41 - log.info("更新终端数据{}",iotTerminal);  
42 baseDao.update(iotTerminal); 44 baseDao.update(iotTerminal);
43 } 45 }
44 } 46 }
@@ -21,6 +21,8 @@ import com.zhonglai.luhui.mqtt.service.db.mode.TerminalDataThingsModeService; @@ -21,6 +21,8 @@ import com.zhonglai.luhui.mqtt.service.db.mode.TerminalDataThingsModeService;
21 import org.apache.commons.lang3.StringUtils; 21 import org.apache.commons.lang3.StringUtils;
22 import org.eclipse.paho.client.mqttv3.MqttException; 22 import org.eclipse.paho.client.mqttv3.MqttException;
23 import org.eclipse.paho.client.mqttv3.MqttMessage; 23 import org.eclipse.paho.client.mqttv3.MqttMessage;
  24 +import org.slf4j.Logger;
  25 +import org.slf4j.LoggerFactory;
24 import org.springframework.beans.factory.annotation.Autowired; 26 import org.springframework.beans.factory.annotation.Autowired;
25 import org.springframework.stereotype.Service; 27 import org.springframework.stereotype.Service;
26 28
@@ -31,6 +33,7 @@ import java.util.Set; @@ -31,6 +33,7 @@ import java.util.Set;
31 33
32 @Service 34 @Service
33 public class DeviceService { 35 public class DeviceService {
  36 + private static final Logger log = LoggerFactory.getLogger(DeviceService.class);
34 37
35 @Autowired 38 @Autowired
36 private ClienNoticeService clienNoticeService; 39 private ClienNoticeService clienNoticeService;
@@ -152,7 +155,7 @@ public class DeviceService { @@ -152,7 +155,7 @@ public class DeviceService {
152 * @throws InterruptedException 155 * @throws InterruptedException
153 */ 156 */
154 public Message read(String clienid,Map<String,Object> map) throws MqttException, InterruptedException { 157 public Message read(String clienid,Map<String,Object> map) throws MqttException, InterruptedException {
155 - 158 + log.info("api请求读取 {} 参数 {}",map);
156 if(null == map || map.size() ==0) 159 if(null == map || map.size() ==0)
157 { 160 {
158 return new Message(MessageCode.DEFAULT_FAIL_CODE,"参数验证失败"); 161 return new Message(MessageCode.DEFAULT_FAIL_CODE,"参数验证失败");
@@ -27,53 +27,53 @@ public class ReadReqTopic implements BusinessAgreement<ReadReqDto> { @@ -27,53 +27,53 @@ public class ReadReqTopic implements BusinessAgreement<ReadReqDto> {
27 @Autowired 27 @Autowired
28 private ClienNoticeService clienNoticeService; //客户端通知服务 28 private ClienNoticeService clienNoticeService; //客户端通知服务
29 @Autowired 29 @Autowired
30 - private BusinessDataUpdateService businessDataUpdateService ; 30 + private TerminalDataThingsModeService terminalDataThingsModeService;
31 31
32 @Override 32 @Override
33 public ServerDto analysis(Topic topic, ReadReqDto data) { 33 public ServerDto analysis(Topic topic, ReadReqDto data) {
34 if(1==data.getCode()) 34 if(1==data.getCode())
35 { 35 {
36 -// JSONObject vjsonObject = data.getData().clone();  
37 - businessDataUpdateService.updataDta(BusinessDataUpdateService.Type.ADD,topic,true,data); 36 + JSONObject vjsonObject = data.getData().clone();
  37 +// businessDataUpdateService.updataDta(BusinessDataUpdateService.Type.ADD,topic,true,data);
38 38
39 -// if(null != vjsonObject && vjsonObject.size() !=0 )  
40 -// {  
41 -// for(String vkey:vjsonObject.keySet())  
42 -// {  
43 -// JSONObject jsData = vjsonObject.getJSONObject(vkey);  
44 -// for(String key:jsData.keySet())  
45 -// {  
46 -// IotThingsModel thingsModel = terminalDataThingsModeService.getIotThingsModel(topic.getUsername(),key);  
47 -// if(null == thingsModel) //没有配置的 都按字符串处理  
48 -// {  
49 -// thingsModel = new IotThingsModel();  
50 -// thingsModel.setData_type(ThingsModelDataTypeEnum.STRING.name());  
51 -// thingsModel.setIdentifier(key);  
52 -// thingsModel.setModel_name(key);  
53 -// thingsModel.setIs_top(0);  
54 -// thingsModel.setIs_monitor(0);  
55 -// thingsModel.setIs_save_log(0);  
56 -// thingsModel.setIs_config(0);  
57 -// JSONObject jsonObject = new JSONObject();  
58 -// jsonObject.put("maxLength",255);  
59 -// thingsModel.setSpecs(jsonObject.toString());  
60 -// }  
61 -// String data_type = thingsModel.getData_type().toUpperCase();  
62 -// if(!EnumUtils.isValidEnum(ThingsModelDataTypeEnum.class,data_type))  
63 -// {  
64 -// data_type = ThingsModelDataTypeEnum.STRING.name();  
65 -// }  
66 -// Class<ThingsModelBase> aClass = Enum.valueOf(ThingsModelDataTypeEnum.class,data_type).getaClass();  
67 -// ThingsModelBase thingsModelBase = JSON.parseObject(thingsModel.getSpecs(),aClass);  
68 -// thingsModelBase.conversionThingsModel(thingsModel);  
69 -//  
70 -// thingsModelBase.addValue(jsData.get(key));  
71 -// jsData.put(key,thingsModelBase);  
72 -// }  
73 -// vjsonObject.put(vkey,jsData);  
74 -// }  
75 -// data.setData(vjsonObject);  
76 -// } 39 + if(null != vjsonObject && vjsonObject.size() !=0 )
  40 + {
  41 + for(String vkey:vjsonObject.keySet())
  42 + {
  43 + JSONObject jsData = vjsonObject.getJSONObject(vkey);
  44 + for(String key:jsData.keySet())
  45 + {
  46 + IotThingsModel thingsModel = terminalDataThingsModeService.getIotThingsModel(topic.getUsername(),key);
  47 + if(null == thingsModel) //没有配置的 都按字符串处理
  48 + {
  49 + thingsModel = new IotThingsModel();
  50 + thingsModel.setData_type(ThingsModelDataTypeEnum.STRING.name());
  51 + thingsModel.setIdentifier(key);
  52 + thingsModel.setModel_name(key);
  53 + thingsModel.setIs_top(0);
  54 + thingsModel.setIs_monitor(0);
  55 + thingsModel.setIs_save_log(0);
  56 + thingsModel.setIs_config(0);
  57 + JSONObject jsonObject = new JSONObject();
  58 + jsonObject.put("maxLength",255);
  59 + thingsModel.setSpecs(jsonObject.toString());
  60 + }
  61 + String data_type = thingsModel.getData_type().toUpperCase();
  62 + if(!EnumUtils.isValidEnum(ThingsModelDataTypeEnum.class,data_type))
  63 + {
  64 + data_type = ThingsModelDataTypeEnum.STRING.name();
  65 + }
  66 + Class<ThingsModelBase> aClass = Enum.valueOf(ThingsModelDataTypeEnum.class,data_type).getaClass();
  67 + ThingsModelBase thingsModelBase = JSON.parseObject(thingsModel.getSpecs(),aClass);
  68 + thingsModelBase.conversionThingsModel(thingsModel);
  69 +
  70 + thingsModelBase.addValue(jsData.get(key));
  71 + jsData.put(key,thingsModelBase);
  72 + }
  73 + vjsonObject.put(vkey,jsData);
  74 + }
  75 + data.setData(vjsonObject);
  76 + }
77 clienNoticeService.replySendMessage(topic, message -> { 77 clienNoticeService.replySendMessage(topic, message -> {
78 if(null != data.getData()) 78 if(null != data.getData())
79 { 79 {
@@ -43,21 +43,18 @@ mqtt: @@ -43,21 +43,18 @@ mqtt:
43 broker: tcp://175.24.61.68:1883 43 broker: tcp://175.24.61.68:1883
44 #唯一标识 44 #唯一标识
45 clientId: ${random.uuid} 45 clientId: ${random.uuid}
  46 + #公司id
  47 + roleid: 2
  48 + mqtt_usernames: 6_WP
46 #订阅的topic 49 #订阅的topic
47 - topics: "/2/6_WP/+/+/ADD_POST,/2/6_WP/+/+/ALL_POST,/2/6_WP/+/+/DB_TOPIC_DISTRIBUTE,/2/6_WP/+/+/GET/+,/2/6_WP/+/+/online,/2/6_WP/+/+/PUT_REQ/+,/2/6_WP/+/+/READ_REQ/+" 50 + topics: ADD_POST,ALL_POST,DB_TOPIC_DISTRIBUTE,GET/+,online,PUT_REQ/+,READ_REQ/+
48 topicconfig: "/{{roleid}}/{{username}}/{{clientid}}/{{payloadtype}}/{{topicType}}/{{messageid}}" 51 topicconfig: "/{{roleid}}/{{username}}/{{clientid}}/{{payloadtype}}/{{topicType}}/{{messageid}}"
49 top_return_map: '{"PUT":"PUT_REQ","READ":"READ_REQ"}' 52 top_return_map: '{"PUT":"PUT_REQ","READ":"READ_REQ"}'
50 username: sysuser 53 username: sysuser
51 password: "!@#1qaz" 54 password: "!@#1qaz"
52 client: 55 client:
53 #客户端操作时间 56 #客户端操作时间
54 - operationTime: 5  
55 - redis:  
56 - key:  
57 - #角色  
58 - roleid:  
59 - #用户  
60 - username: 57 + operationTime: 10
61 58
62 sys: 59 sys:
63 redis: 60 redis:
@@ -67,4 +64,6 @@ sys: @@ -67,4 +64,6 @@ sys:
67 #rocketmq配置信息 64 #rocketmq配置信息
68 rocketmq: 65 rocketmq:
69 #nameservice服务器地址（多个以英文逗号隔开） 66 #nameservice服务器地址（多个以英文逗号隔开）
70 - name-server: 47.115.144.179:9876  
  67 + name-server: 47.115.144.179:9876
  68 + send-topic: lh-mqtt-service-deviceCommand-test
  69 + send-tag: 1