update ws module

manager/src/main/java/cn/reghao/devops/manager/app/service/bd/impl/AppStatImpl.java

@@ -6,7 +6,7 @@ import cn.reghao.devops.common.msg.event.EvtAppStat;
 import cn.reghao.devops.manager.app.db.repository.config.AppConfigRepository;
 import cn.reghao.devops.manager.app.model.po.config.AppConfig;
 import cn.reghao.devops.manager.app.service.bd.AppStat;
-import cn.reghao.devops.manager.event.MessageSenders;
+import cn.reghao.devops.manager.ws.MessageSenders;
 import cn.reghao.devops.manager.machine.service.MachineService;
 import cn.reghao.devops.manager.rbac.model.constant.RoleType;
 import cn.reghao.devops.manager.rbac.model.po.UserAuthority;

manager/src/main/java/cn/reghao/devops/manager/app/service/bd/impl/DeployAppImpl.java

@@ -10,7 +10,7 @@ import cn.reghao.devops.manager.app.model.po.config.AppDeployConfig;
 import cn.reghao.devops.manager.app.model.po.log.BuildLog;
 import cn.reghao.devops.manager.app.service.bd.DeployApp;
 import cn.reghao.devops.manager.app.service.bd.DeployStat;
-import cn.reghao.devops.manager.event.MessageSenders;
+import cn.reghao.devops.manager.ws.MessageSenders;
 import cn.reghao.devops.manager.machine.service.MachineService;
 import cn.reghao.jutil.jdk.event.message.EventMessage;
 import cn.reghao.jutil.jdk.result.Result;

manager/src/main/java/cn/reghao/devops/manager/event/ws/WsConnection.java

@@ -1,53 +0,0 @@
-package cn.reghao.devops.manager.event.ws;
-
-import cn.reghao.jutil.jdk.serializer.JsonConverter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-import org.springframework.web.socket.TextMessage;
-import org.springframework.web.socket.WebSocketSession;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @author reghao
- * @date 2022-09-01 10:44:06
- */
-@Slf4j
-@Component
-public class WsConnection {
-    private final Map<String, WebSocketSession> userSessions = new HashMap<>();
-    private final Map<String, String> sessionUsers = new HashMap<>();
-
-    public void addUserSession(String userId, WebSocketSession webSocketSession) {
-        String sessionId = webSocketSession.getId();
-        userSessions.put(userId, webSocketSession);
-        sessionUsers.put(sessionId, userId);
-    }
-
-    public void removeUserSession(WebSocketSession webSocketSession) {
-        String sessionId = webSocketSession.getId();
-        String userId = sessionUsers.get(sessionId);
-        userSessions.remove(userId);
-        sessionUsers.remove(userId);
-    }
-
-    public String getUserId(String sessionId) {
-        return sessionUsers.get(sessionId);
-    }
-
-    /*public void sendTextMsg(PushMsg pushMsg) throws IOException {
-        String userId = pushMsg.getReceiverId();
-        EventMsg eventMsg = pushMsg.getEventMsg();
-
-        WebSocketSession webSocketSession = userSessions.get(userId);
-        if (webSocketSession == null) {
-            log.error("{} 不在线", userId);
-            return;
-        }
-
-        TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(eventMsg));
-        webSocketSession.sendMessage(textMessage);
-    }*/
-}

manager/src/main/java/cn/reghao/devops/manager/event/ws/handler/LogHandler.java

@@ -1,108 +0,0 @@
-package cn.reghao.devops.manager.event.ws.handler;
-
-import cn.reghao.devops.manager.event.ws.WsConnection;
-import cn.reghao.devops.manager.log.service.LogService;
-import cn.reghao.jutil.jdk.result.AppLog;
-import cn.reghao.jutil.jdk.serializer.JdkSerializer;
-import cn.reghao.jutil.jdk.serializer.JsonConverter;
-import cn.reghao.jutil.tool.jwt.Jwt;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-import org.springframework.web.socket.*;
-
-import java.io.IOException;
-
-/**
- * @author reghao
- * @date 2022-03-11 16:45:52
- */
-@Slf4j
-@Component
-public class LogHandler implements WebSocketHandler {
-    private final WsConnection wsConnection;
-    private LogService logService;
-    private WebSocketSession pullSession;
-
-    public LogHandler(WsConnection wsConnection, LogService logService) {
-        this.wsConnection = wsConnection;
-        this.logService = logService;
-    }
-
-    @Override
-    public void afterConnectionEstablished(WebSocketSession webSocketSession) throws IOException {
-        String path = webSocketSession.getUri().getPath();
-        if (path.equals("/ws/log/pull")) {
-            this.pullSession = webSocketSession;
-        }
-
-        String query = webSocketSession.getUri().getQuery();
-        /*String jwtToken = query.replace("token=", "");
-        String userId = getUserId(jwtToken);
-        if (userId == null) {
-            webSocketSession.close(CloseStatus.NO_STATUS_CODE);
-            return;
-        }*/
-        String userId = "10001";
-        wsConnection.addUserSession(userId, webSocketSession);
-        log.info("WebSocket 建立连接");
-    }
-
-    private String getUserId(String jwtToken) {
-        if (jwtToken == null) {
-            return null;
-        }
-
-        return Jwt.parse(jwtToken, "").getUserId();
-    }
-
-    @Override
-    public void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage)
-            throws IOException {
-        try {
-            String sessionId = webSocketSession.getId();
-            if (webSocketMessage instanceof TextMessage) {
-                String payload = (String) webSocketMessage.getPayload();
-                AppLog appLog = JsonConverter.jsonToObject(payload, AppLog.class);
-                wsConnection.getUserId("");
-            } else if (webSocketMessage instanceof BinaryMessage) {
-                log.info("接收到 WebSocket 二进制消息");
-                BinaryMessage binaryMessage = (BinaryMessage) webSocketMessage;
-                Object object = JdkSerializer.deserialize(binaryMessage.getPayload().array());
-                if (object instanceof AppLog) {
-                    AppLog appLog = (AppLog) object;
-                    //logService.saveAppLog(appLog);
-                    if (pullSession != null) {
-                        String jsonData = JsonConverter.objectToJson(appLog);
-                        WebSocketMessage<String> message1 = new TextMessage(jsonData);
-                        pullSession.sendMessage(message1);
-                    }
-                }
-            } else if (webSocketMessage instanceof PingMessage) {
-                log.info("接收到 WebSocket PingMessage");
-            } else if (webSocketMessage instanceof PongMessage) {
-                log.info("接收到 WebSocket PongMessage");
-            } else {
-                log.error("接收到未知类型的 WebSocket 消息");
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
-
-    @Override
-    public void handleTransportError(WebSocketSession webSocketSession, Throwable throwable) {
-        log.error("WebSocket 数据传输错误");
-        wsConnection.removeUserSession(webSocketSession);
-    }
-
-    @Override
-    public void afterConnectionClosed(WebSocketSession webSocketSession, CloseStatus closeStatus) {
-        log.info("WebSocket 断开连接");
-        wsConnection.removeUserSession(webSocketSession);
-    }
-
-    @Override
-    public boolean supportsPartialMessages() {
-        return false;
-    }
-}

manager/src/main/java/cn/reghao/devops/manager/event/EventCenter.java → manager/src/main/java/cn/reghao/devops/manager/ws/EventCenter.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event;
+package cn.reghao.devops.manager.ws;
 
 import cn.reghao.jutil.jdk.event.message.Event;
 import cn.reghao.jutil.jdk.event.router.EventDispatcher;

manager/src/main/java/cn/reghao/devops/manager/event/EventDispatcherConfig.java → manager/src/main/java/cn/reghao/devops/manager/ws/EventDispatcherConfig.java

@@ -1,8 +1,11 @@
-package cn.reghao.devops.manager.event;
+package cn.reghao.devops.manager.ws;
 
 import cn.reghao.devops.common.msg.event.*;
 import cn.reghao.devops.manager.app.service.bd.DeployStat;
-import cn.reghao.devops.manager.event.handler.*;
+import cn.reghao.devops.manager.ws.event.EvtAgentHeartbeatHandler;
+import cn.reghao.devops.manager.ws.event.EvtAgentStartHandler;
+import cn.reghao.devops.manager.ws.event.EvtAppStatResultHandler;
+import cn.reghao.devops.manager.ws.event.MachineEventService;
 import cn.reghao.jutil.jdk.event.router.EventDispatcher;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;

manager/src/main/java/cn/reghao/devops/manager/ws/MessageDispatcher.java

@@ -0,0 +1,79 @@
+package cn.reghao.devops.manager.ws;
+
+import cn.reghao.devops.manager.ws.model.EventType;
+import cn.reghao.devops.manager.ws.model.resp.EventMessageResp;
+import cn.reghao.jutil.jdk.serializer.JsonConverter;
+import com.google.gson.JsonObject;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author reghao
+ * @date 2022-04-21 10:32:37
+ */
+@Slf4j
+@Component
+public class MessageDispatcher {
+    private final Map<String, WebSocketSession> userSessions = new HashMap<>();
+    private final Map<String, String> sessionUsers = new HashMap<>();
+
+    public void addUserSession(String userId, WebSocketSession webSocketSession) {
+        String sessionId = webSocketSession.getId();
+        userSessions.put(userId, webSocketSession);
+        sessionUsers.put(sessionId, userId);
+    }
+
+    public void removeUserSession(WebSocketSession webSocketSession) {
+        String sessionId = webSocketSession.getId();
+        String userId = sessionUsers.get(sessionId);
+        userSessions.remove(userId);
+        sessionUsers.remove(userId);
+    }
+
+    public String getUserId(String sessionId) {
+        return sessionUsers.get(sessionId);
+    }
+
+    public void dispatch(WebSocketSession session, String payload) throws IOException {
+        JsonObject jsonObject = JsonConverter.jsonToJsonElement(payload).getAsJsonObject();
+        String event = jsonObject.get("event").getAsString();
+        String payload1 = jsonObject.get("payload").getAsString();
+        switch (EventType.valueOf(event)) {
+            case heartbeat:
+                EventMessageResp<String> resp = new EventMessageResp<>(EventType.heartbeat, "pong");
+                sendMessage(session, resp);
+                break;
+            case event_error:
+                log.info("error event");
+                break;
+            case event_content:
+                log.info(payload1);
+                EventMessageResp<String> resp1 = new EventMessageResp<>(EventType.event_content, "hahahahaha");
+                sendMessage(session, resp1);
+                break;
+            case event_sync:
+
+                break;
+            default:
+        }
+    }
+
+    public void sendMessage(String userId, Object payload) throws IOException {
+        WebSocketSession session = userSessions.get(userId);
+        if (session != null) {
+            TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(payload));
+            session.sendMessage(textMessage);
+        }
+    }
+
+    public void sendMessage(WebSocketSession session, Object payload) throws IOException {
+        TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(payload));
+        session.sendMessage(textMessage);
+    }
+}

manager/src/main/java/cn/reghao/devops/manager/event/MessageSenders.java → manager/src/main/java/cn/reghao/devops/manager/ws/MessageSenders.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event;
+package cn.reghao.devops.manager.ws;
 
 import cn.reghao.devops.common.msg.MessageSender;
 import cn.reghao.devops.common.msg.constant.Protocol;

manager/src/main/java/cn/reghao/devops/manager/event/ws/WsSender.java → manager/src/main/java/cn/reghao/devops/manager/ws/WsSender.java

@@ -1,8 +1,7 @@
-package cn.reghao.devops.manager.event.ws;
+package cn.reghao.devops.manager.ws;
 
 import cn.reghao.devops.common.msg.MessageSender;
 import cn.reghao.jutil.jdk.serializer.JdkSerializer;
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.BinaryMessage;
 import org.springframework.web.socket.WebSocketSession;
@@ -15,9 +14,7 @@ import java.util.Map;
  * @author reghao
  * @date 2023-03-01 11:03:31
  */
-@Slf4j
 @Component
-@Deprecated
 public class WsSender implements MessageSender {
     private final static Map<String, WebSocketSession> sessionMap = new HashMap<>();
 

manager/src/main/java/cn/reghao/devops/manager/event/ws/WebSocketConfig.java → manager/src/main/java/cn/reghao/devops/manager/ws/config/WebSocketConfig.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.ws;
+package cn.reghao.devops.manager.ws.config;
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.web.socket.WebSocketHandler;
@@ -17,13 +17,16 @@ public class WebSocketConfig implements WebSocketConfigurer {
     private final WebSocketHandler sshWebSocketHandler;
     private final WebSocketHandler agentWebSocketHandler;
     private final WebSocketHandler logHandler;
+    private final WebSocketHandler messageHandler;
 
     public WebSocketConfig(WebSocketInterceptor webSocketInterceptor, WebSocketHandler sshWebSocketHandler, 
-                           WebSocketHandler agentWebSocketHandler, WebSocketHandler logHandler) {
+                           WebSocketHandler agentWebSocketHandler, WebSocketHandler logHandler,
+                           WebSocketHandler messageHandler) {
         this.webSocketInterceptor = webSocketInterceptor;
         this.sshWebSocketHandler = sshWebSocketHandler;
         this.agentWebSocketHandler = agentWebSocketHandler;
         this.logHandler = logHandler;
+        this.messageHandler = messageHandler;
     }
 
     @Override
@@ -43,5 +46,9 @@ public class WebSocketConfig implements WebSocketConfigurer {
         registry.addHandler(logHandler, "/ws/log/pull")
                 .addInterceptors(webSocketInterceptor)
                 .setAllowedOrigins("*");
+
+        registry.addHandler(messageHandler, "/ws/message")
+                .addInterceptors(webSocketInterceptor)
+                .setAllowedOrigins("*");
     }
 }

manager/src/main/java/cn/reghao/devops/manager/event/ws/WebSocketInterceptor.java → manager/src/main/java/cn/reghao/devops/manager/ws/config/WebSocketInterceptor.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.ws;
+package cn.reghao.devops.manager.ws.config;
 
 import cn.reghao.devops.manager.machine.util.Keys;
 import org.springframework.http.server.ServerHttpRequest;

manager/src/main/java/cn/reghao/devops/manager/event/handler/EvtAgentHeartbeatHandler.java → manager/src/main/java/cn/reghao/devops/manager/ws/event/EvtAgentHeartbeatHandler.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.handler;
+package cn.reghao.devops.manager.ws.event;
 
 import cn.reghao.jutil.jdk.event.message.Event;
 import cn.reghao.jutil.jdk.event.handler.Handler;

manager/src/main/java/cn/reghao/devops/manager/event/handler/EvtAgentStartHandler.java → manager/src/main/java/cn/reghao/devops/manager/ws/event/EvtAgentStartHandler.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.handler;
+package cn.reghao.devops.manager.ws.event;
 
 import cn.reghao.jutil.jdk.event.message.Event;
 import cn.reghao.jutil.jdk.event.handler.Handler;

manager/src/main/java/cn/reghao/devops/manager/event/handler/EvtAppStatResultHandler.java → manager/src/main/java/cn/reghao/devops/manager/ws/event/EvtAppStatResultHandler.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.handler;
+package cn.reghao.devops.manager.ws.event;
 
 import cn.reghao.devops.common.msg.event.EvtAppStatResult;
 import cn.reghao.devops.manager.app.service.bd.DeployStat;

manager/src/main/java/cn/reghao/devops/manager/event/handler/EvtBuildAppResultHandler.java → manager/src/main/java/cn/reghao/devops/manager/ws/event/EvtBuildAppResultHandler.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.handler;
+package cn.reghao.devops.manager.ws.event;
 
 import cn.reghao.devops.common.msg.event.EvtBuildAppResult;
 import cn.reghao.jutil.jdk.event.handler.Handler;

manager/src/main/java/cn/reghao/devops/manager/event/handler/MachineEventService.java → manager/src/main/java/cn/reghao/devops/manager/ws/event/MachineEventService.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.handler;
+package cn.reghao.devops.manager.ws.event;
 
 import cn.reghao.devops.common.agent.app.dto.AppInfo;
 import cn.reghao.devops.common.msg.event.EvtAgentHeartbeat;

manager/src/main/java/cn/reghao/devops/manager/event/ws/handler/AgentWebSocketHandler.java → manager/src/main/java/cn/reghao/devops/manager/ws/handler/AgentWebSocketHandler.java

@@ -1,10 +1,10 @@
-package cn.reghao.devops.manager.event.ws.handler;
+package cn.reghao.devops.manager.ws.handler;
 
-import cn.reghao.devops.manager.event.ws.WsSender;
 import cn.reghao.devops.manager.machine.service.MachineService;
+import cn.reghao.devops.manager.ws.WsSender;
 import cn.reghao.jutil.jdk.serializer.JdkSerializer;
 import cn.reghao.devops.manager.machine.util.Keys;
-import cn.reghao.devops.manager.event.EventCenter;
+import cn.reghao.devops.manager.ws.EventCenter;
 import cn.reghao.jutil.jdk.event.message.EventMessage;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

manager/src/main/java/cn/reghao/devops/manager/ws/handler/LogHandler.java

@@ -0,0 +1,148 @@
+package cn.reghao.devops.manager.ws.handler;
+
+import cn.reghao.devops.manager.log.service.LogService;
+import cn.reghao.jutil.jdk.converter.DateTimeConverter;
+import cn.reghao.jutil.jdk.result.AppLog;
+import cn.reghao.jutil.jdk.serializer.JdkSerializer;
+import cn.reghao.jutil.jdk.serializer.JsonConverter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.*;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author reghao
+ * @date 2022-03-11 16:45:52
+ */
+@Slf4j
+@Component
+public class LogHandler implements WebSocketHandler {
+    private LogService logService;
+    private final Map<String, WebSocketSession> pullSessions = new ConcurrentHashMap<>();
+
+    public LogHandler(LogService logService) {
+        this.logService = logService;
+    }
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession webSocketSession) throws IOException {
+        String path = webSocketSession.getUri().getPath();
+        if (path.equals("/ws/log/push")) {
+            String query = webSocketSession.getUri().getQuery();
+            Map<String, String> map = parseParams(query);
+            String app = map.get("app");
+            String host = map.get("host");
+            String key = String.format("%s-%s", app, host);
+            //pushSessions.put(key, webSocketSession);
+        } else {
+            String query = webSocketSession.getUri().getQuery();
+            Map<String, String> map = parseParams(query);
+            String app = map.get("app");
+            String host = map.get("host");
+            String key = String.format("%s-%s", app, host);
+            putPullSession(app, host, webSocketSession);
+        }
+
+        log.info("WebSocket 建立连接");
+    }
+
+    private Map<String, String> parseParams(String query) {
+        String[] params = query.split("&");
+        Map<String, String> map = new HashMap<>();
+        for (String param : params) {
+            String[] arr = param.split("=");
+            map.put(arr[0], arr[1]);
+        }
+
+        return map;
+    }
+
+    @Override
+    public void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage)
+            throws IOException {
+        try {
+            if (webSocketMessage instanceof TextMessage) {
+                String payload = (String) webSocketMessage.getPayload();
+            } else if (webSocketMessage instanceof BinaryMessage) {
+                log.info("接收到 WebSocket 二进制消息");
+                BinaryMessage binaryMessage = (BinaryMessage) webSocketMessage;
+                Object object = JdkSerializer.deserialize(binaryMessage.getPayload().array());
+                if (object instanceof AppLog) {
+                    AppLog appLog = (AppLog) object;
+                    logService.saveAppLog(appLog);
+                    String dateTimeStr = DateTimeConverter.format(appLog.getTimestamp());
+
+                    String app = appLog.getApp();
+                    String host = appLog.getHost();
+                    WebSocketSession pullSession = getPullSession(app, host);
+                    if (pullSession != null) {
+                        String jsonData = JsonConverter.objectToJson(appLog);
+                        WebSocketMessage<String> message1 = new TextMessage(jsonData);
+                        try {
+                            pullSession.sendMessage(message1);
+                        } catch (IOException e) {
+                            e.printStackTrace();
+                        }
+                    }
+                }
+            } else if (webSocketMessage instanceof PingMessage) {
+                log.info("接收到 WebSocket PingMessage");
+            } else if (webSocketMessage instanceof PongMessage) {
+                log.info("接收到 WebSocket PongMessage");
+            } else {
+                log.error("接收到未知类型的 WebSocket 消息");
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void handleTransportError(WebSocketSession webSocketSession, Throwable throwable) {
+        log.error("WebSocket 数据传输错误");
+        String sessionId = webSocketSession.getId();
+        removePullSession(sessionId);
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession webSocketSession, CloseStatus closeStatus) {
+        log.info("WebSocket 断开连接");
+        String sessionId = webSocketSession.getId();
+        removePullSession(sessionId);
+    }
+
+    private void putPullSession(String app, String host, WebSocketSession session) {
+        String sessionId = session.getId();
+        String key = String.format("%s-%s-%s", sessionId, app, host);
+        pullSessions.put(key, session);
+    }
+
+    private WebSocketSession getPullSession(String app, String host) {
+        String suffix = String.format("%s-%s", app, host);
+        Set<String> keys = pullSessions.keySet();
+        for (String key : keys) {
+            if (key.endsWith(suffix)) {
+                return pullSessions.get(key);
+            }
+        }
+
+        return null;
+    }
+
+    private void removePullSession(String sessionId) {
+        Set<String> keys = pullSessions.keySet();
+        keys.forEach(key -> {
+            if (key.startsWith(sessionId)) {
+                pullSessions.remove(key);
+            }
+        });
+    }
+
+    @Override
+    public boolean supportsPartialMessages() {
+        return false;
+    }
+}

manager/src/main/java/cn/reghao/devops/manager/ws/handler/MessageHandler.java

@@ -0,0 +1,99 @@
+package cn.reghao.devops.manager.ws.handler;
+
+import cn.reghao.devops.manager.ws.MessageDispatcher;
+import cn.reghao.devops.manager.ws.model.EventType;
+import cn.reghao.jutil.jdk.serializer.JdkSerializer;
+import cn.reghao.jutil.jdk.serializer.JsonConverter;
+import com.google.gson.JsonObject;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.*;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author reghao
+ * @date 2022-03-11 16:45:52
+ */
+@Slf4j
+@Component
+public class MessageHandler implements WebSocketHandler {
+    private MessageDispatcher messageDispatcher;
+    private final Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();
+
+    public MessageHandler(MessageDispatcher messageDispatcher) {
+        this.messageDispatcher = messageDispatcher;
+    }
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession webSocketSession) throws IOException {
+        String sessionId = webSocketSession.getId();
+        sessionMap.put(sessionId, webSocketSession);
+        log.info("WebSocket 建立连接");
+    }
+
+    @Override
+    public void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage)
+            throws IOException {
+        if (webSocketMessage instanceof TextMessage) {
+            String payload = (String) webSocketMessage.getPayload();
+            JsonObject jsonObject = JsonConverter.jsonToJsonElement(payload).getAsJsonObject();
+            String event = jsonObject.get("event").getAsString();
+            String payload1 = jsonObject.get("payload").getAsString();
+            switch (EventType.valueOf(event)) {
+                case heartbeat:
+                    break;
+                case event_error:
+                    break;
+                case event_content:
+                    break;
+                case event_sync:
+                    break;
+                default:
+            }
+            /*log.info("接收到 WebSocket 文本消息");
+            TextMessage textMessage = (TextMessage) webSocketMessage;
+            JsonObject jsonObject = JsonConverter.jsonToJsonElement(textMessage.getPayload()).getAsJsonObject();
+            sessionMap.forEach((key, session) -> {
+                try {
+                    session.sendMessage(webSocketMessage);
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+            });*/
+        } else if (webSocketMessage instanceof BinaryMessage) {
+            log.info("接收到 WebSocket 二进制消息");
+            BinaryMessage binaryMessage = (BinaryMessage) webSocketMessage;
+            Object object = JdkSerializer.deserialize(binaryMessage.getPayload().array());
+        } else if (webSocketMessage instanceof PingMessage) {
+            log.info("接收到 WebSocket PingMessage");
+        } else if (webSocketMessage instanceof PongMessage) {
+            log.info("接收到 WebSocket PongMessage");
+        } else {
+            log.error("接收到未知类型的 WebSocket 消息");
+        }
+    }
+
+    @Override
+    public void handleTransportError(WebSocketSession webSocketSession, Throwable throwable) {
+        log.error("WebSocket 数据传输错误");
+        removeSession(webSocketSession);
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession webSocketSession, CloseStatus closeStatus) {
+        log.info("WebSocket 断开连接");
+        removeSession(webSocketSession);
+    }
+
+    private void removeSession(WebSocketSession webSocketSession) {
+        sessionMap.remove(webSocketSession.getId());
+    }
+
+    @Override
+    public boolean supportsPartialMessages() {
+        return false;
+    }
+}

manager/src/main/java/cn/reghao/devops/manager/event/ws/handler/SshWebSocketHandler.java → manager/src/main/java/cn/reghao/devops/manager/ws/handler/SshWebSocketHandler.java

@@ -1,4 +1,4 @@
-package cn.reghao.devops.manager.event.ws.handler;
+package cn.reghao.devops.manager.ws.handler;
 
 import cn.reghao.devops.manager.machine.util.Keys;
 import cn.reghao.devops.manager.machine.util.WebSsh;

manager/src/main/java/cn/reghao/devops/manager/ws/model/EventType.java

@@ -0,0 +1,9 @@
+package cn.reghao.devops.manager.ws.model;
+
+/**
+ * @author reghao
+ * @date 2022-04-19 12:22:51
+ */
+public enum EventType {
+    heartbeat, event_error, event_content, event_sync
+}

manager/src/main/java/cn/reghao/devops/manager/ws/model/event/EvtHeartbeat.java

@@ -0,0 +1,13 @@
+package cn.reghao.devops.manager.ws.model.event;
+
+import cn.reghao.jutil.jdk.event.message.Event;
+import lombok.Getter;
+
+/**
+ * @author reghao
+ * @date 2022-05-13 17:01:50
+ */
+@Getter
+public class EvtHeartbeat extends Event {
+
+}

manager/src/main/java/cn/reghao/devops/manager/ws/model/req/EvtHeartbeatReq.java

@@ -0,0 +1,12 @@
+package cn.reghao.devops.manager.ws.model.req;
+
+import lombok.Getter;
+
+/**
+ * @author reghao
+ * @date 2022-04-19 12:17:37
+ */
+@Getter
+public class EvtHeartbeatReq {
+    private String content;
+}

manager/src/main/java/cn/reghao/devops/manager/ws/model/resp/EventMessageResp.java

@@ -0,0 +1,21 @@
+package cn.reghao.devops.manager.ws.model.resp;
+
+import cn.reghao.devops.manager.ws.model.EventType;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @author reghao
+ * @date 2022-04-21 11:05:56
+ */
+@Getter
+@Setter
+public class EventMessageResp <T> {
+    private final String event;
+    private final T payload;
+
+    public EventMessageResp(EventType eventType, T payload) {
+        this.event = eventType.name();
+        this.payload = payload;
+    }
+}