合并cloud分支代码

bsw215583320

2024-11-08 3b64c08b7defbe7ff1ba4c6a7a4644ddedb5b782

 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,36 @@

        // 接收设备实时数据
      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");
        ThreadUtil.execute(() -> {
          try {
            DryEquipment equipment = (DryEquipment) messageJson.get("equipment");

          DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(equipment.getTenantId() + "", equipment.getCode());
            DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(equipment.getTenantId() + "", equipment.getCode());
            if (dryEquipment == null) {
              equipmentService.save(equipment);
            }

          if (dryEquipment == null) {
            equipmentService.save(equipment);
            // 获取设备所属车间
            DryShop shop = (DryShop) messageJson.get("shop");
            shop.setTenantId(equipment.getTenantId());
            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();
        }
    }

  }
@@ -254,27 +266,39 @@


        break;
      case MqttConstant.SERVICE_REQ_EQU_TOPIC:
      case MqttConstant.SERVICE_REQ_PREFIX:
        log.debug("收到设备详细信息查询请求");
        // 根据设备编码查询设备信息
        String code = messageJson.getString("code");
        DryEquipment equipmentVo = equipmentService.selectByTenantIdEquipmentId(TenantContext.getTenant(),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)) {
            // 根据设备编码查询设备信息
            String code = messageJson.getString("code");
            DryEquipment equipmentVo = equipmentService.selectByTenantIdEquipmentId(tenantId,code);
            // 根据设备车间id查询车间信息
            DryShop shop = dryShopService.selectByTenantIdShopId(tenantId, equipmentVo.getShopId());
            // 根据设备类型ID查询设备类型信息
            DryEqpType eqpType = eqpTypeService.selectByTenantIdTypeId(tenantId, 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", clientId);
            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;