响应应答

pole-service/src/main/java/com/zksy/pole/MQTTServer/callBack/BootNettyMqttMsgBack.java

@@ -1,5 +1,7 @@
 package com.zksy.pole.MQTTServer.callBack;
 
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson.JSONObject;
 
 import com.zksy.pole.MQTTServer.config.MQTTServerProperties;
@@ -90,68 +92,67 @@ public class BootNettyMqttMsgBack {
         mqttPublishMessage.payload().readBytes(headBytes);
         String data = new String(headBytes);
         System.out.println("收到数据-->"+data);
-		handleCmdFactory.HandleCmd(data);
+		String jsonRes = handleCmdFactory.HandleCmd(data);
 		//重置读取的指针
 		mqttPublishMessage.payload().resetReaderIndex();
 		//（qos= 0最多一次的传输，1至少一次的传输，2: 只有一次的传输）
-        switch (qos) {
-	        case AT_MOST_ONCE: 		//	至多一次
-				// 构建自定义应答消息
-				MqttFixedHeader customFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
-				MqttPublishVariableHeader customVariableHeader = new MqttPublishVariableHeader("down/light/ack/40005274", 0); // 使用0作为packetId，因为这里不需要确认
-				JSONObject jsonObject = new JSONObject();
-				jsonObject.put("flag", 1);
-				jsonObject.put("cmd", 1111);
-				byte[] customPayload = jsonObject.toString().getBytes(StandardCharsets.UTF_8);
-				ByteBuf customByteBuf = Unpooled.wrappedBuffer(customPayload);
-				MqttPublishMessage customMessage = new MqttPublishMessage(customFixedHeader, customVariableHeader, customByteBuf);
+		if(StrUtil.isNotBlank(jsonRes)){
+			switch (qos) {
+				case AT_MOST_ONCE: 		//	至多一次
+					// 构建自定义应答消息
+					MqttFixedHeader customFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
+					MqttPublishVariableHeader customVariableHeader = new MqttPublishVariableHeader("down/light/ack/40005274", 0); // 使用0作为packetId，因为这里不需要确认
+					byte[] customPayload = jsonRes.getBytes(StandardCharsets.UTF_8);
+					ByteBuf customByteBuf = Unpooled.wrappedBuffer(customPayload);
+					MqttPublishMessage customMessage = new MqttPublishMessage(customFixedHeader, customVariableHeader, customByteBuf);
 
-				// 发送自定义应答消息给网关
-				if (channel != null) {
-					channel.writeAndFlush(customMessage);
-				} else {
-					log.warn("网关通道未找到，无法发送应答消息");
-				}
-	            break;
-			// QoS 1: 发送PUBACK
-	        case AT_LEAST_ONCE:		//	至少一次
-	    		//	构建返回报文， 可变报头
-	    		MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
-	    		//	构建返回报文， 固定报头
-	    		MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
-	    		//	构建PUBACK消息体
-	    		MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
-	    		log.info("back--"+pubAck.toString());
-	    		channel.writeAndFlush(pubAck);
-				//推送到订阅的客户端
-				subscribSend(mqttMessage,channel);
-	            break;
-			// QoS 2: 发送PUBREC
-	        case EXACTLY_ONCE:		//	刚好一次
-	            //	构建返回报文， 固定报头
-	        	MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
-	            //	构建返回报文， 可变报头
-	            MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
-	            MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2,mqttMessageIdVariableHeaderBack2);
-				//服务端收到publis的QoS2的消息之后，服务端需要保存一个msgid的记录，并且进入一个状态，
-				// 即之后不管来了几个这个msgid的消息，都不管他，认为是重复的，丢弃。
-				//接收到publish的QoS2消息之后，不能马上投递给上层，而是在本地做持久化，将消息保存起来。
-				int mqttMessageId=mqttPublishMessage.variableHeader().packetId();
-				if(!MQTTServer.mqttMessageIdMap.containsKey(mqttMessageId)){
-					//不存在此消息，将此消息暂存 //todo 这里可以换成redis做缓存
-					MQTTServer.mqttMessageIdMap.put(mqttMessageId, mqttMessage);
-					log.info("消息ID"+mqttMessageId+"-->Qos2级别消息，消息缓存");
-				}else{
-					//重复发送消息，直接返回
-					log.info(mqttPublishMessage.variableHeader().packetId()+"消息重复："+mqttPublishMessage.fixedHeader().isDup());
-					return;
-				}
+					// 发送自定义应答消息给网关
+					if (channel != null) {
+						channel.writeAndFlush(customMessage);
+					} else {
+						log.warn("网关通道未找到，无法发送应答消息");
+					}
+					break;
+				// QoS 1: 发送PUBACK
+				case AT_LEAST_ONCE:		//	至少一次
+					//	构建返回报文， 可变报头
+					MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
+					//	构建返回报文， 固定报头
+					MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
+					//	构建PUBACK消息体
+					MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
+					log.info("back--"+pubAck.toString());
+					channel.writeAndFlush(pubAck);
+					//推送到订阅的客户端
+					subscribSend(mqttMessage,channel);
+					break;
+				// QoS 2: 发送PUBREC
+				case EXACTLY_ONCE:		//	刚好一次
+					//	构建返回报文， 固定报头
+					MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
+					//	构建返回报文， 可变报头
+					MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
+					MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2,mqttMessageIdVariableHeaderBack2);
+					//服务端收到publis的QoS2的消息之后，服务端需要保存一个msgid的记录，并且进入一个状态，
+					// 即之后不管来了几个这个msgid的消息，都不管他，认为是重复的，丢弃。
+					//接收到publish的QoS2消息之后，不能马上投递给上层，而是在本地做持久化，将消息保存起来。
+					int mqttMessageId=mqttPublishMessage.variableHeader().packetId();
+					if(!MQTTServer.mqttMessageIdMap.containsKey(mqttMessageId)){
+						//不存在此消息，将此消息暂存 //todo 这里可以换成redis做缓存
+						MQTTServer.mqttMessageIdMap.put(mqttMessageId, mqttMessage);
+						log.info("消息ID"+mqttMessageId+"-->Qos2级别消息，消息缓存");
+					}else{
+						//重复发送消息，直接返回
+						log.info(mqttPublishMessage.variableHeader().packetId()+"消息重复："+mqttPublishMessage.fixedHeader().isDup());
+						return;
+					}
 
-	    		channel.writeAndFlush(mqttMessageBack);
-	            break;
-			default:
-				break;
-        }
+					channel.writeAndFlush(mqttMessageBack);
+					break;
+				default:
+					break;
+			}
+		}
 	}
 
 	/**

pole-service/src/main/java/com/zksy/pole/factory/HandleCmdFactory.java

@@ -55,13 +55,13 @@ public class HandleCmdFactory {
             case 6012:
                 s = handleDgEquipmentStatusCmd.HandleCmdMethod(jsonObject.toBean(DgEquipmentStatus.class));
                 return s;
-            case 1111:
+            case 1111: //todo
                 s = handleDgGatewayRegistrationCmd.HandleCmdMethod(jsonObject.toBean(DgGatewayRegistration.class));
                 return s;
             case 6022:
                 s = handleDgGatewayStrategyCmd.HandleCmdMethod(jsonObject.toBean(DgGatewayStrategy.class));
                 return s;
-            case 6010:
+            case 6010: //todo
                 s = handleDgLampControlledHeartbeatCmd.HandleCmdMethod(jsonObject.toBean(DgLampControlledHeartbeat.class));
                 return s;
             case 6014:
@@ -70,7 +70,7 @@ public class HandleCmdFactory {
             case 6015:
                 s = handleDgRelayTimingStrategyCmd.HandleCmdMethod(jsonObject.toBean(DgRelayTimingStrategy.class));
                 return s;
-            case 2000:
+            case 2000: //todo
                 s = handleDgReportEnvironmentalDataCmd.HandleCmdMethod(jsonObject.toBean(DgReportEnvironmentalData.class));
                 return s;
             case 6011:

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgEquipmentAlarmCmd.java

@@ -1,19 +1,30 @@
 package com.zksy.pole.factory.impl;
 
+import com.zksy.common.exception.DbException;
 import com.zksy.pole.domain.po.DgEquipmentAlarm;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgEquipmentAlarmMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgEquipmentAlarmCmd implements IHandleCmd {
+
     @Autowired
     private DgEquipmentAlarmMapper dgEquipmentAlarmMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgEquipmentAlarm d = (DgEquipmentAlarm) data;
-        dgEquipmentAlarmMapper.insert(d);
-        return "";
+        try {
+            DgEquipmentAlarm d = (DgEquipmentAlarm) data;
+            dgEquipmentAlarmMapper.insert(d);
+            return null;
+        } catch (Exception e) {
+            log.error("DgEquipmentAlarm数据插入失败");
+            return null;
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgEquipmentStatusCmd.java

@@ -3,17 +3,24 @@ package com.zksy.pole.factory.impl;
 import com.zksy.pole.domain.po.DgEquipmentStatus;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgEquipmentStatusMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgEquipmentStatusCmd implements IHandleCmd {
     @Autowired
     private DgEquipmentStatusMapper dgEquipmentStatusMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgEquipmentStatus d = (DgEquipmentStatus) data;
-        dgEquipmentStatusMapper.insert(d);
-        return "";
+        try {
+            DgEquipmentStatus d = (DgEquipmentStatus) data;
+            dgEquipmentStatusMapper.insert(d);
+            return null;
+        } catch (Exception e) {
+            log.error("DgEquipmentStatus数据插入失败");
+            return null;
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgGatewayRegistrationCmd.java

@@ -2,22 +2,38 @@ package com.zksy.pole.factory.impl;
 
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.lang.UUID;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.zksy.pole.domain.po.DgGatewayRegistration;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgGatewayRegistrationMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
 public class HandleDgGatewayRegistrationCmd implements IHandleCmd {
+    private static final Logger log = LoggerFactory.getLogger(HandleDgGatewayRegistrationCmd.class);
     @Autowired
     private DgGatewayRegistrationMapper dgGatewayRegistrationMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgGatewayRegistration d = (DgGatewayRegistration) data;
-        d.setId(UUID.fastUUID().toString(true));
 
-        dgGatewayRegistrationMapper.insert(d);
-        return "";
+        try {
+            DgGatewayRegistration d = (DgGatewayRegistration) data;
+            d.setId(UUID.fastUUID().toString(true));
+            dgGatewayRegistrationMapper.insert(d);
+            JSONObject obj = JSONUtil.createObj();
+            obj.set("cmd",1111);
+            obj.set("flag",1);
+            return obj.toString();
+        } catch (Exception e) {
+            log.error("DgGatewayRegistration数据插入失败");
+            JSONObject obj = JSONUtil.createObj();
+            obj.set("cmd",1111);
+            obj.set("flag",0);
+            return obj.toString();
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgGatewayStrategyCmd.java

@@ -3,17 +3,24 @@ package com.zksy.pole.factory.impl;
 import com.zksy.pole.domain.po.DgGatewayStrategy;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgGatewayStrategyMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgGatewayStrategyCmd implements IHandleCmd {
     @Autowired
     private DgGatewayStrategyMapper dgGatewayStrategyMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgGatewayStrategy d = (DgGatewayStrategy) data;
-        dgGatewayStrategyMapper.insert(d);
-        return "";
+        try {
+            DgGatewayStrategy d = (DgGatewayStrategy) data;
+            dgGatewayStrategyMapper.insert(d);
+            return null;
+        } catch (Exception e) {
+            log.error("DgGatewayStrategy数据插入失败");
+            return null;
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgLampControlledHeartbeatCmd.java

@@ -1,21 +1,34 @@
 package com.zksy.pole.factory.impl;
 
+import cn.hutool.json.JSON;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.zksy.pole.domain.po.DgGatewayStrategy;
 import com.zksy.pole.domain.po.DgLampControlledHeartbeat;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgGatewayStrategyMapper;
 import com.zksy.pole.mapper.DgLampControlledHeartbeatMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgLampControlledHeartbeatCmd implements IHandleCmd {
     @Autowired
     private DgLampControlledHeartbeatMapper dgLampControlledHeartbeatMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgLampControlledHeartbeat d = (DgLampControlledHeartbeat) data;
-        dgLampControlledHeartbeatMapper.insert(d);
-        return "";
+        JSONObject obj = JSONUtil.createObj();
+        obj.set("cmd",6010);
+        obj.set("heard_id",1010);
+        try {
+            DgLampControlledHeartbeat d = (DgLampControlledHeartbeat) data;
+            dgLampControlledHeartbeatMapper.insert(d);
+            return obj.toString();
+        } catch (Exception e) {
+            log.error("DgLampControlledHeartbeat数据插入失败");
+            return obj.toString();
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgRelayControlCmd.java

@@ -5,17 +5,24 @@ import com.zksy.pole.domain.po.DgRelayControl;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgLampControlledHeartbeatMapper;
 import com.zksy.pole.mapper.DgRelayControlMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgRelayControlCmd implements IHandleCmd {
     @Autowired
     private DgRelayControlMapper dgRelayControlMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgRelayControl d = (DgRelayControl) data;
-        dgRelayControlMapper.insert(d);
-        return "";
+        try {
+            DgRelayControl d = (DgRelayControl) data;
+            dgRelayControlMapper.insert(d);
+            return null;
+        } catch (Exception e) {
+            log.error("DgRelayControl数据插入失败");
+            return null;
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgRelayTimingStrategyCmd.java

@@ -5,17 +5,24 @@ import com.zksy.pole.domain.po.DgRelayTimingStrategy;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgRelayControlMapper;
 import com.zksy.pole.mapper.DgRelayTimingStrategyMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgRelayTimingStrategyCmd implements IHandleCmd {
     @Autowired
     private DgRelayTimingStrategyMapper dgRelayTimingStrategyMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgRelayTimingStrategy d = (DgRelayTimingStrategy) data;
-        dgRelayTimingStrategyMapper.insert(d);
-        return "";
+        try {
+            DgRelayTimingStrategy d = (DgRelayTimingStrategy) data;
+            dgRelayTimingStrategyMapper.insert(d);
+            return null;
+        } catch (Exception e) {
+            log.error("DgRelayTimingStrategy数据插入失败");
+            return null;
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgReportEnvironmentalDataCmd.java

@@ -1,5 +1,7 @@
 package com.zksy.pole.factory.impl;
 
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.zksy.pole.domain.po.DgRelayControl;
 import com.zksy.pole.domain.po.DgReportEnvironmentalData;
 import com.zksy.pole.factory.IHandleCmd;
@@ -14,8 +16,15 @@ public class HandleDgReportEnvironmentalDataCmd implements IHandleCmd {
     private DgReportEnvironmentalDataMapper dgReportEnvironmentalDataMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgReportEnvironmentalData d = (DgReportEnvironmentalData) data;
-        dgReportEnvironmentalDataMapper.insert(d);
-        return "";
+        JSONObject obj = JSONUtil.createObj();
+        obj.set("cmd",2000);
+        obj.set("sersor_id",1);
+        try {
+            DgReportEnvironmentalData d = (DgReportEnvironmentalData) data;
+            dgReportEnvironmentalDataMapper.insert(d);
+            return obj.toString();
+        } catch (Exception e) {
+            return obj.toString();
+        }
     }
 }

pole-service/src/main/java/com/zksy/pole/factory/impl/HandleDgTimelyControlCmd.java

@@ -5,17 +5,24 @@ import com.zksy.pole.domain.po.DgTimelyControl;
 import com.zksy.pole.factory.IHandleCmd;
 import com.zksy.pole.mapper.DgRelayControlMapper;
 import com.zksy.pole.mapper.DgTimelyControlMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class HandleDgTimelyControlCmd implements IHandleCmd {
     @Autowired
     private DgTimelyControlMapper dgTimelyControlMapper;
     @Override
     public <T> String HandleCmdMethod(T data) {
-        DgTimelyControl d = (DgTimelyControl) data;
-        dgTimelyControlMapper.insert(d);
-        return "";
+        try {
+            DgTimelyControl d = (DgTimelyControl) data;
+            dgTimelyControlMapper.insert(d);
+            return null;
+        } catch (Exception e) {
+            log.error("DgTimelyControl数据插入失败");
+            return null;
+        }
     }
 }