feat: 增加websocket通知

pom.xml

@@ -35,6 +35,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-validation</artifactId>

src/main/java/com/xydl/cac/config/WebSocketConfig.java

@@ -0,0 +1,20 @@
+package com.xydl.cac.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+@Configuration
+public class WebSocketConfig {
+    /**
+     * ServerEndpointExporter 作用
+     *
+     * 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint
+     *
+     * @return ServerEndpointExporter
+     */
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
+}

src/main/java/com/xydl/cac/controller/I2syncController.java

@@ -120,7 +120,7 @@ public class I2syncController extends BasicController {
         String xml = JSONUtil.object2Xml(request);
         Response<String> resp = Response.success(xml);
         if (StringUtils.isBlank(node.getType())) {
-            resp.setWarnMsg("改类型还未配置对应的监测类型编码");
+            resp.setWarnMsg("该类型还未配置对应的监测类型编码");
         }
         return resp;
     }

src/main/java/com/xydl/cac/socket/WebSocketServer.java

@@ -0,0 +1,54 @@
+package com.xydl.cac.socket;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.ServerEndpoint;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+@ServerEndpoint(value = "/websocket")
+@Component
+@Slf4j
+public class WebSocketServer {
+
+    private static List<Session> sessionPools = new ArrayList<>();
+
+    @OnOpen
+    public void onOpen(Session session) {
+        sessionPools.add(session);
+        log.info("当前连接数" + sessionPools.size());
+    }
+
+    @OnClose
+    public void onClose(Session session) {
+        sessionPools.remove(session);
+        log.info("剩余连接数" + sessionPools.size());
+    }
+
+    public void sendMessage(String message) {
+        Iterator<Session> it = sessionPools.listIterator();
+        while (it.hasNext()) {
+            Session session = it.next();
+            try {
+                session.getBasicRemote().sendText(message);
+            } catch (IOException ignore) {
+            }
+        }
+    }
+
+    //收到客户端信息
+    @OnMessage
+    public void onMessage(String message) {
+//        log.info("收到客户端信息:{}", message);
+    }
+
+    //错误时调用
+    @OnError
+    public void onError(Session session, Throwable throwable) {
+    }
+
+}

src/main/java/com/xydl/cac/task/RuleCheckTask.java

@@ -7,6 +7,7 @@ import com.xydl.cac.entity.Warning;
 import com.xydl.cac.repository.WarnRuleRepository;
 import com.xydl.cac.repository.WarningRepository;
 import com.xydl.cac.service.DataService;
+import com.xydl.cac.socket.WebSocketServer;
 import com.xydl.cac.util.DateUtil;
 import com.xydl.cac.util.DingTalkPushUtil;
 import lombok.extern.slf4j.Slf4j;
@@ -31,6 +32,8 @@ public class RuleCheckTask {
     WarningRepository warningRepository;
     @Resource
     DingTalkPushUtil dingTalkPushUtil;
+    @Resource
+    WebSocketServer webSocketServer;
 
     @Scheduled(initialDelay = 90 * 1000, fixedDelay = 3 * 60 * 1000)
     private void checkAll() {
@@ -94,6 +97,7 @@ public class RuleCheckTask {
         String str = warning.getZsbName() + "-" + warning.getWarnDesc()
                 + ", 采集时间:" + dateStr + ", 当前值:" + value + " " + warning.getTriggerDesc();
         log.warn("触发规则告警: " + str);
+        webSocketServer.sendMessage(str);
         dingTalkPushUtil.pushText(str);
     }