From 34d98570899e3035578e0c1a76bfb4b41a9c49a1 Mon Sep 17 00:00:00 2001 From: caixypromise Date: Tue, 20 May 2025 20:02:16 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E6=99=BA?= =?UTF-8?q?=E6=8E=A7=E5=8F=B0=E7=AE=A1=E7=90=86websocket=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E5=99=A8=E7=AB=AF=E3=80=90=E9=85=8D=E7=BD=AE=E6=8B=89=E5=8F=96?= =?UTF-8?q?=E3=80=91=E5=92=8C=E3=80=90=E9=87=8D=E5=90=AF=E3=80=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../common/exception/RenExceptionHandler.java | 11 +- .../java/xiaozhi/common/utils/DateUtils.java | 15 + .../ServerSideManageController.java | 115 ++++ .../sys/controller/SysParamsController.java | 28 +- .../modules/sys/dto/EmitSeverActionDTO.java | 29 + .../sys/dto/ServerActionPayloadDTO.java | 39 ++ .../sys/dto/ServerActionResponseDTO.java | 34 + .../modules/sys/enums/ServerActionEnum.java | 43 ++ .../sys/enums/ServerActionResponseEnum.java | 49 ++ .../sys/utils/WebSocketClientManager.java | 315 +++++++++ main/manager-web/src/apis/module/admin.js | 32 + main/manager-web/src/components/HeaderBar.vue | 6 + main/manager-web/src/router/index.js | 12 + .../src/views/ServerSideManager.vue | 597 ++++++++++++++++++ main/xiaozhi-server/core/connection.py | 6 +- main/xiaozhi-server/core/handle/textHandle.py | 12 +- 16 files changed, 1320 insertions(+), 23 deletions(-) create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java create mode 100644 main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java create mode 100644 main/manager-web/src/views/ServerSideManager.vue diff --git a/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java b/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java index f6b19d31..e8eaf462 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java +++ b/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java @@ -71,10 +71,15 @@ public class RenExceptionHandler { List allErrors = ex.getBindingResult().getAllErrors(); String errorMsg = allErrors.stream() .filter(Objects::nonNull) - .map(DefaultMessageSourceResolvable::getDefaultMessage) + .map(err -> { + String msg = err.getDefaultMessage(); + return (msg != null && !msg.trim().isEmpty()) ? msg : null; + }) + .filter(Objects::nonNull) .findFirst() - .orElse(""); - return new Result().error(400, errorMsg); + .orElse("请求参数错误！"); + + return new Result().error(ErrorCode.PARAM_VALUE_NULL, errorMsg); } } \ No newline at end of file diff --git a/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java b/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java index bfa7afed..800733c1 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java +++ b/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java @@ -21,6 +21,8 @@ public class DateUtils { * 时间格式(yyyy-MM-dd HH:mm:ss) */ public final static String DATE_TIME_PATTERN = "yyyy-MM-dd HH:mm:ss"; + public final static String DATE_TIME_MILLIS_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS"; + /** * 日期格式化 日期格式为：yyyy-MM-dd @@ -63,6 +65,19 @@ public class DateUtils { return null; } + + public static String getDateTimeNow() { + return getDateTimeNow(DATE_TIME_PATTERN); + } + + public static String getDateTimeNow(String pattern) { + return format(new Date(), pattern); + } + + public static String millsToSecond(long mills) { + return String.format("%.3f", mills / 1000.0); + } + /** * 获取简短的时间字符串：10秒前返回刚刚，多少秒前，几小时前，超过一周返回年月日时分秒 * @param date diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java new file mode 100644 index 00000000..650d0a1e --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java @@ -0,0 +1,115 @@ +package xiaozhi.modules.sys.controller; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.validation.Valid; +import lombok.AllArgsConstructor; +import org.apache.commons.lang3.StringUtils; +import org.apache.shiro.authz.annotation.RequiresPermissions; +import org.springframework.web.bind.annotation.*; +import org.springframework.web.socket.WebSocketHttpHeaders; +import xiaozhi.common.annotation.LogOperation; +import xiaozhi.common.constant.Constant; +import xiaozhi.common.exception.RenException; +import xiaozhi.common.utils.Result; +import xiaozhi.modules.sys.dto.EmitSeverActionDTO; +import xiaozhi.modules.sys.dto.ServerActionPayloadDTO; +import xiaozhi.modules.sys.dto.ServerActionResponseDTO; +import xiaozhi.modules.sys.enums.ServerActionEnum; +import xiaozhi.modules.sys.service.SysParamsService; +import xiaozhi.modules.sys.utils.WebSocketClientManager; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * 服务端管理控制器 + */ +@RestController +@RequestMapping("admin/server") +@Tag(name = "服务端管理") +@AllArgsConstructor +public class ServerSideManageController +{ + private final SysParamsService sysParamsService; + private static final ObjectMapper objectMapper; + static { + objectMapper = new ObjectMapper(); + objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); + } + + @Operation(summary = "获取Ws服务端列表") + @GetMapping("/server-list") + @RequiresPermissions("sys:role:superAdmin") + public Result> getWsServerList() { + String cachedWs = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true); + return new Result>().ok(Arrays.asList(cachedWs.split(";"))); + } + + @Operation(summary = "通知python服务端更新配置") + @PostMapping("/emit-action") + @LogOperation("通知python服务端更新配置") +// @RequiresPermissions("sys:role:superAdmin") + public Result emitServerAction(@RequestBody @Valid EmitSeverActionDTO emitSeverActionDTO) { + if (emitSeverActionDTO.getAction() == null) { + throw new RenException("无效服务端操作"); + } + String wsText = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true); + if (StringUtils.isBlank(wsText)) { + throw new RenException("未配置服务端WebSocket地址"); + } + String targetWs = emitSeverActionDTO.getTargetWs(); + String[] wsList = wsText.split(";"); + // 找到需要发起的 + if (StringUtils.isBlank(targetWs) || !Arrays.asList(wsList).contains(targetWs)) { + throw new RenException("目标WebSocket地址不存在"); + } + return new Result().ok(emitServerActionByWs(targetWs, emitSeverActionDTO.getAction())); + } + + private Boolean emitServerActionByWs(String targetWsUri, ServerActionEnum actionEnum) { + if (StringUtils.isBlank(targetWsUri) || actionEnum == null) { + return false; + } + String serverSK = sysParamsService.getValue(Constant.SERVER_SECRET, true); + WebSocketHttpHeaders headers = new WebSocketHttpHeaders(); + headers.add("Authorization", "Bearer " + serverSK); + headers.add("device-id", serverSK); + + try (WebSocketClientManager client = new WebSocketClientManager.Builder() + .connectTimeout(3, TimeUnit.SECONDS) + .maxSessionDuration(5, TimeUnit.SECONDS) + .uri(targetWsUri) + .headers(headers) + .build()) { + // 如果连接成功则发送一个json数据包并等待服务端响应 + client.sendJson( + ServerActionPayloadDTO.build( + actionEnum, + Map.of("secret", serverSK) + )); + // 等待服务端响应并持续监听信息 + client.listener((jsonText)-> { + if (StringUtils.isBlank(jsonText)) { + return false; + } + try { + return ServerActionResponseDTO.isSuccess(objectMapper.readValue(jsonText, ServerActionResponseDTO.class)); + } + catch (JsonProcessingException e) { + return false; + } + }); + } + catch (Exception e) { + // 捕获全部错误，由全局异常处理器返回 + throw new RenException("WebSocket连接失败或连接超时"); + } + return true; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java index e1774280..4860cf1c 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java @@ -111,7 +111,7 @@ public class SysParamsController { /** * 验证WebSocket地址列表 - * + * * @param urls WebSocket地址列表，以分号分隔 * @return 验证结果 */ @@ -143,6 +143,19 @@ public class SysParamsController { } } + @PostMapping("/delete") + @Operation(summary = "删除") + @LogOperation("删除") + @RequiresPermissions("sys:role:superAdmin") + public Result delete(@RequestBody String[] ids) { + // 效验数据 + AssertUtils.isArrayEmpty(ids, "id"); + + sysParamsService.delete(ids); + configService.getConfig(false); + return new Result(); + } + /** * 验证OTA地址 */ @@ -182,17 +195,4 @@ public class SysParamsController { throw new RenException("OTA接口验证失败：" + e.getMessage()); } } - - @PostMapping("/delete") - @Operation(summary = "删除") - @LogOperation("删除") - @RequiresPermissions("sys:role:superAdmin") - public Result delete(@RequestBody String[] ids) { - // 效验数据 - AssertUtils.isArrayEmpty(ids, "id"); - - sysParamsService.delete(ids); - configService.getConfig(false); - return new Result(); - } } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java new file mode 100644 index 00000000..888ef4ec --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java @@ -0,0 +1,29 @@ +package xiaozhi.modules.sys.dto; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.NotNull; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; +import xiaozhi.modules.sys.enums.ServerActionEnum; + +/** + * 发送python服务端操作DTO + * + * @Author CAIXYPROMISE + * @since 2025/5/15 4:55 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class EmitSeverActionDTO +{ + @Schema(description = "目标ws地址") + @NotEmpty(message = "目标ws地址不能为空") + private String targetWs; + + @Schema(description = "指定操作") + @NotNull(message = "操作不能为空") + private ServerActionEnum action; +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java new file mode 100644 index 00000000..cb158d25 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java @@ -0,0 +1,39 @@ +package xiaozhi.modules.sys.dto; + +import lombok.Data; +import xiaozhi.modules.sys.enums.ServerActionEnum; + +import java.util.Map; + +/** + * 服务端动作DTO + * + * @Author CAIXYPROMISE + * @since 2025/5/15 5:10 + */ +@Data +public class ServerActionPayloadDTO +{ + /** + * 类型（智控台发往服务端的都是server） + */ + private String type; + /** + * 动作 + */ + private ServerActionEnum action; + /** + * 内容 + */ + private Map content; + + public static ServerActionPayloadDTO build(ServerActionEnum action, Map content) { + ServerActionPayloadDTO serverActionPayloadDTO = new ServerActionPayloadDTO(); + serverActionPayloadDTO.setAction(action); + serverActionPayloadDTO.setContent(content); + serverActionPayloadDTO.setType("server"); + return serverActionPayloadDTO; + } + // 私有化 + private ServerActionPayloadDTO() {} +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java new file mode 100644 index 00000000..6286b74a --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java @@ -0,0 +1,34 @@ +package xiaozhi.modules.sys.dto; + +import lombok.Data; +import xiaozhi.modules.sys.enums.ServerActionResponseEnum; + +import java.util.Map; + +/** + * 服务端动作响应体 + */ +@Data +public class ServerActionResponseDTO +{ + private ServerActionResponseEnum status; + private String message; + private String type; + private Map content; // 后续这个字段可以移除，并把这个类作为基类，针对业务写自己的content类型 + public static final String DEFAULT_TYPE_FORM_SERVER = "server"; + + public static Boolean isSuccess(ServerActionResponseDTO actionResponseDTO) { + System.out.println(actionResponseDTO); + if (actionResponseDTO == null) { + return false; + } + if (actionResponseDTO.getStatus() == null || !actionResponseDTO.getStatus().equals(ServerActionResponseEnum.SUCCESS)) { + return false; + } + Object actionType = actionResponseDTO.getContent().get("action"); + if (actionType == null) { + return false; + } + return actionResponseDTO.getType() != null && actionResponseDTO.getType().equals(DEFAULT_TYPE_FORM_SERVER); + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java new file mode 100644 index 00000000..33babf45 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java @@ -0,0 +1,43 @@ +package xiaozhi.modules.sys.enums; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import xiaozhi.common.exception.RenException; + +/** + * 服务端动作枚举 + * + * @Author CAIXYPROMISE + * @since 2025/5/15 4:57 + */ +public enum ServerActionEnum +{ + RESTART("restart"), + UPDATE_CONFIG("update_config"); + + private final String value; + + ServerActionEnum(String value) + { + this.value = value; + } + + @JsonValue + public String getValue() + { + return value; + } + + @JsonCreator + public static ServerActionEnum fromValue(String value) + { + for (ServerActionEnum action : ServerActionEnum.values()) + { + if (action.value.equalsIgnoreCase(value)) + { + return action; + } + } + return null; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java new file mode 100644 index 00000000..e1b54a66 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java @@ -0,0 +1,49 @@ +package xiaozhi.modules.sys.enums; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import lombok.Getter; +import org.apache.commons.lang3.StringUtils; + +/** + * 服务端调用响应枚举 + * + * @Author CAIXYPROMISE + * @since 2025/5/15 5:50 + */ +public enum ServerActionResponseEnum +{ + SUCCESS("success"), FAIL("fail"); + private final String value; + + ServerActionResponseEnum(String value) { + this.value = value; + } + + @JsonValue + public String getValue() + { + return value; + } + + @JsonCreator + public static ServerActionResponseEnum fromValue(String value) { + ServerActionResponseEnum byValue = getByValue(value); + if (byValue == null) { + throw new IllegalArgumentException("Unknown enum value: " + value); + } + return byValue; + } + + public static ServerActionResponseEnum getByValue(String value) { + if (StringUtils.isBlank(value)) { + return null; + } + for (ServerActionResponseEnum action : ServerActionResponseEnum.values()) { + if (action.value.equals(value)) { + return action; + } + } + return null; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java new file mode 100644 index 00000000..0c49aead --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java @@ -0,0 +1,315 @@ +package xiaozhi.modules.sys.utils; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.util.StopWatch; +import org.springframework.web.socket.*; +import org.springframework.web.socket.client.standard.StandardWebSocketClient; +import org.springframework.web.socket.handler.AbstractWebSocketHandler; +import xiaozhi.common.utils.DateUtils; + +import java.io.Closeable; +import java.io.IOException; +import java.net.URI; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import java.util.function.Predicate; + +/** + * WebSocketClientResource：支持 try-with-resources 模式 + */ +@Slf4j +public class WebSocketClientManager implements Closeable +{ + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + // 全局回调线程池 + private static final ExecutorService CALLBACK_EXECUTOR = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() { + private final AtomicInteger cnt = new AtomicInteger(); + + public Thread newThread(Runnable r) + { + Thread t = new Thread(r, "ws-callback-" + cnt.getAndIncrement()); + t.setDaemon(true); + return t; + } + }); + + private volatile WebSocketSession session; + private final BlockingQueue textMessageQueue; + private final BlockingQueue binaryMessageQueue; + private final CompletableFuture errorFuture; + private final long maxSessionDuration; + private final TimeUnit maxSessionDurationUnit; + + private volatile Consumer onText; + private volatile Consumer onBinary; + private volatile Consumer onError; + + private final String uri; + private final WebSocketHttpHeaders headers; + private final long connectTimeout; + private final TimeUnit connectUnit; + private final int queueCapacity; + + // 私有构造，仅由 Builder 调用 + private WebSocketClientManager(Builder b) { + this.uri = b.uri; + this.headers = b.headers != null ? b.headers : new WebSocketHttpHeaders(); + this.connectTimeout = b.connectTimeout; + this.connectUnit = b.connectUnit; + this.maxSessionDuration = b.maxSessionDuration; + this.maxSessionDurationUnit = b.maxSessionDurationUnit; + this.queueCapacity = b.queueCapacity; + this.textMessageQueue = new LinkedBlockingQueue<>(queueCapacity); + this.binaryMessageQueue = new LinkedBlockingQueue<>(queueCapacity); + this.errorFuture = new CompletableFuture<>(); + } + + public static WebSocketClientManager build(Builder b) throws InterruptedException, ExecutionException, TimeoutException, IOException { + WebSocketClientManager ws = new WebSocketClientManager(b); + StandardWebSocketClient client = new StandardWebSocketClient(); + CompletableFuture future = client.execute(ws.new InternalHandler(b.uri), b.headers, URI.create(b.uri)); + WebSocketSession sess = future.get(b.connectTimeout, b.connectUnit); + if (sess == null || !sess.isOpen()) + { + throw new IOException("握手失败或会话未打开"); + } + ws.session = sess; + return ws; + } + + /** + * 发送 Text + */ + public void sendText(String text) throws IOException { + session.sendMessage(new TextMessage(text)); + } + + public void sendBinary(byte[] data) throws IOException { + session.sendMessage(new BinaryMessage(data)); + } + + public void sendJson(Object payload) throws IOException { + String json = OBJECT_MAPPER.writeValueAsString(payload); + session.sendMessage(new TextMessage(json)); + } + + + + private List listenerCustom( + BlockingQueue queue, + Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException + { + List collected = new ArrayList<>(); + long deadline = System.currentTimeMillis() + maxSessionDurationUnit.toMillis(maxSessionDuration); + + while (true) { + if (errorFuture.isDone()) { + errorFuture.get(); + } + + long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + throw new TimeoutException("等待批量消息超时"); + } + + T msg = queue.poll(remaining, TimeUnit.MILLISECONDS); + if (msg == null) { + throw new TimeoutException("等待批量消息超时"); + } + + collected.add(msg); + if (predicate.test(msg)) { + break; + } + } + close(); + return collected; + } + + /** + * 同步接收多条消息，直到 predicate 为 true 或超时抛异常； + * @return 返回监听期间的所有消息列表 + */ + public List listener(Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException + { + return listenerCustom(textMessageQueue, predicate); + } + + public List listenerBinary(Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException + { + return listenerCustom(binaryMessageQueue, predicate); + } + + /** + * 注册文本回调 + */ + public WebSocketClientManager onText(Consumer c) { + this.onText = c; + return this; + } + + /** + * 注册二进制回调 + */ + public WebSocketClientManager onBinary(Consumer c) { + this.onBinary = c; + return this; + } + + /** + * 注册错误回调 + */ + public WebSocketClientManager onError(Consumer c) { + this.onError = c; + return this; + } + + /** + * 关闭会话，try-with-resources / finally 自动调用 + */ + @Override + public void close() { + try { + if (session != null && session.isOpen()) { + session.close(CloseStatus.NORMAL); + } + } + catch (IOException ignored) {} + textMessageQueue.clear(); + binaryMessageQueue.clear(); + errorFuture.completeExceptionally(new IOException("WebSocket 已关闭")); + } + + private class InternalHandler extends AbstractWebSocketHandler { + private final String targetUri; + private final StopWatch stopWatch; + + InternalHandler(String targetUri) { + this.targetUri = targetUri; + this.stopWatch = new StopWatch(); + } + + /** + * 连接建立时回调 + */ + @Override + public void afterConnectionEstablished(WebSocketSession session) { + // 保存会话 + WebSocketClientManager.this.session = session; + this.stopWatch.start(); + log.info("ws连接成功, 目标URI: {}, 连接时间: {}", targetUri, DateUtils.getDateTimeNow(DateUtils.DATE_TIME_MILLIS_PATTERN)); + } + + /** + * 处理文本消息 + */ + @Override + protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { + String payload = message.getPayload(); + // 入队 + textMessageQueue.offer(payload); + // 回调用户注册的 onText + if (onText != null) { + CALLBACK_EXECUTOR.submit(() -> onText.accept(payload)); + } + } + + /** + * 处理二进制消息 + */ + @Override + protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception { + ByteBuffer buf = message.getPayload(); + byte[] data = new byte[buf.remaining()]; + buf.get(data); + // 入队 + binaryMessageQueue.offer(data); + // 回调用户注册的 onBinary + if (onBinary != null) { + CALLBACK_EXECUTOR.submit(() -> onBinary.accept(data)); + } + } + + /** + * 传输错误时回调 + */ + @Override + public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { + super.handleTransportError(session, exception); + // 保持原有逻辑：完成 errorFuture、回调 onError、关闭会话、异步通知连接失败 + errorFuture.completeExceptionally(exception); + if (onError != null) { + CALLBACK_EXECUTOR.submit(() -> onError.accept(exception)); + } + session.close(CloseStatus.SERVER_ERROR); + } + + /** + * 连接关闭时回调 + */ + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { + super.afterConnectionClosed(session, status); + if (stopWatch.isRunning()) { + stopWatch.stop(); + } + log.info("ws连接关闭, 目标URI: {}, 关闭时间: {}, 连接总时长: {}s", + targetUri, DateUtils.getDateTimeNow(DateUtils.DATE_TIME_MILLIS_PATTERN), DateUtils.millsToSecond(stopWatch.getTotalTimeMillis())); + } + } + + public static class Builder { + private String uri; // 目标 WS URI + private long connectTimeout = 3; // 请求连接等待时间 + private TimeUnit connectUnit = TimeUnit.SECONDS; // 请求连接等待时间单位 + private long maxSessionDuration = 5; // 最大连线时间，默认5秒 + private TimeUnit maxSessionDurationUnit = TimeUnit.SECONDS; // 最大连线时间单位 + private int queueCapacity = 100; // 消息队列容量 + private WebSocketHttpHeaders headers; // 请求头 + + /** + * 目标 WS URI + */ + public Builder uri(String uri) { + this.uri = Objects.requireNonNull(uri); + return this; + } + + public Builder headers(WebSocketHttpHeaders h) { + this.headers = h; + return this; + } + + public Builder connectTimeout(long t, TimeUnit u) { + this.connectTimeout = t; + this.connectUnit = u; + return this; + } + + public Builder maxSessionDuration(long t, TimeUnit u) { + this.maxSessionDuration = t; + this.maxSessionDurationUnit = u; + return this; + } + + public Builder queueCapacity(int c) { + this.queueCapacity = c; + return this; + } + + public WebSocketClientManager build() throws InterruptedException, ExecutionException, TimeoutException, IOException { + return WebSocketClientManager.build(this); + } + + } +} diff --git a/main/manager-web/src/apis/module/admin.js b/main/manager-web/src/apis/module/admin.js index 92c344e2..47cfa9f9 100644 --- a/main/manager-web/src/apis/module/admin.js +++ b/main/manager-web/src/apis/module/admin.js @@ -130,5 +130,37 @@ export default { }) }).send() }, + // 获取ws服务端列表 + getWsServerList(params, callback) { + RequestService.sendRequest() + .url(`${getServiceUrl()}/admin/server/server-list`) + .method('GET') + .success((res) => { + RequestService.clearRequestTime() + callback(res) + }) + .fail((err) => { + console.error('获取ws服务端列表失败:', err) + RequestService.reAjaxFun(() => { + this.getWsServerList(params, callback) + }) + }).send(); + }, + // 发送ws服务器动作指令 + sendWsServerAction(data, callback) { + RequestService.sendRequest() + .url(`${getServiceUrl()}/admin/server/emit-action`) + .method('POST') + .data(data) + .success((res) => { + RequestService.clearRequestTime() + callback(res) + }) + .fail((err) => { + RequestService.reAjaxFun(() => { + this.sendWsServerAction(data, callback) + }) + }).send(); + } } diff --git a/main/manager-web/src/components/HeaderBar.vue b/main/manager-web/src/components/HeaderBar.vue index 842441ce..3843151a 100644 --- a/main/manager-web/src/components/HeaderBar.vue +++ b/main/manager-web/src/components/HeaderBar.vue @@ -52,6 +52,9 @@ 供应器管理 + + 服务端管理 + @@ -145,6 +148,9 @@ export default { goProviderManagement() { this.$router.push('/provider-management') }, + goServerSideManagement() { + this.$router.push('/server-side-management') + }, // 获取用户信息 fetchUserInfo() { userApi.getUserInfo(({ data }) => { diff --git a/main/manager-web/src/router/index.js b/main/manager-web/src/router/index.js index 49ee7e4c..7e304f76 100644 --- a/main/manager-web/src/router/index.js +++ b/main/manager-web/src/router/index.js @@ -80,6 +80,18 @@ const routes = [ title: '参数管理' } }, + + { + path: '/server-side-management', + name: 'ServerSideManager', + component: function () { + return import('../views/ServerSideManager.vue') + }, + meta: { + requiresAuth: true, + title: '服务端管理' + } + }, { path: '/ota-management', name: 'OtaManagement', diff --git a/main/manager-web/src/views/ServerSideManager.vue b/main/manager-web/src/views/ServerSideManager.vue new file mode 100644 index 00000000..f0118df9 --- /dev/null +++ b/main/manager-web/src/views/ServerSideManager.vue @@ -0,0 +1,597 @@ + + + + + diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 0abb3cec..06883983 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -273,9 +273,10 @@ class ConnectionHandler: await self.websocket.send( json.dumps( { - "type": "server_response", + "type": "server", "status": "success", "message": "服务器重启中...", + "content": {'action': "restart"} } ) ) @@ -302,9 +303,10 @@ class ConnectionHandler: await self.websocket.send( json.dumps( { - "type": "server_response", + "type": "server", "status": "error", "message": f"Restart failed: {str(e)}", + 'content': {'action': "restart"} } ) ) diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 79874f1c..4c44ece4 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -95,9 +95,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "error", "message": "无法获取服务器实例", + "content": {"action": "update_config"} } ) ) @@ -107,9 +108,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "error", "message": "更新服务器配置失败", + "content": {"action": "update_config"} } ) ) @@ -119,9 +121,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "success", "message": "配置更新成功", + "content": {"action": "update_config"} } ) ) @@ -130,9 +133,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "error", "message": f"更新配置失败: {str(e)}", + "content": {"action": "update_config"} } ) ) From b8da0714c21d03915fac16c4aadd6617aa51ece1 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Tue, 20 May 2025 22:58:42 +0800 Subject: [PATCH 2/3] =?UTF-8?q?update:=E4=BC=98=E5=8C=96=E6=97=A0=E7=94=A8?= =?UTF-8?q?=E7=BB=84=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ServerSideManageController.java | 50 +++--- .../src/views/ServerSideManager.vue | 166 +++--------------- main/xiaozhi-server/core/connection.py | 39 +--- main/xiaozhi-server/core/handle/textHandle.py | 20 ++- main/xiaozhi-server/core/utils/util.py | 37 +++- 5 files changed, 102 insertions(+), 210 deletions(-) diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java index 650d0a1e..7ee84d36 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java @@ -1,16 +1,28 @@ package xiaozhi.modules.sys.controller; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang3.StringUtils; +import org.apache.shiro.authz.annotation.RequiresPermissions; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.socket.WebSocketHttpHeaders; + import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; + import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; import jakarta.validation.Valid; import lombok.AllArgsConstructor; -import org.apache.commons.lang3.StringUtils; -import org.apache.shiro.authz.annotation.RequiresPermissions; -import org.springframework.web.bind.annotation.*; -import org.springframework.web.socket.WebSocketHttpHeaders; import xiaozhi.common.annotation.LogOperation; import xiaozhi.common.constant.Constant; import xiaozhi.common.exception.RenException; @@ -22,11 +34,6 @@ import xiaozhi.modules.sys.enums.ServerActionEnum; import xiaozhi.modules.sys.service.SysParamsService; import xiaozhi.modules.sys.utils.WebSocketClientManager; -import java.util.Arrays; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; - /** * 服务端管理控制器 */ @@ -34,11 +41,10 @@ import java.util.concurrent.TimeUnit; @RequestMapping("admin/server") @Tag(name = "服务端管理") @AllArgsConstructor -public class ServerSideManageController -{ +public class ServerSideManageController { private final SysParamsService sysParamsService; private static final ObjectMapper objectMapper; - static { + static { objectMapper = new ObjectMapper(); objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); } @@ -54,7 +60,7 @@ public class ServerSideManageController @Operation(summary = "通知python服务端更新配置") @PostMapping("/emit-action") @LogOperation("通知python服务端更新配置") -// @RequiresPermissions("sys:role:superAdmin") + @RequiresPermissions("sys:role:superAdmin") public Result emitServerAction(@RequestBody @Valid EmitSeverActionDTO emitSeverActionDTO) { if (emitSeverActionDTO.getAction() == null) { throw new RenException("无效服务端操作"); @@ -78,8 +84,8 @@ public class ServerSideManageController } String serverSK = sysParamsService.getValue(Constant.SERVER_SECRET, true); WebSocketHttpHeaders headers = new WebSocketHttpHeaders(); - headers.add("Authorization", "Bearer " + serverSK); - headers.add("device-id", serverSK); + headers.add("device-id", UUID.randomUUID().toString()); + headers.add("client-id", UUID.randomUUID().toString()); try (WebSocketClientManager client = new WebSocketClientManager.Builder() .connectTimeout(3, TimeUnit.SECONDS) @@ -91,22 +97,20 @@ public class ServerSideManageController client.sendJson( ServerActionPayloadDTO.build( actionEnum, - Map.of("secret", serverSK) - )); + Map.of("secret", serverSK))); // 等待服务端响应并持续监听信息 - client.listener((jsonText)-> { + client.listener((jsonText) -> { if (StringUtils.isBlank(jsonText)) { return false; } try { - return ServerActionResponseDTO.isSuccess(objectMapper.readValue(jsonText, ServerActionResponseDTO.class)); - } - catch (JsonProcessingException e) { + return ServerActionResponseDTO + .isSuccess(objectMapper.readValue(jsonText, ServerActionResponseDTO.class)); + } catch (JsonProcessingException e) { return false; } }); - } - catch (Exception e) { + } catch (Exception e) { // 捕获全部错误，由全局异常处理器返回 throw new RenException("WebSocket连接失败或连接超时"); } diff --git a/main/manager-web/src/views/ServerSideManager.vue b/main/manager-web/src/views/ServerSideManager.vue index f0118df9..973fcd6f 100644 --- a/main/manager-web/src/views/ServerSideManager.vue +++ b/main/manager-web/src/views/ServerSideManager.vue @@ -1,14 +1,9 @@