WebSocket改造

magic-api/src/main/java/org/ssssssss/magicapi/config/MessageType.java

@@ -23,6 +23,8 @@ public enum MessageType {
 	ONLINE_USERS,
 	/* 通知客户端，他人进入文件*/
 	INTO_FILE_ID,
+	/* PONG */
+	PONG,
 
 	/* C -> S message */
 	/* 设置断点 */
@@ -33,6 +35,8 @@ public enum MessageType {
 	LOGIN,
 	/* 设置当前所在文件 */
 	SET_FILE_ID,
+	/* ping */
+	PING,
 
 	/* S <-> S -> C message*/
 	/* 获取当前在线用户 */

magic-api/src/main/java/org/ssssssss/magicapi/config/WebSocketSessionManager.java

@@ -11,11 +11,7 @@ import org.ssssssss.magicapi.provider.MagicNotifyService;
 import org.ssssssss.magicapi.utils.JsonUtils;
 import org.ssssssss.script.MagicScriptDebugContext;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -39,13 +35,19 @@ public class WebSocketSessionManager {
 		SESSIONS.put(session.getClientId(), session);
 	}
 
-	static{
+	public static MagicConsoleSession getConsoleSession(String clientId) {
+		return SESSIONS.get(clientId);
+	}
+
+	static {
 		// 1秒1次发送日志
 		new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "magic-api-send-log-task")).scheduleAtFixedRate(WebSocketSessionManager::flushLog, 1, 1, TimeUnit.SECONDS);
+		// 25秒检测一次是否在线
+		new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "magic-api-websocket-clean-task")).scheduleAtFixedRate(WebSocketSessionManager::checkSession, 25, 25, TimeUnit.SECONDS);
 	}
 
-	public static List<MagicConsoleSession> getSessions() {
-		return new ArrayList<>(SESSIONS.values());
+	public static Collection<MagicConsoleSession> getSessions() {
+		return SESSIONS.values();
 	}
 
 	public static void remove(MagicConsoleSession session) {
@@ -64,7 +66,7 @@ public class WebSocketSessionManager {
 	}
 
 	private static void sendToAll(String content) {
-		SESSIONS.values().stream().filter(MagicConsoleSession::writeable).forEach(session -> sendBySession(session, content));
+		getSessions().stream().filter(MagicConsoleSession::writeable).forEach(session -> sendBySession(session, content));
 		sendToMachineByClientId(null, content);
 	}
 
@@ -92,7 +94,7 @@ public class WebSocketSessionManager {
 				}
 			});
 		} catch (Exception e) {
-			logger.warn("发生日志失败", e);
+			logger.warn("发送日志失败", e);
 		}
 	}
 
@@ -108,7 +110,7 @@ public class WebSocketSessionManager {
 
 	public static void sendToOther(String excludeClientId, MessageType messageType, Object... values) {
 		String content = buildMessage(messageType, values);
-		SESSIONS.values().stream()
+		getSessions().stream()
 				.filter(MagicConsoleSession::writeable)
 				.filter(it -> !it.getClientId().equals(excludeClientId))
 				.forEach(session -> sendBySession(session, content));
@@ -122,7 +124,7 @@ public class WebSocketSessionManager {
 		}
 	}
 
-	public static void sendToMachine(MessageType messageType, Object ... args) {
+	public static void sendToMachine(MessageType messageType, Object... args) {
 		if (magicNotifyService != null) {
 			// 通知其他机器去发送消息
 			magicNotifyService.sendNotify(new MagicNotify(instanceId, EventAction.WS_S_S, null, buildMessage(messageType, args)));
@@ -163,13 +165,13 @@ public class WebSocketSessionManager {
 					session.getWebSocketSession().sendMessage(new TextMessage(content));
 				}
 			}
-		} catch (IOException e) {
-			logger.error("发送WebSocket消息失败", e);
+		} catch (Exception e) {
+			logger.warn("发送WebSocket消息失败: {}", e.getMessage());
 		}
 	}
 
 	public static MagicConsoleSession findSession(String clientId) {
-		return SESSIONS.values().stream()
+		return getSessions().stream()
 				.filter(it -> Objects.equals(clientId, it.getClientId()))
 				.findFirst()
 				.orElse(null);
@@ -195,4 +197,20 @@ public class WebSocketSessionManager {
 		CONTEXTS.remove(sessionAndScriptId);
 	}
 
+	private static void checkSession() {
+		try {
+			long activateTime = System.currentTimeMillis() - 20 * 1000;
+			SESSIONS.entrySet().stream()
+					.filter(it -> it.getValue().getActivateTime() < activateTime)
+					.collect(Collectors.toList())
+					.forEach(entry -> {
+						MagicConsoleSession session = entry.getValue();
+						SESSIONS.remove(entry.getKey());
+						session.close();
+						sendToAll(MessageType.USER_LOGOUT, session.getAttributes());
+					});
+		} catch (Exception ignored) {
+		}
+	}
+
 }

magic-api/src/main/java/org/ssssssss/magicapi/controller/MagicWorkbenchHandler.java

@@ -39,7 +39,14 @@ public class MagicWorkbenchHandler {
 				session.setAttribute(Constants.WEBSOCKET_ATTRIBUTE_USER_IP, ip);
 				session.setAttribute(Constants.WEBSOCKET_ATTRIBUTE_USER_NAME, user.getUsername());
 				session.setClientId(clientId);
-				WebSocketSessionManager.add(session);
+				session.setActivateTime(System.currentTimeMillis());
+				synchronized (MagicWorkbenchHandler.class){
+					if(WebSocketSessionManager.getConsoleSession(clientId) != null){
+						WebSocketSessionManager.sendBySession(session, WebSocketSessionManager.buildMessage(MessageType.LOGIN_RESPONSE, "-1"));
+						return;
+					}
+					WebSocketSessionManager.add(session);
+				}
 				WebSocketSessionManager.sendBySession(session, WebSocketSessionManager.buildMessage(MessageType.LOGIN_RESPONSE, "1", session.getAttributes()));
 				List<Map<String, Object>> messages = getOnlineUsers();
 				if(!messages.isEmpty()){
@@ -61,6 +68,13 @@ public class MagicWorkbenchHandler {
 		}
 	}
 
+	@Message(MessageType.PING)
+	public void ping(MagicConsoleSession session){
+		long activateTime = System.currentTimeMillis();
+		session.setActivateTime(activateTime);
+		WebSocketSessionManager.sendBySession(session, WebSocketSessionManager.buildMessage(MessageType.PONG, activateTime));
+	}
+
 	private List<Map<String, Object>> getOnlineUsers(){
 		return WebSocketSessionManager.getSessions().stream()
 				.map(MagicConsoleSession::getAttributes)

magic-api/src/main/java/org/ssssssss/magicapi/model/MagicConsoleSession.java

@@ -1,10 +1,10 @@
 package org.ssssssss.magicapi.model;
 
+import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.WebSocketSession;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class MagicConsoleSession {
@@ -13,10 +13,12 @@ public class MagicConsoleSession {
 
 	private String clientId;
 
-	private final WebSocketSession webSocketSession;
+	private WebSocketSession webSocketSession;
 
 	private final Map<String, Object> attributes = new HashMap<>();
 
+	private long activateTime = System.currentTimeMillis();
+
 	public MagicConsoleSession(WebSocketSession webSocketSession) {
 		this.webSocketSession = webSocketSession;
 	}
@@ -62,4 +64,24 @@ public class MagicConsoleSession {
 		this.clientId = clientId;
 		setAttribute(Constants.WEBSOCKET_ATTRIBUTE_CLIENT_ID, clientId);
 	}
+
+	public long getActivateTime() {
+		return activateTime;
+	}
+
+	public void setActivateTime(long activateTime) {
+		this.activateTime = activateTime;
+	}
+
+	public void close(){
+		if(this.webSocketSession != null){
+			remove(this.webSocketSession);
+			try {
+				this.webSocketSession.close(CloseStatus.SESSION_NOT_RELIABLE);
+			} catch (Exception ignored) {
+
+			}
+			this.webSocketSession = null;
+		}
+	}
 }