fix(equip):换流站添加消息转发iot，存储influxdb，兼容绍兴换流站消息

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -358,18 +358,24 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
         JSONObject jsonObject = JSONObject.parseObject(message);
         String dataType = jsonObject.getString("datatype");
-        String indexAddress, value, timeStamp, quality = null;
+        String indexAddress = null, value, timeStamp, quality = null;
 
         //如果消息是遥信类型，进行指标转换
-        if (dataType != null && dataType.equals("state")) {
+        assert dataType != null;
+        if (dataType.equals("state")) {
             indexAddress = jsonObject.getString("scadaid");
             value = jsonObject.getInteger("value") == 1 ? "true" : "false";
             timeStamp = jsonObject.getString("timestamp");
-        } else {
+        }else if (dataType.equals("analog")){
             indexAddress = jsonObject.getString("key");
             value = jsonObject.getString("value");
             timeStamp = jsonObject.getString("time_stamp");
             quality = jsonObject.getString("quality");
+        }else {
+            indexAddress = jsonObject.getString("key");
+            value = jsonObject.getFloat("value") == 0.0 ? "false" : "true";
+            timeStamp = jsonObject.getString("time_stamp");
+            quality = jsonObject.getString("quality");
         }
         Map<Object, Object> equipmentIndexKeyMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS_KEY);
         if (equipmentIndexKeyMap.get(indexAddress) != null) {
@@ -389,7 +395,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             iotDataVO.setValue(value);
             iotDatalist.add(iotDataVO);
 
-//        iEquipmentSpecificSerivce.getEquipSpecificDetailsByEquipmentId(equipmentSpeIndex.getEquipmentSpecificId());
             QueryWrapper<EquipmentSpecific> queryWrapper = new QueryWrapper<>();
             queryWrapper.eq("id", equipmentSpeIndex.getEquipmentSpecificId());
             EquipmentSpecific equipmentSpecific = iEquipmentSpecificSerivce.getOne(queryWrapper);
@@ -397,6 +402,13 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 return;
             }
             String iotCode = equipmentSpecific.getIotCode();
+            StringBuilder endIndex = new StringBuilder(iotCode).insert(8, '/');
+            String iotTopic = "influxdb/" + endIndex;
+            if (isSendIot) {
+                JSONObject msg = new JSONObject();
+                msg.put(equipmentSpeIndex.getEquipmentIndexKey(), value);
+                mqttSendGateway.sendToMqtt(iotTopic, JSON.toJSONString(msg));
+            }
 
             List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);