更新

src/main/java/com/zswl/dataservice/service/artemis/ArtemisListenerService.java

@@ -1,10 +1,17 @@
 package com.zswl.dataservice.service.artemis;
 
 import com.zswl.dataservice.service.base.SuperService;
+import com.zswl.dataservice.service.mqtt.GateWayInfoService;
 import com.zswl.dataservice.utils.mqtt.mqttConfig.client.MQClient;
 import com.zswl.dataservice.utils.mqtt.mqttConfig.constant.MQConstant;
+import jakarta.jms.Destination;
 import jakarta.jms.Message;
+import jakarta.jms.Session;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.jms.client.ActiveMQBytesMessage;
+import org.apache.activemq.artemis.jms.client.ActiveMQMessage;
+import org.apache.activemq.artemis.jms.client.ActiveMQTopic;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jms.annotation.JmsListener;
 import org.springframework.stereotype.Service;
@@ -23,6 +30,9 @@ public class ArtemisListenerService {
     @Autowired
     OperationMessageService operationMessageService;
 
+    @Autowired
+    GateWayInfoService gateWayInfoService;
+
     // 网关来的消息
     @JmsListener(destination = "/v1/gateway/#", containerFactory = MQConstant.TopicListenerContainerFactory)
     @JmsListener(destination = ".v1.gateway.#", containerFactory = MQConstant.TopicListenerContainerFactory)
@@ -30,4 +40,42 @@ public class ArtemisListenerService {
         operationMessageService.handlerGateWayMessage(message);
     }
 
+    // 网关来的消息
+    @JmsListener(destination = "#", containerFactory = MQConstant.TopicListenerContainerFactory)
+    public void receiveOnLineChangeMessage(Message message, Session session) {
+        try {
+            ActiveMQTopic activeMQTopic = (ActiveMQTopic) message.getJMSDestination();
+            String topicName = activeMQTopic.getTopicName();
+                if (!topicName.equals("$sys.mqtt.sessions")) {
+                return;
+            }
+            log.info("==sezz {}", session.getTransacted());
+            String messageId = message.getJMSMessageID();
+            String clientId = message.getStringProperty("_AMQ_LVQ_NAME");
+            log.info("receiveMessage {} 消息监听clientId: {}", messageId, clientId);
+            log.info("Topic: {}", topicName);
+
+            if (message instanceof ActiveMQBuffer) {
+                log.info("---------------------");
+            }
+            Destination destination = message.getJMSDestination();
+
+            log.info("---------------------AA {} ", destination.toString());
+            ActiveMQMessage activeMQBytesMessage = (ActiveMQMessage) message;
+            ActiveMQBuffer buffer = activeMQBytesMessage.getCoreMessage().getBodyBuffer();
+
+            activeMQBytesMessage.checkBuffer();
+            log.info("bb: {}", buffer.readableBytes());
+            byte[] bytes = new byte[buffer.readableBytes()];
+            buffer.readBytes(bytes);
+
+            // 处理消息，解析出会话信息
+            String sessionInfo = new String(bytes, "gbk");
+            log.info("-------------------: {} {}", sessionInfo, sessionInfo.length());
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
 }

src/main/java/com/zswl/dataservice/service/mqtt/DevicePingInfoService.java

@@ -80,7 +80,7 @@ public class DevicePingInfoService extends SuperService {
      * 检查
      */
     public void checkDeviceState() {
-        log.info("checkDeviceState: {}", DateUtils.paresTime(System.currentTimeMillis(), DateUtils.FORMAT_LONG));
+//        log.info("checkDeviceState: {}", DateUtils.paresTime(System.currentTimeMillis(), DateUtils.FORMAT_LONG));
         List<DeviceInfo> list = deviceInfoDao.findAll();
         if (ObjectUtils.isNotEmpty(list)) {
             long time = System.currentTimeMillis();
@@ -91,7 +91,6 @@ public class DevicePingInfoService extends SuperService {
                     Long firstTime = _list.get(0).getCreateTime();
                     Long lastTime = _list.get(_list.size() - 1).getCreateTime();
                     long avgTime = (firstTime - lastTime) / _list.size();
-                    log.info("avgTime: {} {}", (firstTime - lastTime), avgTime);
                     if ((time - firstTime) > avgTime) {
                         onLineState = OnLineState.OffLine;
                     }
@@ -102,8 +101,9 @@ public class DevicePingInfoService extends SuperService {
                     standardData.put("id", deviceInfo.getId());
                     standardData.put("onLineState", onLineState);
                     commonService.updateData(standardData, DeviceInfo.class.getSimpleName());
+                    deviceSyncFullCardService.noticeSyncDeviceOnlineStateChange(deviceInfo.getDeviceId());
                 }
-                deviceSyncFullCardService.noticeSyncDeviceOnlineStateChange(deviceInfo.getDeviceId());
+
             }
         }
     }

src/main/java/com/zswl/dataservice/service/mqtt/GateWayInfoService.java

@@ -91,9 +91,7 @@ public class GateWayInfoService extends SuperService {
         gateWayInfo.setOnLineState(OnLineState.OnLine);
         gateWayInfo.setLastOnlineTime(System.currentTimeMillis());
         BeanUtils.copyProperties(param, gateWayInfo, "id");
-        if (param.getState() == null) {
-            gateWayInfo.setOnLineState(OnLineState.OffLine);
-        }
+
         gateWayInfoDao.save(gateWayInfo);
         if (StringUtils.isNotEmpty(param.getProjectInfoCode())) {
             ProjectInfo projectInfo = projectInfoDao.findTopByCode(param.getProjectInfoCode());
@@ -130,12 +128,18 @@ public class GateWayInfoService extends SuperService {
         // 网关信息
         GateWayInfo gateWayInfo = gateWayInfoDao.findTopByGateWayId(param.getGateWayId());
         if (ObjectUtils.isEmpty(gateWayInfo)) {
-            return ResultContent.buildFail(String.format("网关未注册：%s", param.getGateWayId()));
+            log.error("网关未注册");
+            return ResultContent.buildFail(String.format("网关未注册，清先注册网关：%s", param.getGateWayId()));
         }
 
         String projectInfoCode = param.getProjectInfoCode();
+        if (ObjectUtils.isEmpty(projectInfoCode)) {
+            projectInfoCode = gateWayInfo.getProjectInfoCode();
+        }
+
         ProjectInfo projectInfo = projectInfoDao.findTopByCode(projectInfoCode);
         if (ObjectUtils.isEmpty(projectInfo)) {
+            log.error("分组不存在");
             return ResultContent.buildFail(String.format("分组不存在：%s", projectInfoCode));
         }
 
@@ -325,4 +329,14 @@ public class GateWayInfoService extends SuperService {
         return ResultContent.buildSuccess();
     }
 
+    /**
+     * 在线状态改变
+     *
+     * @param gateWayId
+     * @param onLineState
+     */
+    public void stateChange(String gateWayId, OnLineState onLineState) {
+
+    }
+
 }