webSocket

business-service/src/main/java/com/ozs/service/mapper/BaseRailwayManagementMapper.java

@@ -3,6 +3,7 @@ package com.ozs.service.mapper;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.ozs.service.entity.BaseRailwayManagement;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
 
 /**
  * <p>
@@ -15,4 +16,5 @@ import org.apache.ibatis.annotations.Mapper;
 @Mapper
 public interface BaseRailwayManagementMapper extends BaseMapper<BaseRailwayManagement> {
 
+    String getCameraNameByCameraCode(@Param("cameraCode") String cameraCode);
 }

business-service/src/main/java/com/ozs/service/service/BaseRailwayManagementService.java

@@ -2,6 +2,7 @@ package com.ozs.service.service;
 
 import com.baomidou.mybatisplus.extension.service.IService;
 import com.ozs.service.entity.BaseRailwayManagement;
+import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
 
@@ -21,4 +22,6 @@ public interface BaseRailwayManagementService extends IService<BaseRailwayManage
      * @return
      */
     String importBaseLineManagement(List<BaseRailwayManagement> accountManageList, boolean updateSupport);
+
+    String getCameraNameByCameraCode(String cameraCode);
 }

business-service/src/main/java/com/ozs/service/service/impl/BaseRailwayManagementServiceImpl.java

@@ -83,4 +83,9 @@ public class BaseRailwayManagementServiceImpl extends ServiceImpl<BaseRailwayMan
         }
         return successMsg.toString();
     }
+
+    @Override
+    public String getCameraNameByCameraCode(String cameraCode) {
+        return baseRailwayManagementMapper.getCameraNameByCameraCode(cameraCode);
+    }
 }

business-service/src/main/resources/mapper/service/BaseRailwayManagementMapper.xml

@@ -2,4 +2,7 @@
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.ozs.service.mapper.BaseRailwayManagementMapper">
 
+    <select id="getCameraNameByCameraCode" resultType="java.lang.String" parameterType="java.lang.String">
+        select railway_name railwayName from base_railway_management where railway_code = #{cameraCode}
+    </select>
 </mapper>

hazard-admin/src/main/java/com/ozs/web/controller/accountmanagment/MsgAlarmController.java

@@ -1,6 +1,7 @@
 package com.ozs.web.controller.accountmanagment;
 
 
+import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -14,10 +15,8 @@ import com.ozs.service.entity.MsgAlarm;
 import com.ozs.service.entity.MsgAlarmFrequency;
 import com.ozs.service.entity.vo.MsgAlarmResp;
 import com.ozs.service.entity.vo.MsgAlarmVo;
-import com.ozs.service.service.BaseCameraManagementService;
-import com.ozs.service.service.BaseRailwayManagementService;
-import com.ozs.service.service.MsgAlarmFrequencyService;
-import com.ozs.service.service.MsgAlarmService;
+import com.ozs.service.service.*;
+import com.ozs.web.controller.websocket.WebSocketServer;
 import io.swagger.annotations.ApiOperation;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,10 +31,7 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Date;
-import java.util.List;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -58,27 +54,43 @@ public class MsgAlarmController extends BaseController {
     @Autowired
     BaseRailwayManagementService baseRailwayManagementService;
     @Value("${base.imgUrl:http://124.71.171.71:18801/picbucket}")
-    private String imgUrl; 
+    private String imgUrl;
+    @Autowired
+    private BaseUserService baseUserService;
+    @Autowired
+    private WebSocketServer webSocketServer;
 
     /**
      * 测试消息推送
      */
     @PostMapping("/push")
     @ApiOperation(value = "预报警信息分页")
-    public AjaxResult insertDate(String[] args) {
-        // 模拟需要推送的用户群
-        ArrayList<String> ids = new ArrayList<>();
-        ids.add("001");
-        ids.add("002");
-        ids.add("003");
-        ids.add("004");
-        ids.add("005");
-//
-//        WebSocketService webSocketService = new WebSocketService();
-//        webSocketService.sendMsgToUsers(ids);
+    public AjaxResult insertDate(String args) {
+        String userId = getUserId();
+        // 初始化 -> 运行中 -> 成功或失败
+        try {
+            // 1、初始化
+            webSocketServer.sendOneMessage(userId, objStr(args, 1));
+            Thread.sleep(3 * 1000);
+            // 2、运行中
+            webSocketServer.sendOneMessage(userId, objStr(args, 2));
+            Thread.sleep(5 * 1000);
+            // 3、成功或失败
+            webSocketServer.sendOneMessage(userId, Math.random() * 10 > 5 ? objStr(args, 6) : objStr(args, 7));
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+            throw new RuntimeException("构建失败");
+        }
         return AjaxResult.success("推送成功");
     }
 
+    private String objStr(String str, int status) {
+        JSONObject obj = new JSONObject();
+        obj.put("pipelineId", str); //流水线ID
+        obj.put("status", status);
+        return obj.toJSONString();
+    }
+
     /**
      * 报警信息分页
      *

hazard-admin/src/main/java/com/ozs/web/controller/websocket/WebSocketConfig.java

@@ -1,21 +1,21 @@
-//package com.ozs.web.controller.websocket;
-//
-//import org.springframework.context.annotation.Bean;
-//import org.springframework.context.annotation.Configuration;
-//import org.springframework.web.socket.server.standard.ServerEndpointExporter;
-//
-///**
-// * @Author : sunhh
-// * @create 2023/3/7 16:17
-// */
-//@Configuration
-//public class WebSocketConfig {
-//    /**
-//     * 	注入ServerEndpointExporter，
-//     * 	这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
-//     */
-//    @Bean
-//    public ServerEndpointExporter serverEndpointExporter() {
-//        return new ServerEndpointExporter();
-//    }
-//}
+package com.ozs.web.controller.websocket;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+/**
+ * @Author : sunhh
+ * @create 2023/3/7 16:17
+ */
+@Configuration
+public class WebSocketConfig {
+    /**
+     * 	注入ServerEndpointExporter，
+     * 	这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
+     */
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
+}

hazard-admin/src/main/java/com/ozs/web/controller/websocket/WebSocketServer.java

@@ -1,129 +1,129 @@
-//package com.ozs.web.controller.websocket;
-//
-//import lombok.extern.slf4j.Slf4j;
-//import org.springframework.stereotype.Component;
-//
-//import javax.websocket.*;
-//import javax.websocket.server.PathParam;
-//import javax.websocket.server.ServerEndpoint;
-//import java.util.HashMap;
-//import java.util.List;
-//import java.util.Map;
-//import java.util.concurrent.ConcurrentHashMap;
-//import java.util.concurrent.CopyOnWriteArraySet;
-//
-///**
-// * @Author : sunhh
-// * @create 2023/3/7 16:15
-// */
-//@Slf4j
-//@Component
-//@ServerEndpoint("/websocket/{userId}")  // 接口路径 ws://localhost:8087/webSocket/userId;
-//public class WebSocketServer {
-//
-//    //与某个客户端的连接会话，需要通过它来给客户端发送数据
-//    private Session session;
-//    /**
-//     * 用户ID
-//     */
-//    private String userId;
-//
-//    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
-//    //虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。
-//    //  注：底下WebSocket是当前类名
-//    private static CopyOnWriteArraySet<WebSocketServer> webSockets =new CopyOnWriteArraySet<>();
-//    // 用来存在线连接用户信息
-//    private static ConcurrentHashMap<String,Session> sessionPool = new ConcurrentHashMap<String,Session>();
-//
-//    /**
-//     * 链接成功调用的方法
-//     */
-//    @OnOpen
-//    public void onOpen(Session session, @PathParam(value="userId")String userId) {
-//        try {
-//            this.session = session;
-//            this.userId = userId;
-//            webSockets.add(this);
-//            sessionPool.put(userId, session);
-//            log.info("【websocket消息】有新的连接，总数为:"+webSockets.size());
-//        } catch (Exception e) {
-//        }
-//    }
-//
-//    /**
-//     * 链接关闭调用的方法
-//     */
-//    @OnClose
-//    public void onClose() {
-//        try {
-//            webSockets.remove(this);
-//            sessionPool.remove(this.userId);
-//            log.info("【websocket消息】连接断开，总数为:"+webSockets.size());
-//        } catch (Exception e) {
-//        }
-//    }
-//    /**
-//     * 收到客户端消息后调用的方法
-//     *
-//     * @param message
-//     */
-//    @OnMessage
-//    public void onMessage(String message) {
-//        log.info("【websocket消息】收到客户端消息:"+message);
-//    }
-//
-//    /** 发送错误时的处理
-//     * @param session
-//     * @param error
-//     */
-//    @OnError
-//    public void onError(Session session, Throwable error) {
-//
-//        log.error("用户错误,原因:"+error.getMessage());
-//        error.printStackTrace();
-//    }
-//
-//
-//    // 此为广播消息
-//    public void sendAllMessage(String message) {
-//        log.info("【websocket消息】广播消息:"+message);
-//        for(WebSocketServer webSocket : webSockets) {
-//            try {
-//                if(webSocket.session.isOpen()) {
-//                    webSocket.session.getAsyncRemote().sendText(message);
-//                }
-//            } catch (Exception e) {
-//                e.printStackTrace();
-//            }
-//        }
-//    }
-//
-//    // 此为单点消息
-//    public void sendOneMessage(String userId, String message) {
-//        Session session = sessionPool.get(userId);
-//        if (session != null&&session.isOpen()) {
-//            try {
-//                log.info("【websocket消息】 单点消息:"+message);
-//                session.getAsyncRemote().sendText(message);
-//            } catch (Exception e) {
-//                e.printStackTrace();
-//            }
-//        }
-//    }
-//
-//    // 此为单点消息(多人)
-//    public void sendMoreMessage(String[] userIds, String message) {
-//        for(String userId:userIds) {
-//            Session session = sessionPool.get(userId);
-//            if (session != null&&session.isOpen()) {
-//                try {
-//                    log.info("【websocket消息】 单点消息:"+message);
-//                    session.getAsyncRemote().sendText(message);
-//                } catch (Exception e) {
-//                    e.printStackTrace();
-//                }
-//            }
-//        }
-//
-//    }
-//}
+package com.ozs.web.controller.websocket;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.PathParam;
+import javax.websocket.server.ServerEndpoint;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+/**
+ * @Author : sunhh
+ * @create 2023/3/7 16:15
+ */
+@Slf4j
+@Component
+@ServerEndpoint("/websocket/{userId}")  // 接口路径 ws://localhost:8087/webSocket/userId;
+public class WebSocketServer {
+
+    //与某个客户端的连接会话，需要通过它来给客户端发送数据
+    private Session session;
+    /**
+     * 用户ID
+     */
+    private String userId;
+
+    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
+    //虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。
+    //  注：底下WebSocket是当前类名
+    private static CopyOnWriteArraySet<WebSocketServer> webSockets =new CopyOnWriteArraySet<>();
+    // 用来存在线连接用户信息
+    private static ConcurrentHashMap<String,Session> sessionPool = new ConcurrentHashMap<String,Session>();
+
+    /**
+     * 链接成功调用的方法
+     */
+    @OnOpen
+    public void onOpen(Session session, @PathParam(value="userId")String userId) {
+        try {
+            this.session = session;
+            this.userId = userId;
+            webSockets.add(this);
+            sessionPool.put(userId, session);
+            log.info("【websocket消息】有新的连接，总数为:"+webSockets.size());
+        } catch (Exception e) {
+        }
+    }
+
+    /**
+     * 链接关闭调用的方法
+     */
+    @OnClose
+    public void onClose() {
+        try {
+            webSockets.remove(this);
+            sessionPool.remove(this.userId);
+            log.info("【websocket消息】连接断开，总数为:"+webSockets.size());
+        } catch (Exception e) {
+        }
+    }
+    /**
+     * 收到客户端消息后调用的方法
+     *
+     * @param message
+     */
+    @OnMessage
+    public void onMessage(String message) {
+        log.info("【websocket消息】收到客户端消息:"+message);
+    }
+
+    /** 发送错误时的处理
+     * @param session
+     * @param error
+     */
+    @OnError
+    public void onError(Session session, Throwable error) {
+
+        log.error("用户错误,原因:"+error.getMessage());
+        error.printStackTrace();
+    }
+
+
+    // 此为广播消息
+    public void sendAllMessage(String message) {
+        log.info("【websocket消息】广播消息:"+message);
+        for(WebSocketServer webSocket : webSockets) {
+            try {
+                if(webSocket.session.isOpen()) {
+                    webSocket.session.getAsyncRemote().sendText(message);
+                }
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    // 此为单点消息
+    public void sendOneMessage(String userId, String message) {
+        Session session = sessionPool.get(userId);
+        if (session != null&&session.isOpen()) {
+            try {
+                log.info("【websocket消息】 单点消息:"+message);
+                session.getAsyncRemote().sendText(message);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    // 此为单点消息(多人)
+    public void sendMoreMessage(String[] userIds, String message) {
+        for(String userId:userIds) {
+            Session session = sessionPool.get(userId);
+            if (session != null&&session.isOpen()) {
+                try {
+                    log.info("【websocket消息】 单点消息:"+message);
+                    session.getAsyncRemote().sendText(message);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+
+    }
+}

hazard-sdk/src/main/java/com/ozs/controller/upload/GeoHazardMonitorTokenController.java

@@ -7,6 +7,7 @@ import com.ozs.common.core.domain.Result;
 import com.ozs.common.core.domain.entity.SysUser;
 import com.ozs.common.utils.ApiTokenUtils;
 import com.ozs.common.utils.AppTokenUtils;
+import com.ozs.common.utils.AppendUtils;
 import com.ozs.common.utils.StringUtils;
 import com.ozs.common.utils.stateSecrets.SM4Utils;
 import com.ozs.service.entity.BaseCameraManagement;
@@ -14,13 +15,9 @@ import com.ozs.service.entity.BaseDeviceDynamicManagement;
 import com.ozs.service.entity.MsgAlarm;
 import com.ozs.service.entity.MsgAlarmFrequency;
 import com.ozs.service.entity.SvcAddress;
+import com.ozs.service.entity.vo.MsgAlarmResp;
 import com.ozs.service.mapper.BaseUserMapper;
-import com.ozs.service.service.BaseCameraManagementService;
-import com.ozs.service.service.BaseDeviceDynamicManagementService;
-import com.ozs.service.service.BaseUserService;
-import com.ozs.service.service.MsgAlarmFrequencyService;
-import com.ozs.service.service.MsgAlarmService;
-import com.ozs.service.service.SvcAddressService;
+import com.ozs.service.service.*;
 import com.ozs.vo.ReqDeviceVo;
 import com.ozs.vo.ReqMsgAlarmVo;
 import com.ozs.vo.RespGeoHazardMonitorVo;
@@ -65,6 +62,8 @@ public class GeoHazardMonitorTokenController {
     private WebSocketServer webSocketServer;
     @Autowired
     private BaseUserService baseUserService;
+    @Resource
+    private BaseRailwayManagementService baseRailwayManagementService;
 
     /**
      * 获取web访问令牌
@@ -137,17 +136,27 @@ public class GeoHazardMonitorTokenController {
             Set<String> set = baseUserService.getUserIdList(baseCameraManagement.getCreateBy(), baseCameraManagement.getDeptId());
             // List<String> list = new ArrayList<>(set);
             String[] list = set.toArray(new String[set.size()]);
+            MsgAlarmResp msgAlarmResp = new MsgAlarmResp();
+            String begin = AppendUtils.stringAppend(baseCameraManagement.getBeginMile());
+            String end = AppendUtils.stringAppend(baseCameraManagement.getEndMile());
+            String endMile = begin + "-" + end;
+            msgAlarmResp.setBeginMile(endMile);
+            String mile = AppendUtils.stringAppend(baseCameraManagement.getInstallMile());
+            msgAlarmResp.setAlarmMile(mile);
+            String cameraCode = baseCameraManagement.getCameraCode();
+            String cameraName = baseRailwayManagementService.getCameraNameByCameraCode(cameraCode);
+            msgAlarmResp.setRailwayName(cameraName);
             // 推送消息reqMsgAlarmVo
             // 初始化 -> 运行中 -> 成功或失败
             try {
                 // 1、初始化
-                webSocketServer.sendMoreMessage(list, objStr(baseCameraManagement, 1));
+                webSocketServer.sendMoreMessage(list, objStr(msgAlarmResp, 1));
                 Thread.sleep(3 * 1000);
                 // 2、运行中
-                webSocketServer.sendMoreMessage(list, objStr(baseCameraManagement, 2));
+                webSocketServer.sendMoreMessage(list, objStr(msgAlarmResp, 2));
                 Thread.sleep(5 * 1000);
                 // 3、成功或失败
-                webSocketServer.sendMoreMessage(list, Math.random() * 10 > 5 ? objStr(baseCameraManagement, 6) : objStr(baseCameraManagement, 7));
+                webSocketServer.sendMoreMessage(list, Math.random() * 10 > 5 ? objStr(msgAlarmResp, 6) : objStr(msgAlarmResp, 7));
             } catch (InterruptedException e) {
                 e.printStackTrace();
                 throw new RuntimeException("构建失败");
@@ -215,9 +224,9 @@ public class GeoHazardMonitorTokenController {
         }
     }
 
-    private String objStr(BaseCameraManagement baseCameraManagement, int status) {
+    private String objStr(MsgAlarmResp msgAlarmResp, int status) {
         JSONObject obj = new JSONObject();
-        obj.put("pipelineId", baseCameraManagement); //流水线ID
+        obj.put("pipelineId", msgAlarmResp); //流水线ID
         obj.put("status", status);
         return obj.toJSONString();
     }