mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-23 15:43:54 +08:00
feat: 新增智控台管理websocket服务器端【配置拉取】和【重启】
This commit is contained in:
caixypromise
@@ -71,10 +71,15 @@ public class RenExceptionHandler {
|
||||
List<ObjectError> 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<Void>().error(400, errorMsg);
|
||||
.orElse("请求参数错误！");
|
||||
|
||||
return new Result<Void>().error(ErrorCode.PARAM_VALUE_NULL, errorMsg);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+115
@@ -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<List<String>> getWsServerList() {
|
||||
String cachedWs = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true);
|
||||
return new Result<List<String>>().ok(Arrays.asList(cachedWs.split(";")));
|
||||
}
|
||||
|
||||
@Operation(summary = "通知python服务端更新配置")
|
||||
@PostMapping("/emit-action")
|
||||
@LogOperation("通知python服务端更新配置")
|
||||
// @RequiresPermissions("sys:role:superAdmin")
|
||||
public Result<Boolean> 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<Boolean>().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;
|
||||
}
|
||||
}
|
||||
+14
-14
@@ -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<Void> delete(@RequestBody String[] ids) {
|
||||
// 效验数据
|
||||
AssertUtils.isArrayEmpty(ids, "id");
|
||||
|
||||
sysParamsService.delete(ids);
|
||||
configService.getConfig(false);
|
||||
return new Result<Void>();
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证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<Void> delete(@RequestBody String[] ids) {
|
||||
// 效验数据
|
||||
AssertUtils.isArrayEmpty(ids, "id");
|
||||
|
||||
sysParamsService.delete(ids);
|
||||
configService.getConfig(false);
|
||||
return new Result<Void>();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<String, Object> content;
|
||||
|
||||
public static ServerActionPayloadDTO build(ServerActionEnum action, Map<String, Object> content) {
|
||||
ServerActionPayloadDTO serverActionPayloadDTO = new ServerActionPayloadDTO();
|
||||
serverActionPayloadDTO.setAction(action);
|
||||
serverActionPayloadDTO.setContent(content);
|
||||
serverActionPayloadDTO.setType("server");
|
||||
return serverActionPayloadDTO;
|
||||
}
|
||||
// 私有化
|
||||
private ServerActionPayloadDTO() {}
|
||||
}
|
||||
@@ -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<String, Object> 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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
+49
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<String> textMessageQueue;
|
||||
private final BlockingQueue<byte[]> binaryMessageQueue;
|
||||
private final CompletableFuture<Void> errorFuture;
|
||||
private final long maxSessionDuration;
|
||||
private final TimeUnit maxSessionDurationUnit;
|
||||
|
||||
private volatile Consumer<String> onText;
|
||||
private volatile Consumer<byte[]> onBinary;
|
||||
private volatile Consumer<Throwable> 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<WebSocketSession> 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 <T> List<T> listenerCustom(
|
||||
BlockingQueue<T> queue,
|
||||
Predicate<T> predicate)
|
||||
throws InterruptedException, TimeoutException, ExecutionException
|
||||
{
|
||||
List<T> 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<String> listener(Predicate<String> predicate)
|
||||
throws InterruptedException, TimeoutException, ExecutionException
|
||||
{
|
||||
return listenerCustom(textMessageQueue, predicate);
|
||||
}
|
||||
|
||||
public List<byte[]> listenerBinary(Predicate<byte[]> predicate)
|
||||
throws InterruptedException, TimeoutException, ExecutionException
|
||||
{
|
||||
return listenerCustom(binaryMessageQueue, predicate);
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册文本回调
|
||||
*/
|
||||
public WebSocketClientManager onText(Consumer<String> c) {
|
||||
this.onText = c;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册二进制回调
|
||||
*/
|
||||
public WebSocketClientManager onBinary(Consumer<byte[]> c) {
|
||||
this.onBinary = c;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册错误回调
|
||||
*/
|
||||
public WebSocketClientManager onError(Consumer<Throwable> 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);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user