#update 地图统一推送功能

maibu-common/src/main/java/com/maibu/config/DeviceTask.java

@@ -102,6 +102,11 @@ public class DeviceTask {
         return builder(FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR);
     }
 
+    @Bean(FastBeeConstant.TASK.MAP_MESSAGE_PUSH)
+    public Executor mapMessagePush() {
+        return builder(FastBeeConstant.TASK.MAP_MESSAGE_PUSH);
+    }
+
     @Bean(FastBeeConstant.TASK.DEVICE_TASK_HANDLER)
     public Executor deviceTaskHandler() {
         return builder(FastBeeConstant.TASK.DEVICE_TASK_HANDLER);

maibu-common/src/main/java/com/maibu/constant/FastBeeConstant.java

@@ -105,6 +105,13 @@ public interface FastBeeConstant {
          * 设备错误
          */
         String DEVICE_HEARTBEAT_MONITOR = "deviceHeartBeatMonitor";
+
+        /**
+         * 设备错误
+         */
+        String MAP_MESSAGE_PUSH = "mapMessagePush";
+
+
         /**
          * 设备任务处理线程
          */

maibu-common/src/main/java/com/maibu/core/business/dto/DeviceMapPushDTO.java

@@ -0,0 +1,18 @@
+package com.maibu.core.business.dto;
+
+import lombok.Data;
+
+@Data
+public class DeviceMapPushDTO {
+    private String id;
+    private double lat;
+    private double lng;
+    private double y; //yaw
+    private double f; //fix_status
+    private double a; //altitude
+    private String s;//status 运行状态
+    private Long t; //taskId 任务id
+    private double b; //电量
+    private boolean ha = false;//hasAlarm 有无告警
+
+}

maibu-common/src/main/java/com/maibu/core/business/dto/WebMapMessageDTO.java

@@ -0,0 +1,12 @@
+package com.maibu.core.business.dto;
+
+import com.maibu.core.enums.MesType;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class WebMapMessageDTO {
+    private List<DeviceMapPushDTO> data;
+    private MesType type;
+}

maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterace.java

@@ -1,7 +0,0 @@
-package com.maibu.core.business.inter;
-
-public interface DeviceWebSocketInterace {
-
-    void sendWebsoketMes(String key, String data);
-
-}

maibu-common/src/main/java/com/maibu/core/business/inter/DeviceWebSocketInterface.java

@@ -0,0 +1,10 @@
+package com.maibu.core.business.inter;
+
+public interface DeviceWebSocketInterface {
+
+    void sendWebsocketMes(String key, String data);
+
+
+    void sendMapWebsocketMes(Long siteId,String data);
+
+}

maibu-common/src/main/java/com/maibu/core/business/inter/WebsocketMesDispather.java

@@ -14,10 +14,19 @@ public class WebsocketMesDispather {
     private ApplicationContext applicationContext;
 
     public void dispather(String key, String data) {
-        Map<String, DeviceWebSocketInterace> beanMap = applicationContext.getBeansOfType(DeviceWebSocketInterace.class);
+        Map<String, DeviceWebSocketInterface> beanMap = applicationContext.getBeansOfType(DeviceWebSocketInterface.class);
         // 遍历
         beanMap.values().forEach(x -> {
-            x.sendWebsoketMes(key, data);
+            x.sendWebsocketMes(key, data);
+        });
+    }
+
+
+    public void mapMessageDispather(Long siteId, String data) {
+        Map<String, DeviceWebSocketInterface> beanMap = applicationContext.getBeansOfType(DeviceWebSocketInterface.class);
+        // 遍历
+        beanMap.values().forEach(x -> {
+            x.sendMapWebsocketMes(siteId, data);
         });
     }
 

maibu-common/src/main/java/com/maibu/core/enums/MesType.java

@@ -15,5 +15,6 @@ public enum MesType {
     detectionReport, // 障碍物消息
     switchPermission, // 切换控制申请
     switchResult,// 切换结果
-    route_post;// 路径规划 点位推送
+    route_post,// 路径规划 点位推送
+    mapPush;
 }

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

@@ -7,6 +7,7 @@ import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
 import com.maibu.core.business.*;
+import com.maibu.core.business.dto.DeviceMapPushDTO;
 import com.maibu.mqtt.CustomMqttDeviceMonitor;
 import org.apache.commons.lang3.StringUtils;
 import org.eclipse.paho.client.mqttv3.MqttClient;
@@ -56,6 +57,17 @@ public class SiteMemory {
     // 记录 设备上线后的运行数据 下线后清除并保存到数据库
     public static ConcurrentHashMap<String, DeviceRunStatistics> deviceRunStatisticsMap = new ConcurrentHashMap<>();
 
+    public static ConcurrentHashMap<String, DeviceMapPushDTO> deviceMapPushMap = new ConcurrentHashMap<>();
+
+    public Map<String, DeviceMapPushDTO> getAllDeviceMapPush() {
+        return deviceMapPushMap;
+    }
+
+    public void saveMaPushDTO(String deviceId, DeviceMapPushDTO dto) {
+        deviceMapPushMap.put(deviceId, dto);
+    }
+
+
     public void saveRunStatistics(DeviceRunStatistics deviceRunStatistics) {
         if (deviceRunStatistics != null && !StringUtils.isEmpty(deviceRunStatistics.getDeviceId())) {
             deviceRunStatisticsMap.put(deviceRunStatistics.getDeviceId(), deviceRunStatistics);

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 com.maibu.core.business.dto.DeviceMapPushDTO;
 import org.springframework.util.CollectionUtils;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -160,7 +161,7 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
     }
 
     public WebStatusMessageDTO createWebDeviceLocationMessage(HostLocationRelTimeDTO locationRelTimeDTO,
-            String deviceId) {
+                                                              String deviceId) {
         WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO();
         List<DeviceStatusDetail> transferData = new ArrayList<>();
         DeviceStatusDetail lat = new DeviceStatusDetail();
@@ -217,6 +218,27 @@ public class HostMessageHandler implements CustomMqttMessageHandler {
 
         webStatusMessageDTO.setData(transferData);
         webStatusMessageDTO.setType(MesType.locationInfo);
+
+        if (nettyDevice != null && nettyDevice.getDevice() != null) {
+            Long siteId = nettyDevice.getDevice().getSiteId();
+            Long orgId = nettyDevice.getDevice().getOrgId();
+            SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
+            if (siteMemory != null) {
+                DeviceMapPushDTO dto = new DeviceMapPushDTO();
+                dto.setA(Double.parseDouble(altitude.getValue()));
+                dto.setId(deviceId);
+//                dto.setB();
+//                dto.setHa();
+                dto.setF(Double.parseDouble(fix_status.getValue()));
+                dto.setS(status.getValue());
+                dto.setLat(Double.parseDouble(lat.getValue()));
+                dto.setLng(Double.parseDouble(lng.getValue()));
+                dto.setT(nettyDevice.getCurrentTaskId());
+                dto.setY(Double.parseDouble(yaw.getValue()));
+                siteMemory.saveMaPushDTO(deviceId, dto);
+            }
+        }
+
         return webStatusMessageDTO;
     }
 

maibu-netty-server/src/main/java/com/maibu/init/InitThread.java

@@ -42,6 +42,9 @@ public class InitThread implements ApplicationRunner {
     @Resource(name = FastBeeConstant.TASK.DEVICE_HEARTBEAT_MONITOR)
     private Executor deviceHeartBeatMonitorExecutor;
 
+    @Resource(name = FastBeeConstant.TASK.MAP_MESSAGE_PUSH)
+    private Executor mapMessagePushExecutor;
+
     @Resource(name = FastBeeConstant.TASK.COMMON_DEVICE_MONITOR)
     private Executor commonDeviceMonitorExecutor;
 
@@ -127,6 +130,11 @@ public class InitThread implements ApplicationRunner {
 
         GlobalMemory.initMqttDeviceMonitor(mqttHostUrl, mqttClientId, mqttUsername, mqttPassword);
 
+        // 地图监控推送
+        mapMessagePushExecutor.execute(() -> {
+            deviceThreadService.mapMessagePush();
+        });
+
         // 上位机设备心跳监控
         deviceHeartBeatMonitorExecutor.execute(() -> {
             deviceThreadService.heartBeatMonitor();

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

@@ -229,9 +229,9 @@ public class DevicePlanTaskMonitorService {
                                             String deviceId = devicePlanTask.getDeviceId();
                                             if (!StringUtils.isEmpty(deviceId) && deviceId.equals(key)) {
                                                 devicePlanTask.setTaskStaus(DeviceTaskStaus.EXECUTING);
-                                                siteMemory.addDevicePlanExecute(devicePlanTask);
                                                 device.setCurrentTaskId(devicePlanTask.getId());
                                                 device.setTask(devicePlanTask);
+                                                siteMemory.addDevicePlanExecute(devicePlanTask);
                                                 siteMemory.removeDevicePlanPrepare(devicePlanTask);
                                             }
                                         }

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

@@ -14,6 +14,10 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import com.maibu.core.business.dto.DeviceMapPushDTO;
+import com.maibu.core.business.dto.WebMapMessageDTO;
+import com.maibu.core.enums.*;
+import com.maibu.utils.json.JsonUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.checkerframework.checker.units.qual.s;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -35,11 +39,6 @@ import com.maibu.core.business.ErrorIdentificationStandard;
 import com.maibu.core.business.alarm_center.AlarmMessage;
 import com.maibu.core.business.device.NettyDevice;
 import com.maibu.core.business.inter.WebsocketMesDispather;
-import com.maibu.core.enums.CompareEnum;
-import com.maibu.core.enums.ConnectorStatus;
-import com.maibu.core.enums.DeviceStatus;
-import com.maibu.core.enums.ErrorSource;
-import com.maibu.core.enums.RespondCode;
 import com.maibu.core.host.HeartBeatDTO;
 import com.maibu.core.host.LocationMessage;
 import com.maibu.core.redis.RedisCache;
@@ -224,7 +223,7 @@ public class DeviceThreadService {
     }
 
     public List<AlarmMessage> compareErrorStandard(Long siteId, Long orgId, DeviceRunningStatusHistory history,
-            List<ErrorIdentificationStandard> standards) {
+                                                   List<ErrorIdentificationStandard> standards) {
         if (history == null || CollectionUtils.isEmpty(standards))
             return null;
         List<AlarmMessage> list = new ArrayList<>();
@@ -397,4 +396,36 @@ public class DeviceThreadService {
         }, 0, 500, TimeUnit.MILLISECONDS);
     }
 
+
+    /**
+     * 监控心跳
+     */
+    public void mapMessagePush() {
+        executor.scheduleWithFixedDelay(() -> {
+            try {
+                List<SiteMemory> list = GlobalMemory.getAllSiteMemory();
+                if (!CollectionUtils.isEmpty(list)) {
+                    list.forEach(x -> {
+                        Map<String, DeviceMapPushDTO> map = x.getAllDeviceMapPush();
+                        if (!CollectionUtils.isEmpty(map)) {
+                            List<DeviceMapPushDTO> pushList = new ArrayList<>(map.values());
+                            WebMapMessageDTO dto = new WebMapMessageDTO();
+                            dto.setData(pushList);
+                            dto.setType(MesType.mapPush);
+                            try {
+                                websocketMesDispather.mapMessageDispather(x.getSiteId(), JsonUtils.toJsonString(pushList));
+                            } catch (Exception e) {
+                                log.error("实时地图数据websocket推送失败: siteId={}, error={}", x.getSiteId(),
+                                        e.getMessage(), e);
+                            }
+
+                        }
+                    });
+                }
+            } catch (Exception e) {
+                log.error("执行监测心跳线程错误：{}", e.getMessage());
+            }
+        }, 0, 100, TimeUnit.MILLISECONDS);
+    }
+
 }

maibu-web-middleware/src/main/java/com/maibu/dto/WebMessageBaseDTO.java

@@ -8,6 +8,8 @@ import lombok.Data;
 public class WebMessageBaseDTO {
     private MesType type;
     private String userName;
+    private Long siteId;
+    private Long orgId;
     private String deviceId;
     private String token;
     private String message;

maibu-web-middleware/src/main/java/com/maibu/memory/MiddleGlobalMemory.java

@@ -9,8 +9,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
-import java.util.Map;
+import java.util.*;
-import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
 @Component
@@ -25,6 +24,8 @@ public class MiddleGlobalMemory {
      */
     public static Vertx vertx;
 
+    //TODO 推送全局的时候需要 判断是不是单个的控制
+    public static ConcurrentHashMap<Long, Map<String, WebSocket>> mapPushMap = new ConcurrentHashMap<>();
     // 存储所有在线WebSocket连接
     // key: test:web:token
     public static final ConcurrentHashMap<String, WebSocket> onlineSockets = new ConcurrentHashMap<>();
@@ -84,6 +85,7 @@ public class MiddleGlobalMemory {
                 .findFirst().orElse(null);
     }
 
+
     public static String findNettyClient(NettyClient nettyClient) {
         return nettyClientMap.entrySet().stream()
                 .filter(entry -> Objects.equals(entry.getValue(), nettyClient))
@@ -91,5 +93,38 @@ public class MiddleGlobalMemory {
                 .findFirst().orElse(null);
     }
 
+    public static void addMapSocket(Long siteId, String token, WebSocket webSocket) {
+        mapPushMap
+                .computeIfAbsent(siteId, k -> new ConcurrentHashMap<>())
+                .put(token, webSocket);
+    }
+
+    public static void removeByWebSocket(WebSocket webSocket) {
+        if (webSocket == null) {
+            return;
+        }
+
+        mapPushMap.forEach((siteId, socketMap) -> {
+            socketMap.entrySet().removeIf(entry ->
+                    entry.getValue() == webSocket
+            );
+
+            // siteId 下没有连接了，删除整个 siteId
+            if (socketMap.isEmpty()) {
+                mapPushMap.remove(siteId, socketMap);
+            }
+        });
+    }
+
+    public static List<WebSocket> getMapSockets(Long siteId) {
+        Map<String, WebSocket> map = mapPushMap.get(siteId);
+
+        if (map == null || map.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        return new ArrayList<>(map.values());
+    }
+
 
 }

maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketHandler.java

@@ -64,6 +64,7 @@ public class WebsocketHandler extends WebSocketServer {
             releaseResource(key);
             logger.info("客户端断开连接: remoteAddress={}, code={}, reason={}, 当前在线数: {}", conn.getRemoteSocketAddress(), code, reason, MiddleGlobalMemory.onlineSockets.size());
         }
+        MiddleGlobalMemory.removeByWebSocket(conn);
     }
 
     @Override
@@ -77,22 +78,15 @@ public class WebsocketHandler extends WebSocketServer {
                 String deviceId = webMessageBaseDTO.getDeviceId();
                 MesType type = webMessageBaseDTO.getType();
                 String key = userName + ":web" + ":" + token;
-                if (MesType.AUTH.equals(type)) {
+                if (MesType.mapPush.equals(type)) {
+                    Long siteId = webMessageBaseDTO.getSiteId();
+                    MiddleGlobalMemory.addMapSocket(siteId, token, conn);
+                } else if (MesType.AUTH.equals(type)) {
                     if (StringUtils.isEmpty(deviceId)) {
                         return;
                     }
-                    //todo 先关闭原先的
-//                    WebSocket oldWebsocket = MiddleGlobalMemory.onlineSockets.get(key);
                     NettyClient client = MiddleGlobalMemory.nettyClientMap.get(key);
-
-//                    if (oldWebsocket != null && oldWebsocket.isOpen()) {
-//                        oldWebsocket.close();
-//                    }
-//                    if (oldNettyClient != null && oldNettyClient.isConnected()) {
-//                        oldNettyClient.disconnect();
-//                    }
                     MiddleGlobalMemory.onlineSockets.put(key, conn);
-                    //这里可以根据需要初始化Netty客户端
                     if (client == null) {
                         client = new NettyClient();
                     }

maibu-web-middleware/src/main/java/com/maibu/websocket/WebsocketMesHandler.java

@@ -1,16 +1,19 @@
 package com.maibu.websocket;
 
-import com.maibu.core.business.inter.DeviceWebSocketInterace;
+import com.maibu.core.business.inter.DeviceWebSocketInterface;
 import com.maibu.memory.MiddleGlobalMemory;
 
 import org.java_websocket.WebSocket;
 import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
 
 @Component
-public class WebsocketMesHandler implements DeviceWebSocketInterace {
+public class WebsocketMesHandler implements DeviceWebSocketInterface {
 
     @Override
-    public void sendWebsoketMes(String key, String data) {
+    public void sendWebsocketMes(String key, String data) {
         MiddleGlobalMemory.onlineSockets.keySet().forEach(x -> {
             // key: test:web:token
             if (x.contains(key)) {
@@ -21,4 +24,14 @@ public class WebsocketMesHandler implements DeviceWebSocketInterace {
 
     }
 
+    @Override
+    public void sendMapWebsocketMes(Long siteId, String data) {
+        List<WebSocket> webSocketList = MiddleGlobalMemory.getMapSockets(siteId);
+        if (!CollectionUtils.isEmpty(webSocketList)) {
+            webSocketList.forEach(webSocket -> {
+                WebsocketHandler.sendMessageToClient(webSocket, data);
+            });
+        }
+    }
+
 }