From db82b41a3018de6aa5a2622677018026bca4d59f Mon Sep 17 00:00:00 2001
From: bsw215583320 <baoshiwei121@163.com>
Date: 星期五, 08 十一月 2024 11:28:47 +0800
Subject: [PATCH] Merge remote-tracking branch 'origin/herb' into herb
---
jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java | 119 +++++++++++++++++++++++++++++++++++++----------------------
1 files changed, 74 insertions(+), 45 deletions(-)
diff --git a/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java b/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java
index 565cebd..e9e23cf 100644
--- a/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java
+++ b/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttSampleCallback.java
@@ -14,8 +14,10 @@
import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.base.service.BaseCommonService;
import org.jeecg.modules.dry.api.EmqxApi;
+import org.jeecg.modules.dry.entity.DryEqpType;
import org.jeecg.modules.dry.entity.DryEquipment;
import org.jeecg.modules.dry.entity.DryShop;
+import org.jeecg.modules.dry.service.IDryEqpTypeService;
import org.jeecg.modules.dry.service.IDryEquipmentService;
import org.jeecg.modules.dry.service.IDryRealTimeDataService;
import org.jeecg.modules.dry.service.IDryShopService;
@@ -47,6 +49,9 @@
@Autowired
private IDryEquipmentService equipmentService;
+
+ @Autowired
+ private IDryEqpTypeService eqpTypeService;
@Autowired
private IDryShopService dryShopService;
@@ -97,9 +102,14 @@
}
if (topic.startsWith("$SYS/brokers/") && topic.endsWith("disconnected")) {
- String clientid = messageJson.getString("clientid");
- redisUtil.del(MqttConstant.MQTT_ONLINE_CLIENT + clientid);
- System.err.println(String.format("璁惧: %s涓嬬嚎", clientid));
+ try {
+ String clientid = messageJson.getString("clientid");
+ redisUtil.del(MqttConstant.MQTT_ONLINE_CLIENT + clientid);
+ System.err.println(String.format("璁惧: %s涓嬬嚎", clientid));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
}
parseAdminCommand(topic, mqttMessage);
@@ -183,34 +193,44 @@
// 鎺ユ敹璁惧瀹炴椂鏁版嵁
case MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA:
- try {
- RealTimeDataVo vo = JSON.parseObject(message, RealTimeDataVo.class);
- realTimeDataService.realTimeDataHandle(vo);
- } catch (Exception e) {
- e.printStackTrace();
- }
+ ThreadUtil.execute(() -> {
+ try {
+ RealTimeDataVo vo = JSON.parseObject(message, RealTimeDataVo.class);
+ realTimeDataService.realTimeDataHandle(vo);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
break;
case MqttConstant.TENANT_UP_PREFIX_EQU:
- try {
- DryEquipment equipment = (DryEquipment) messageJson.get("equipment");
-
- DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(equipment.getTenantId() + "", equipment.getCode());
-
- if (dryEquipment == null) {
- equipmentService.save(equipment);
+ ThreadUtil.execute(() -> {
+ try {
+ Object equObj = messageJson.get("equipment");
+ DryEquipment equipment = JSON.parseObject(equObj.toString(), DryEquipment.class);
+ TenantContext.setTenant(equipment.getTenantId()+"");
+ DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(equipment.getTenantId() + "", equipment.getCode());
+ if (dryEquipment == null) {
+ equipmentService.save(equipment);
+ }
+ Object typeObj = messageJson.get("eqpType");
+ DryEqpType eqpType = JSON.parseObject(typeObj.toString(), DryEqpType.class);
+ DryEqpType dryEqpType = eqpTypeService.getById(eqpType.getId());
+ if (dryEqpType == null) {
+ eqpTypeService.save(eqpType);
+ }
+ // 鑾峰彇璁惧鎵�灞炶溅闂�
+ Object shopObj = messageJson.get("shop");
+ DryShop shop = JSON.parseObject(shopObj.toString(), DryShop.class);
+ DryShop dryShop = dryShopService.getById(shop.getId());
+ if (dryShop == null) {
+ dryShopService.save(shop);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
}
+ });
-
-
- // 鑾峰彇璁惧鎵�灞炶溅闂�
- DryShop shop = (DryShop) messageJson.get("shop");
- shop.setTenantId(equipment.getTenantId());
- dryShopService.save(shop);
-
- } catch (Exception e) {
- e.printStackTrace();
- }
}
}
@@ -256,28 +276,37 @@
break;
case MqttConstant.SERVICE_REQ_PREFIX:
log.debug("鏀跺埌璁惧璇︾粏淇℃伅鏌ヨ璇锋眰");
- String tenantId = messageJson.getString("tenantId");
- if (tenantId!=null && tenantId.equals(TenantContext.getTenant())) {
- // 鏍规嵁璁惧缂栫爜鏌ヨ璁惧淇℃伅
- String code = messageJson.getString("code");
- DryEquipment equipmentVo = equipmentService.selectByTenantIdEquipmentId(tenantId,code);
- // 鏍规嵁璁惧杞﹂棿id鏌ヨ杞﹂棿淇℃伅
- DryShop shop = dryShopService.getById(equipmentVo.getShopId());
+ ThreadUtil.execute(() -> {
+ String tenantId = messageJson.getString("tenantId");
+ String clientId = mqttUtil.getMqttClient().getClientId();
+ String tenant = clientId.substring(clientId.lastIndexOf("_")+1);
+ if (tenantId!=null && tenantId.equals(tenant)) {
+ TenantContext.setTenant(tenantId);
+ // 鏍规嵁璁惧缂栫爜鏌ヨ璁惧淇℃伅
+ String code = messageJson.getString("code");
+ DryEquipment equipmentVo = equipmentService.selectByTenantIdEquipmentId(tenantId,code);
+ // 鏍规嵁璁惧杞﹂棿id鏌ヨ杞﹂棿淇℃伅
+ DryShop shop = dryShopService.getById(equipmentVo.getShopId());
+ // 鏍规嵁璁惧绫诲瀷ID鏌ヨ璁惧绫诲瀷淇℃伅
+ DryEqpType eqpType = eqpTypeService.getById(equipmentVo.getType());
- JSONObject res = new JSONObject();
+ JSONObject res = new JSONObject();
- res.put("tenant", TenantContext.getTenant());
- res.put("equipment", equipmentVo);
- res.put("shop", shop);
- try {
- MqttMessage sendMessage = new MqttMessage(res.toJSONString().getBytes());
- sendMessage.setQos(0);
- mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_EQU, sendMessage);
- // baseCommonService.addLog(res.toString(), CommonConstant.LOG_TYPE_MQTT, CommonConstant.OPERATE_MQTT_2);
- } catch (Exception e) {
- e.printStackTrace();
+ res.put("tenant", tenantId);
+ res.put("equipment", equipmentVo);
+ res.put("shop", shop);
+ res.put("eqpType", eqpType);
+ try {
+ MqttMessage sendMessage = new MqttMessage(res.toJSONString().getBytes());
+ sendMessage.setQos(0);
+ mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_EQU, sendMessage);
+ // baseCommonService.addLog(res.toString(), CommonConstant.LOG_TYPE_MQTT, CommonConstant.OPERATE_MQTT_2);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
}
- }
+ });
+
break;
--
Gitblit v1.9.3