#接受路径规划接口日志

maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java

@@ -5,8 +5,12 @@ import com.maibu.core.business.device.Connector;
 import com.maibu.core.business.device.NettyDevice;
 import com.maibu.core.enums.ConnectorStatus;
 import com.maibu.core.enums.ConnectorType;
+import com.maibu.utils.json.JsonUtils;
+
 import io.netty.channel.Channel;
 import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,6 +27,7 @@ import java.util.stream.Collectors;
  */
 @Component
 @Data
+@Slf4j
 public class DeviceSessionManager {
 
     private static final Logger logger = LoggerFactory.getLogger(DeviceSessionManager.class);
@@ -187,7 +192,7 @@ public class DeviceSessionManager {
     public List<NettyDevice> getSlaveControl(String slaveId) {
         // 获取所有连接的主机 且是active状态的
         return deviceMap.values().stream()
-                .filter(device -> ObjectUtil.equals(device.getCurrentSlaveId(), slaveId)
+                .filter(device -> Objects.equals(device.getCurrentSlaveId(), slaveId)
                         && ConnectorStatus.ACTIVE.equals(device.getConnectorStatus()))
                 .collect(Collectors.toList());
     }
@@ -195,14 +200,16 @@ public class DeviceSessionManager {
     public List<NettyDevice> getAllSlaveControl(String slaveId) {
         // 获取所有连接的主机 不区分状态
         return deviceMap.values().stream()
-                .filter(device -> ObjectUtil.equals(device.getCurrentSlaveId(), slaveId))
+                .filter(device -> !StringUtils.isEmpty(device.getCurrentSlaveId())
+                        && Objects.equals(device.getCurrentSlaveId(), slaveId)
+                        && device.getDeviceType().equals(ConnectorType.MASTER))
                 .collect(Collectors.toList());
     }
 
     public NettyDevice getCurrentControl(String slaveId) {
         // 获取所有连接的主机 不区分状态
         return deviceMap.values().stream()
-                .filter(device -> ObjectUtil.equals(device.getCurrentSlaveId(), slaveId)
+                .filter(device -> Objects.equals(device.getCurrentSlaveId(), slaveId)
                         && ConnectorStatus.ACTIVE.equals(device.getConnectorStatus()))
                 .findFirst().orElse(null);
     }
@@ -312,8 +319,11 @@ public class DeviceSessionManager {
         if (taskId == null) {
             return null;
         }
+        log.debug("sendMessage deviceMap：{},taskId={}", JsonUtils.toJsonString(deviceMap), taskId);
         for (String key : deviceMap.keySet()) {
             NettyDevice nettyDevice = deviceMap.get(key);
+            log.debug("sendMessage currentTaskId:{}，equal {}", JsonUtils.toJsonString(nettyDevice.getCurrentTaskId()),
+                    Objects.equals(nettyDevice.getCurrentTaskId(), taskId));
             if (nettyDevice != null && Objects.equals(nettyDevice.getCurrentTaskId(), taskId)) {
                 return nettyDevice.getConnectorId();
             }

maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java

@@ -3,6 +3,7 @@ package com.maibu.mqtt;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.util.CollectionUtils;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -354,6 +355,14 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
                         }
                     }
                 }
+                //todo 推送实时位置和状态消息到外部
+                try {
+                    GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId),
+                            JsonUtils.toJsonString(sendDto));
+                } catch (MqttException e) {
+                    // TODO Auto-generated catch block
+                    e.printStackTrace();
+                }
 
             }
             if ("cancelled".equals(dto.getStatus())) {

maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java

@@ -285,9 +285,17 @@ public class DeviceThreadService {
                 List<HeartBeatDTO> heartBeatDTOS = GlobalMemory.customMqttDeviceMonitor.getAllHeartBeat();
                 if (!CollectionUtils.isEmpty(heartBeatDTOS)) {
                     heartBeatDTOS.forEach(x -> {
+                        if (x == null) {
+                            return;
+                        }
                         String deviceId = x.getSn();
                         Long lastActiveTime = x.getSysTimestamp();
                         NettyDevice nettyDevice = sessionManager.getDevice(x.getSn());
+
+                        if (lastActiveTime == null) {
+                            log.error("设备心跳时间为空，deviceId={}", deviceId);
+                            return;
+                        }
                         if (nowTime - lastActiveTime >= Constant.hostOfflineInterval) {
                             // 超时了判断离线
                             if (nettyDevice != null) {
@@ -367,23 +375,11 @@ public class DeviceThreadService {
                                     device.setHasHost(true);
                                     deviceService.updateDeviceBySelf(device);
                                 }
-                                // 推送json 找到设备绑定的主机进行推送
+                                NettyDevice slaveDevice = sessionManager.getDevice(deviceId);
-                                // List<NettyDevice> controlMasters =
+                                if (slaveDevice != null) {
-                                // sessionManager.getAllSlaveControl(device.getSerialNumber());
+                                    slaveDevice.setOnlineStatus(1);
-                                // if (!CollectionUtils.isEmpty(controlMasters)) {
+                                    slaveDevice.setDevice(device);
-                                // controlMasters.forEach(x -> {
+                                }
-                                // DeviceStatusChangeDTO changeDTO = new DeviceStatusChangeDTO();
-                                // changeDTO.setDeviceId(data);
-                                // changeDTO.setStatus(DeviceStatus.online.getCode());
-                                // changeDTO.setEvent(RespondCode.device_status_changed);
-                                // ByteBuf buf = ctx.alloc().buffer();
-                                // CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction);
-                                // String content = buf.toString(CharsetUtil.UTF_8);
-                                // logger.info("推送下位机登录状态消息:{}", content);
-                                // if (x.getChannel() != null && x.getChannel().isActive()) {
-                                // x.getChannel().writeAndFlush(buf);
-                                // }
-                                // });
                             } else {
                                 // todo 推送设备上线
                                 nettyDevice.status(1);
@@ -392,7 +388,7 @@ public class DeviceThreadService {
                     });
                 }
             } catch (Exception e) {
-                log.error("执行监测心跳线程错误：{}", e.getMessage());
+                log.error("执行监测心跳线程错误", e);
             }
         }, 0, 500, TimeUnit.MILLISECONDS);
     }