调整Mqtt消息到达后获取设备实现类逻辑

src/main/java/com/yunfeiyun/agmp/iots/core/manager/ConnectionManager.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 import com.yunfeiyun.agmp.common.utils.StringUtils;
 import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceDictEnum;
+import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceTypeLv1Enum;
 import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
 import com.yunfeiyun.agmp.iot.common.domain.IotDeviceconn;
 import com.yunfeiyun.agmp.iot.common.model.mq.IotDeviceEditMqModel;
@@ -309,6 +310,7 @@ public class ConnectionManager {
             MqttTopicValue mqttTopicValue = new MqttTopicValue();
             mqttTopicValue.setDevCode(iotDevice.getDevCode());
             mqttTopicValue.setDevId(iotDevice.getDevBid());
+            mqttTopicValue.setServiceName(IotDeviceDictEnum.findServiceNameByDevTypeBid(iotDevice.getDevtypeBid()));
             mqttTopicValue.setTopic(s);
             mqttTopicValues.add(mqttTopicValue);
         }
@@ -336,6 +338,7 @@ public class ConnectionManager {
         for (String s : topics) {
             MqttTopicValue mqttTopicValue = new MqttTopicValue();
             mqttTopicValue.setDevCode(iotDevice.getDevCode());
+            mqttTopicValue.setServiceName(IotDeviceDictEnum.findServiceNameByDevTypeBid(iotDevice.getDevtypeBid()));
             mqttTopicValue.setDevId(iotDevice.getDevBid());
             mqttTopicValue.setTopic(s);
             mqttTopicValues.add(mqttTopicValue);

src/main/java/com/yunfeiyun/agmp/iots/core/manager/MqttManager.java

@@ -6,6 +6,8 @@ import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 import com.yunfeiyun.agmp.common.constant.ErrorCode;
 import com.yunfeiyun.agmp.iot.common.constant.IotErrorCode;
+import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceDictEnum;
+import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceTypeLv1Enum;
 import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
 import com.yunfeiyun.agmp.iot.common.domain.IotDeviceconn;
 import com.yunfeiyun.agmp.iot.common.enums.IotDeviceconnTypeEnum;
@@ -98,6 +100,7 @@ public class MqttManager {
                     for (String s : topics) {
                         MqttTopicValue mqttTopicValue = new MqttTopicValue();
                         mqttTopicValue.setDevCode(iotDevice.getDevCode());
+                        mqttTopicValue.setServiceName(IotDeviceDictEnum.findServiceNameByDevTypeBid(iotDevice.getDevtypeBid()));
                         mqttTopicValue.setDevId(iotDevice.getDevBid());
                         mqttTopicValue.setTopic(s);
                         mqttTopicValues.add(mqttTopicValue);

src/main/java/com/yunfeiyun/agmp/iots/core/mqtt/modal/MqttTopicValue.java

@@ -7,4 +7,5 @@ public class MqttTopicValue {
     private String devCode;
     private String devId;
     private String topic;
+    private String serviceName;
 }

src/main/java/com/yunfeiyun/agmp/iots/core/mqtt/network/MqttCore.java

@@ -1,6 +1,7 @@
 package com.yunfeiyun.agmp.iots.core.mqtt.network;
 
 import com.yunfeiyun.agmp.common.utils.ip.IpUtils;
+import com.yunfeiyun.agmp.common.utils.spring.SpringUtils;
 import com.yunfeiyun.agmp.iots.core.mqtt.modal.MqttTopicValue;
 import com.yunfeiyun.agmp.iots.device.common.Device;
 import lombok.Getter;
@@ -48,6 +49,7 @@ public class MqttCore {
      * topic -> device
      */
     Map<String,String> topicToDevId=new HashMap<>();
+    Map<String,MqttTopicValue> topicValueMap = new HashMap<>();
 
     public String getConnectionId() {
         return connectionId;
@@ -71,8 +73,8 @@ public class MqttCore {
      *
      * @return
      */
-    public Device getDevice() {
-        return mqttConfig.getDevice();
+    public Device getDevice(String topic) {
+       return SpringUtils.getBean(topicValueMap.get(topic).getServiceName());
     }
 
     public String getServiceType() {
@@ -253,7 +255,9 @@ public class MqttCore {
     public void bindTopicToDeviceId(List<MqttTopicValue> mqttTopicValues){
         for(MqttTopicValue mqttTopicValue:mqttTopicValues){
             topicToDevId.put(mqttTopicValue.getTopic(),mqttTopicValue.getDevId());
+            topicValueMap.put(mqttTopicValue.getTopic(),mqttTopicValue);
         }
+
     }
 //    public void unBindTopicToDeviceId(List<MqttTopicValue> mqttTopicValues){
 //        for(MqttTopicValue mqttTopicValue:mqttTopicValues){

src/main/java/com/yunfeiyun/agmp/iots/core/mqtt/network/MqttSubscriber.java

@@ -51,7 +51,7 @@ public class MqttSubscriber implements MqttCallbackExtended {
                 return;
             }
 
-            Device device = mqttCore.getDevice();
+            Device device = mqttCore.getDevice(topic);
             if (null == device) {
                 // 当收到mqtt消息无对应的设备类型时，则丢弃消息。
                 IotException.UNKNOWN_DEVICE.throwException();