webSocketSessions = new ConcurrentHashMap<>();
+ public record LogStreamSession(
+ WebSocketSession webSocket,
+ Future task,
+ AtomicBoolean paused,
+ ILogStreamStrategy strategy,
+ Long userId
+ ) {}
/**
- * 日志流任务存储：sessionId → Future
+ * 会话存储：sessionId → LogStreamSession
+ * 一个Map管理所有会话状态，保证原子性
*/
- protected final Map> streamTasks = new ConcurrentHashMap<>();
+ private final Map sessions = new ConcurrentHashMap<>();
- /**
- * 暂停标志存储：sessionId → AtomicBoolean
- */
- protected final Map pausedFlags = new ConcurrentHashMap<>();
+ // ==================== 依赖注入 ====================
- /**
- * 目标信息存储：sessionId → LogStreamTarget
- */
- protected final Map sessionTargets = new ConcurrentHashMap<>();
-
- /**
- * 策略实例存储：sessionId → ILogStreamStrategy
- */
- protected final Map sessionStrategies = new ConcurrentHashMap<>();
-
- /**
- * SSH会话管理器（用于配额管理）
- */
@Resource
private SSHSessionManager sshSessionManager;
- /**
- * 日志流输出监听线程池（Framework自动注入）
- */
@Resource(name = "logStreamOutputExecutor")
private AsyncTaskExecutor logStreamOutputExecutor;
- /**
- * 最大SSH连接数（与SSH终端共享配额）
- */
+ /** 最大SSH连接数（与SSH终端共享配额） */
private static final int MAX_SSH_SESSIONS = 5;
- // ========== 辅助方法 ==========
+ // ==================== 抽象方法（子类实现） ====================
/**
- * 获取增强的SessionId（线程安全）
- *
- * 使用SessionIdGenerator增强原始WebSocket SessionId，确保并发场景下的唯一性
- *
- * @param session WebSocket会话
- * @return 增强的SessionId
- */
- protected String getSessionId(WebSocketSession session) {
- // 1. 先尝试从session attributes中获取（避免重复生成）
- String sessionId = (String) session.getAttributes().get("logStreamSessionId");
-
- if (sessionId == null) {
- // 2. 懒加载：如果没有，就生成并存储（保证一致性）
- sessionId = SessionIdGenerator.enhanceWebSocketSessionId(session.getId());
- session.getAttributes().put("logStreamSessionId", sessionId);
- }
-
- return sessionId;
- }
-
- // ========== 子类必须实现的抽象方法 ==========
-
- /**
- * 获取日志流目标信息（由子类实现）
- *
- * @param session WebSocket会话
- * @param request 日志流请求
- * @return 日志流目标信息
- * @throws Exception 获取失败时抛出
+ * 获取日志流目标信息
*/
protected abstract LogStreamTarget getLogStreamTarget(WebSocketSession session, LogStreamRequest request) throws Exception;
/**
- * 检查用户权限（由子类实现）
- *
- * @param userId 用户ID
- * @param target 日志流目标
- * @return 是否有权限
+ * 检查用户权限
*/
protected abstract boolean checkPermission(Long userId, LogStreamTarget target);
/**
- * 获取日志流策略（由子类实现）
- *
- * @param target 日志流目标
- * @return 日志流策略
+ * 获取日志流策略
*/
protected abstract ILogStreamStrategy getLogStreamStrategy(LogStreamTarget target);
- // ========== Framework 提供的核心能力 ==========
+ // ==================== WebSocket生命周期 ====================
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
String sessionId = getSessionId(session);
- log.info("日志流WebSocket连接建立: webSocketId={}, logStreamSessionId={}",
- session.getId(), sessionId);
+ log.info("日志流WebSocket连接建立: sessionId={}", sessionId);
- try {
- // 1. 获取用户信息
- Long userId = (Long) session.getAttributes().get("userId");
- String username = (String) session.getAttributes().get("username");
-
- if (userId == null) {
- log.error("无法获取用户信息: sessionId={}", sessionId);
- sendError(session, "认证失败");
- session.close(CloseStatus.POLICY_VIOLATION);
- return;
- }
-
- // 2. 保存会话
- webSocketSessions.put(sessionId, session);
-
- // 3. 不立即发送状态消息，等待客户端发送START消息
- // 参照SSH WebSocket的做法，避免在客户端未准备好时发送消息
- log.info("日志流连接成功，等待START消息: sessionId={}, userId={}, username={}",
- sessionId, userId, username);
-
- } catch (Exception e) {
- log.error("建立日志流连接失败: sessionId={}", sessionId, e);
- cleanupSession(sessionId);
-
- try {
- session.close(CloseStatus.SERVER_ERROR);
- } catch (IOException ex) {
- log.error("关闭WebSocket会话失败: sessionId={}", sessionId, ex);
- }
+ Long userId = (Long) session.getAttributes().get("userId");
+ if (userId == null) {
+ log.error("无法获取用户信息: sessionId={}", sessionId);
+ sendError(session, "认证失败");
+ session.close(CloseStatus.POLICY_VIOLATION);
+ return;
}
+
+ // 预注册会话（只有基本信息，task/strategy在START时填充）
+ sessions.put(sessionId, new LogStreamSession(session, null, null, null, userId));
+
+ log.info("日志流连接成功，等待START消息: sessionId={}, userId={}", sessionId, userId);
}
@Override
@@ -186,118 +127,113 @@ public abstract class AbstractLogStreamWebSocketHandler extends TextWebSocketHan
try {
LogWebSocketMessage msg = JsonUtils.fromJson(message.getPayload(), LogWebSocketMessage.class);
- if (msg.getType() == LogMessageType.START) {
- // 启动日志流
- handleStartMessage(session, msg);
- } else if (msg.getType() == LogMessageType.CONTROL) {
- // 控制消息
- handleControlMessage(session, msg);
- } else {
- log.warn("未知的消息类型: sessionId={}, type={}", sessionId, msg.getType());
+ switch (msg.getType()) {
+ case START -> handleStartMessage(session, sessionId, msg);
+ case CONTROL -> handleControlMessage(sessionId, msg);
+ default -> log.warn("未知的消息类型: sessionId={}, type={}", sessionId, msg.getType());
}
-
} catch (Exception e) {
log.error("处理WebSocket消息失败: sessionId={}", sessionId, e);
sendError(session, "消息处理失败: " + e.getMessage());
}
}
- /**
- * 处理START消息
- */
- private void handleStartMessage(WebSocketSession session, LogWebSocketMessage msg) {
+ @Override
+ public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+ String sessionId = getSessionId(session);
+ log.info("日志流WebSocket连接关闭: sessionId={}, status={}", sessionId, status);
+ cleanupSession(sessionId);
+ }
+
+ @Override
+ public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
String sessionId = getSessionId(session);
+ if (exception instanceof java.io.EOFException) {
+ log.debug("客户端关闭连接: sessionId={}", sessionId);
+ } else {
+ log.error("日志流WebSocket传输错误: sessionId={}", sessionId, exception);
+ sendError(session, "传输错误: " + exception.getMessage());
+ }
+
+ cleanupSession(sessionId);
+
try {
- // 1. 提取请求参数
+ session.close(CloseStatus.SERVER_ERROR);
+ } catch (IOException ignored) {}
+ }
+
+ // ==================== 消息处理 ====================
+
+ private void handleStartMessage(WebSocketSession session, String sessionId, LogWebSocketMessage msg) {
+ try {
+ // 1. 验证请求
LogStreamRequest request = msg.getRequest(LogStreamRequest.class);
if (request == null || !request.isValid()) {
- log.warn("START消息参数无效: sessionId={}", sessionId);
sendError(session, "请求参数无效");
return;
}
- // 2. 获取用户信息
- Long userId = (Long) session.getAttributes().get("userId");
+ // 2. 获取当前会话
+ LogStreamSession currentSession = sessions.get(sessionId);
+ if (currentSession == null) {
+ sendError(session, "会话不存在");
+ return;
+ }
- // 3. 获取日志流目标信息
+ Long userId = currentSession.userId();
+
+ // 3. 获取日志流目标
LogStreamTarget target = getLogStreamTarget(session, request);
if (target == null) {
- log.error("无法获取日志流目标: sessionId={}", sessionId);
sendError(session, "无法获取日志流目标");
return;
}
- // 保存target信息
- sessionTargets.put(sessionId, target);
-
- log.info("获取日志流目标成功: sessionId={}, runtimeType={}, name={}",
+ log.info("获取日志流目标: sessionId={}, runtimeType={}, name={}",
sessionId, target.getRuntimeType(), target.getName());
// 4. 权限验证
if (!checkPermission(userId, target)) {
- log.warn("用户无权访问日志: userId={}, target={}", userId, target.getName());
sendError(session, "无权访问此日志");
return;
}
- // 5. 对于需要SSH的运行时类型，检查配额
- if (target.getRuntimeType() == RuntimeTypeEnum.SERVER
- || target.getRuntimeType() == RuntimeTypeEnum.DOCKER) {
-
- // 尝试注册会话（包含配额检查）
- boolean registered = sshSessionManager.tryRegisterSession(
- sessionId, userId,
- SSHTargetType.LOG_STREAM,
- target.getName(), MAX_SSH_SESSIONS);
-
- if (!registered) {
- long currentCount = sshSessionManager.countUserTotalSessions(userId);
- log.warn("用户SSH连接数超过限制: userId={}, current={}, max={}",
- userId, currentCount, MAX_SSH_SESSIONS);
- sendError(session, "SSH连接数已达上限(" + MAX_SSH_SESSIONS + "个)，请关闭其他连接后重试");
- return; // 不启动日志流
+ // 5. SSH配额检查（仅Server/Docker）
+ if (requiresSSH(target.getRuntimeType())) {
+ if (!tryRegisterSSHSession(sessionId, userId, target.getName())) {
+ sendError(session, "SSH连接数已达上限(" + MAX_SSH_SESSIONS + "个)");
+ return;
}
-
- log.debug("日志流SSH会话注册成功: sessionId={}, userId={}", sessionId, userId);
}
- // 6. 发送流式传输状态
- sendStatus(session, LogStatusEnum.STREAMING);
-
- // 6. 创建暂停标志
- AtomicBoolean paused = new AtomicBoolean(false);
- pausedFlags.put(sessionId, paused);
-
- // 7. 获取日志流策略
+ // 6. 获取策略
ILogStreamStrategy strategy = getLogStreamStrategy(target);
if (strategy == null) {
- log.error("无法获取日志流策略: sessionId={}, runtimeType={}",
- sessionId, target.getRuntimeType());
sendError(session, "不支持的运行时类型");
return;
}
- // 保存策略实例，用于后续清理
- sessionStrategies.put(sessionId, strategy);
+ // 7. 创建暂停标志
+ AtomicBoolean paused = new AtomicBoolean(false);
- // 8. 启动日志流任务（异步，使用Spring管理的虚拟线程池）
+ // 8. 启动日志流任务
Future task = logStreamOutputExecutor.submit(() -> {
try {
- // 使用策略执行日志流
- strategy.streamLogs(session, target, paused,
+ // Strategy 只接收 sessionId，不接触 WebSocket
+ strategy.streamLogs(sessionId, target, paused::get,
(timestamp, content) -> sendLogLine(session, timestamp, content));
} catch (Exception e) {
log.error("日志流异常: sessionId={}", sessionId, e);
- try {
- sendError(session, "日志流中断: " + e.getMessage());
- } catch (Exception ex) {
- log.error("发送错误消息失败: sessionId={}", sessionId, ex);
- }
+ sendError(session, "日志流中断: " + e.getMessage());
}
});
- streamTasks.put(sessionId, task);
+ // 9. 更新会话（原子替换）
+ sessions.put(sessionId, new LogStreamSession(session, task, paused, strategy, userId));
+
+ // 10. 发送状态
+ sendStatus(session, LogStatusEnum.STREAMING);
log.info("日志流已启动: sessionId={}", sessionId);
} catch (Exception e) {
@@ -306,228 +242,173 @@ public abstract class AbstractLogStreamWebSocketHandler extends TextWebSocketHan
}
}
- /**
- * 处理CONTROL消息
- */
- private void handleControlMessage(WebSocketSession session, LogWebSocketMessage msg) {
- String sessionId = getSessionId(session);
+ private void handleControlMessage(String sessionId, LogWebSocketMessage msg) {
+ LogStreamSession session = sessions.get(sessionId);
+ if (session == null || session.paused() == null) {
+ log.warn("日志流未启动，无法控制: sessionId={}", sessionId);
+ return;
+ }
try {
LogControlRequest request = msg.getRequest(LogControlRequest.class);
if (request == null || !request.isValid()) {
- log.warn("CONTROL消息参数无效: sessionId={}", sessionId);
- return;
- }
-
- AtomicBoolean paused = pausedFlags.get(sessionId);
- if (paused == null) {
- log.warn("日志流未启动，无法控制: sessionId={}", sessionId);
return;
}
LogControlAction action = request.getAction();
- if (action == LogControlAction.PAUSE) {
- paused.set(true);
- sendStatus(session, LogStatusEnum.PAUSED);
- log.info("日志流已暂停: sessionId={}", sessionId);
- } else if (action == LogControlAction.RESUME) {
- paused.set(false);
- sendStatus(session, LogStatusEnum.STREAMING);
- log.info("日志流已恢复: sessionId={}", sessionId);
- } else if (action == LogControlAction.STOP) {
- log.info("收到停止请求: sessionId={}", sessionId);
- session.close(CloseStatus.NORMAL);
+ switch (action) {
+ case PAUSE -> {
+ session.paused().set(true);
+ sendStatus(session.webSocket(), LogStatusEnum.PAUSED);
+ log.info("日志流已暂停: sessionId={}", sessionId);
+ }
+ case RESUME -> {
+ session.paused().set(false);
+ sendStatus(session.webSocket(), LogStatusEnum.STREAMING);
+ log.info("日志流已恢复: sessionId={}", sessionId);
+ }
+ case STOP -> {
+ log.info("收到停止请求: sessionId={}", sessionId);
+ session.webSocket().close(CloseStatus.NORMAL);
+ }
}
-
} catch (Exception e) {
log.error("处理CONTROL消息失败: sessionId={}", sessionId, e);
}
}
- @Override
- public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
- String sessionId = getSessionId(session);
- log.info("日志流WebSocket连接关闭: logStreamSessionId={}, status={}", sessionId, status);
- cleanupSession(sessionId);
- }
+ // ==================== 资源清理 ====================
- @Override
- public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
- String sessionId = getSessionId(session);
-
- // EOFException通常表示客户端正常关闭连接，不需要记录ERROR日志
- if (exception instanceof java.io.EOFException) {
- log.debug("客户端关闭连接: sessionId={}", sessionId);
- cleanupSession(sessionId);
+ private void cleanupSession(String sessionId) {
+ LogStreamSession session = sessions.remove(sessionId);
+ if (session == null) {
return;
}
- log.error("日志流WebSocket传输错误: sessionId={}", sessionId, exception);
+ log.info("清理日志流会话: sessionId={}", sessionId);
- try {
- sendError(session, "传输错误: " + exception.getMessage());
- } catch (Exception e) {
- // 忽略发送错误消息时的异常
- log.debug("发送错误消息失败: sessionId={}", sessionId);
- }
-
- cleanupSession(sessionId);
-
- try {
- session.close(CloseStatus.SERVER_ERROR);
- } catch (IOException e) {
- log.debug("关闭WebSocket会话失败: sessionId={}", sessionId);
- }
- }
-
- /**
- * 清理会话资源
- */
- private void cleanupSession(String sessionId) {
- log.info("开始清理日志流会话资源: sessionId={}", sessionId);
-
- try {
- // 1. 调用Strategy的stop方法清理资源（SSH连接等）
- ILogStreamStrategy strategy = sessionStrategies.remove(sessionId);
- if (strategy != null) {
- try {
- log.debug("调用Strategy.stop清理资源: sessionId={}", sessionId);
- strategy.stop(sessionId);
- } catch (Exception e) {
- log.error("Strategy清理资源失败: sessionId={}", sessionId, e);
- }
- }
-
- // 2. 从SSH会话管理器移除（释放配额）
+ // 1. 停止策略（清理SSH/K8S连接）
+ if (session.strategy() != null) {
try {
- sshSessionManager.removeSession(sessionId);
- log.debug("SSH会话已从配额管理器移除: sessionId={}", sessionId);
+ session.strategy().stop(sessionId);
} catch (Exception e) {
- log.error("从SSH会话管理器移除失败: sessionId={}", sessionId, e);
+ log.error("Strategy清理失败: sessionId={}", sessionId, e);
}
-
- // 3. 取消日志流任务
- Future task = streamTasks.remove(sessionId);
- if (task != null && !task.isDone()) {
- log.debug("取消日志流任务: sessionId={}", sessionId);
- task.cancel(true);
- }
-
- // 4. 移除WebSocketSession
- webSocketSessions.remove(sessionId);
-
- // 5. 移除暂停标志
- pausedFlags.remove(sessionId);
-
- // 6. 移除target信息
- sessionTargets.remove(sessionId);
-
- log.info("日志流会话资源清理完成: sessionId={}", sessionId);
-
- } catch (Exception e) {
- log.error("清理会话资源失败: sessionId={}", sessionId, e);
}
+
+ // 2. 释放SSH配额
+ sshSessionManager.removeSession(sessionId);
+
+ // 3. 取消异步任务
+ if (session.task() != null && !session.task().isDone()) {
+ session.task().cancel(true);
+ }
+
+ log.info("日志流会话清理完成: sessionId={}", sessionId);
}
- // ========== 辅助方法（供子类使用） ==========
+ // ==================== 辅助方法 ====================
- /**
- * 发送日志行到前端
- *
- * @param session WebSocket会话
- * @param timestamp 时间戳
- * @param content 日志内容
- */
- protected void sendLogLine(WebSocketSession session, String timestamp, String content) {
- String sessionId = getSessionId(session);
+ private boolean requiresSSH(RuntimeTypeEnum runtimeType) {
+ return runtimeType == RuntimeTypeEnum.SERVER || runtimeType == RuntimeTypeEnum.DOCKER;
+ }
+
+ private boolean tryRegisterSSHSession(String sessionId, Long userId, String targetName) {
+ boolean registered = sshSessionManager.tryRegisterSession(
+ sessionId, userId, SSHTargetType.LOG_STREAM, targetName, MAX_SSH_SESSIONS);
+
+ if (!registered) {
+ long currentCount = sshSessionManager.countUserTotalSessions(userId);
+ log.warn("SSH连接数超限: userId={}, current={}, max={}", userId, currentCount, MAX_SSH_SESSIONS);
+ }
+
+ return registered;
+ }
+
+ protected String getSessionId(WebSocketSession session) {
+ String sessionId = (String) session.getAttributes().get("logStreamSessionId");
+ if (sessionId == null) {
+ sessionId = SessionIdGenerator.enhanceWebSocketSessionId(session.getId());
+ session.getAttributes().put("logStreamSessionId", sessionId);
+ }
+ return sessionId;
+ }
+
+ // ==================== 消息发送 ====================
+
+ private void sendLogLine(WebSocketSession session, String timestamp, String content) {
+ if (session == null || !session.isOpen()) {
+ return;
+ }
try {
- if (session == null) {
- log.warn("发送日志行失败: session为null, sessionId={}", sessionId);
- return;
- }
+ LogWebSocketMessage msg = LogWebSocketMessage.of(
+ LogMessageType.LOG,
+ new LogLineResponse(timestamp, content)
+ );
- if (!session.isOpen()) {
- log.warn("发送日志行失败: session已关闭, sessionId={}", sessionId);
- return;
- }
-
- LogLineResponse response = new LogLineResponse(timestamp, content);
-
- Map data = new HashMap<>();
- data.put("response", response);
-
- LogWebSocketMessage msg = new LogWebSocketMessage(LogMessageType.LOG, data);
-
- // WebSocketSession.sendMessage()不是线程安全的，必须加锁
synchronized (session) {
session.sendMessage(new TextMessage(JsonUtils.toJson(msg)));
}
-
- log.trace("日志行已发送: sessionId={}, content={}", sessionId, content.substring(0, Math.min(50, content.length())));
-
} catch (IOException e) {
- log.error("发送日志行失败(IOException): sessionId={}, error={}", sessionId, e.getMessage(), e);
- } catch (Exception e) {
- log.error("发送日志行失败(Exception): sessionId={}, error={}", sessionId, e.getMessage(), e);
+ log.debug("发送日志行失败: sessionId={}", getSessionId(session));
}
}
- /**
- * 发送状态消息到前端
- */
protected void sendStatus(WebSocketSession session, LogStatusEnum status) {
+ if (session == null || !session.isOpen()) {
+ return;
+ }
+
try {
- if (session == null || !session.isOpen()) {
- log.debug("会话未打开，跳过发送状态消息: status={}", status);
- return;
- }
+ LogWebSocketMessage msg = LogWebSocketMessage.of(
+ LogMessageType.STATUS,
+ new LogStatusResponse(status)
+ );
- LogStatusResponse response = new LogStatusResponse(status);
-
- Map data = new HashMap<>();
- data.put("response", response);
-
- LogWebSocketMessage msg = new LogWebSocketMessage(LogMessageType.STATUS, data);
-
- // WebSocketSession.sendMessage()不是线程安全的，必须加锁
synchronized (session) {
session.sendMessage(new TextMessage(JsonUtils.toJson(msg)));
}
-
} catch (IOException e) {
- // 降低日志级别，客户端断开是正常情况
- log.debug("发送状态消息失败（客户端可能已断开）: logStreamSessionId={}, status={}",
- getSessionId(session), status);
+ log.debug("发送状态消息失败: sessionId={}", getSessionId(session));
}
}
- /**
- * 发送错误消息到前端
- */
protected void sendError(WebSocketSession session, String error) {
+ if (session == null || !session.isOpen()) {
+ return;
+ }
+
try {
- if (!session.isOpen()) {
- return;
- }
+ LogWebSocketMessage msg = LogWebSocketMessage.of(
+ LogMessageType.ERROR,
+ new LogErrorResponse(error)
+ );
- LogErrorResponse response = new LogErrorResponse(error);
-
- Map data = new HashMap<>();
- data.put("response", response);
-
- LogWebSocketMessage msg = new LogWebSocketMessage(LogMessageType.ERROR, data);
-
- // WebSocketSession.sendMessage()不是线程安全的，必须加锁
synchronized (session) {
session.sendMessage(new TextMessage(JsonUtils.toJson(msg)));
}
-
} catch (IOException e) {
- if (session.isOpen()) {
- log.error("发送错误消息失败: logStreamSessionId={}", getSessionId(session), e);
- }
+ log.debug("发送错误消息失败: sessionId={}", getSessionId(session));
}
}
+
+ // ==================== 监控方法（供子类或管理接口使用） ====================
+
+ /**
+ * 获取当前活跃会话数
+ */
+ protected int getActiveSessionCount() {
+ return sessions.size();
+ }
+
+ /**
+ * 获取指定用户的会话数
+ */
+ protected long getUserSessionCount(Long userId) {
+ return sessions.values().stream()
+ .filter(s -> userId.equals(s.userId()))
+ .count();
+ }
}
diff --git a/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/ILogStreamStrategy.java b/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/ILogStreamStrategy.java
index 0ffd6598..a9d910c1 100644
--- a/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/ILogStreamStrategy.java
+++ b/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/ILogStreamStrategy.java
@@ -1,13 +1,14 @@
package com.qqchen.deploy.backend.framework.websocket.log;
import com.qqchen.deploy.backend.deploy.enums.RuntimeTypeEnum;
-import org.springframework.web.socket.WebSocketSession;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
/**
* 日志流策略接口
- * 定义不同运行时类型的日志流获取方式
+ *
+ * 定义不同运行时类型的日志流获取方式。
+ * Strategy 只负责连接管理和日志读取，不涉及 WebSocket 协议。
*
* @author Framework
* @since 2025-12-16
@@ -16,31 +17,27 @@ public interface ILogStreamStrategy {
/**
* 支持的运行时类型
- *
- * @return 运行时类型
*/
RuntimeTypeEnum supportedType();
/**
* 执行日志流推送
- * 此方法应该持续读取日志并通过callback推送到前端
*
- * @param session WebSocket会话
+ * @param sessionId 会话ID（由 Handler 生成）
* @param target 日志流目标信息
- * @param paused 暂停标志（实现应定期检查此标志）
- * @param callback 日志行回调接口
+ * @param isPaused 暂停状态查询（由 Handler 控制）
+ * @param callback 日志行回调
* @throws Exception 流式推送失败时抛出
*/
- void streamLogs(WebSocketSession session,
- LogStreamTarget target,
- AtomicBoolean paused,
- LogLineCallback callback) throws Exception;
+ void streamLogs(String sessionId,
+ LogStreamTarget target,
+ Supplier isPaused,
+ LogLineCallback callback) throws Exception;
/**
* 停止日志流并清理资源
- * 当WebSocket连接关闭时调用，确保SSH连接等资源被正确释放
*
- * @param sessionId WebSocket会话ID
+ * @param sessionId 会话ID
*/
void stop(String sessionId);
@@ -49,12 +46,6 @@ public interface ILogStreamStrategy {
*/
@FunctionalInterface
interface LogLineCallback {
- /**
- * 发送日志行
- *
- * @param timestamp 时间戳
- * @param content 日志内容
- */
void sendLogLine(String timestamp, String content);
}
}
diff --git a/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/LogWebSocketMessage.java b/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/LogWebSocketMessage.java
index 9ee720d0..900e0df8 100644
--- a/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/LogWebSocketMessage.java
+++ b/backend/src/main/java/com/qqchen/deploy/backend/framework/websocket/log/LogWebSocketMessage.java
@@ -34,6 +34,17 @@ public class LogWebSocketMessage {
*/
private Map data;
+ /**
+ * 创建响应消息的工厂方法
+ *
+ * @param type 消息类型
+ * @param response 响应对象
+ * @return LogWebSocketMessage
+ */
+ public static LogWebSocketMessage of(LogMessageType type, Object response) {
+ return new LogWebSocketMessage(type, Map.of("response", response));
+ }
+
/**
* 从data中提取request对象（强类型）
*