BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 158576740dbb1d12ecead6c6853d51c857109984

Authored by puhui999
1 parent dfd1f90a

feat：【iot】新增 WebSocket 的 {@link IotDataRuleAction} 实现类

Inline Side-by-side
Showing 2 changed files with 258 additions and 0 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java 0 → 100644
  View file @1585767
  1 +package cn.iocoder.yudao.module.iot.service.rule.data.action;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  4 +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkWebSocketConfig;
  5 +import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
  6 +import cn.iocoder.yudao.module.iot.service.rule.data.action.websocket.IotWebSocketClient;
  7 +import lombok.extern.slf4j.Slf4j;
  8 +import org.springframework.stereotype.Component;
  9 +
  10 +/**
  11 + * WebSocket 的 {@link IotDataRuleAction} 实现类
  12 + * <p>
  13 + * 负责将设备消息发送到外部 WebSocket 服务器
  14 + * 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
  15 + * 使用连接池管理 WebSocket 连接，提高性能和资源利用率
  16 + *
  17 + * @author HUIHUI
  18 + */
  19 +@Component
  20 +@Slf4j
  21 +public class IotWebSocketDataRuleAction extends
  22 + IotDataRuleCacheableAction<IotDataSinkWebSocketConfig, IotWebSocketClient> {
  23 +
  24 + @Override
  25 + public Integer getType() {
  26 + return IotDataSinkTypeEnum.WEBSOCKET.getType();
  27 + }
  28 +
  29 + @Override
  30 + protected IotWebSocketClient initProducer(IotDataSinkWebSocketConfig config) throws Exception {
  31 + // 1.1 参数校验
  32 + if (config.getServerUrl() == null || config.getServerUrl().trim().isEmpty()) {
  33 + throw new IllegalArgumentException("WebSocket 服务器地址不能为空");
  34 + }
  35 + if (!config.getServerUrl().startsWith("ws://") && !config.getServerUrl().startsWith("wss://")) {
  36 + throw new IllegalArgumentException("WebSocket 服务器地址必须以 ws:// 或 wss:// 开头");
  37 + }
  38 +
  39 + // 2.1 创建 WebSocket 客户端
  40 + IotWebSocketClient webSocketClient = new IotWebSocketClient(
  41 + config.getServerUrl(),
  42 + config.getConnectTimeoutMs(),
  43 + config.getSendTimeoutMs(),
  44 + config.getDataFormat()
  45 + );
  46 + // 2.2 连接服务器
  47 + webSocketClient.connect();
  48 + log.info("[initProducer][WebSocket 客户端创建并连接成功，服务器: {}，数据格式: {}]",
  49 + config.getServerUrl(), config.getDataFormat());
  50 + return webSocketClient;
  51 + }
  52 +
  53 + @Override
  54 + protected void closeProducer(IotWebSocketClient producer) throws Exception {
  55 + if (producer != null) {
  56 + producer.close();
  57 + }
  58 + }
  59 +
  60 + @Override
  61 + protected void execute(IotDeviceMessage message, IotDataSinkWebSocketConfig config) throws Exception {
  62 + try {
  63 + // 1.1 获取或创建 WebSocket 客户端
  64 + IotWebSocketClient webSocketClient = getProducer(config);
  65 + // 1.2 检查连接状态，如果断开则重新连接
  66 + if (!webSocketClient.isConnected()) {
  67 + log.warn("[execute][WebSocket 连接已断开，尝试重新连接，服务器: {}]", config.getServerUrl());
  68 + webSocketClient.connect();
  69 + }
  70 +
  71 + // 2.1 发送消息
  72 + webSocketClient.sendMessage(message);
  73 + // 2.2 记录发送成功日志
  74 + log.info("[execute][message({}) config({}) 发送成功，WebSocket 服务器: {}]",
  75 + message, config, config.getServerUrl());
  76 + } catch (Exception e) {
  77 + log.error("[execute][message({}) config({}) 发送失败，WebSocket 服务器: {}]",
  78 + message, config, config.getServerUrl(), e);
  79 + throw e;
  80 + }
  81 + }
  82 +
  83 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java 0 → 100644
  View file @1585767
  1 +package cn.iocoder.yudao.module.iot.service.rule.data.action.websocket;
  2 +
  3 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  4 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  5 +import lombok.extern.slf4j.Slf4j;
  6 +
  7 +import java.net.URI;
  8 +import java.net.http.HttpClient;
  9 +import java.net.http.WebSocket;
  10 +import java.time.Duration;
  11 +import java.util.concurrent.CompletableFuture;
  12 +import java.util.concurrent.CompletionStage;
  13 +import java.util.concurrent.TimeUnit;
  14 +import java.util.concurrent.atomic.AtomicBoolean;
  15 +
  16 +/**
  17 + * IoT WebSocket 客户端
  18 + * <p>
  19 + * 负责与外部 WebSocket 服务器建立连接并发送设备消息
  20 + * 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
  21 + * 基于 Java 11+ 内置的 java.net.http.WebSocket 实现
  22 + *
  23 + * @author HUIHUI
  24 + */
  25 +@Slf4j
  26 +public class IotWebSocketClient implements WebSocket.Listener {
  27 +
  28 + private final String serverUrl;
  29 + private final Integer connectTimeoutMs;
  30 + private final Integer sendTimeoutMs;
  31 + private final String dataFormat;
  32 +
  33 + private WebSocket webSocket;
  34 + private final AtomicBoolean connected = new AtomicBoolean(false);
  35 + private final StringBuilder messageBuffer = new StringBuilder();
  36 +
  37 + public IotWebSocketClient(String serverUrl, Integer connectTimeoutMs, Integer sendTimeoutMs, String dataFormat) {
  38 + this.serverUrl = serverUrl;
  39 + this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : 5000;
  40 + this.sendTimeoutMs = sendTimeoutMs != null ? sendTimeoutMs : 10000;
  41 + this.dataFormat = dataFormat != null ? dataFormat : "JSON";
  42 + }
  43 +
  44 + /**
  45 + * 连接到 WebSocket 服务器
  46 + */
  47 + public void connect() throws Exception {
  48 + if (connected.get()) {
  49 + log.warn("[connect][WebSocket 客户端已经连接，无需重复连接]");
  50 + return;
  51 + }
  52 +
  53 + try {
  54 + HttpClient httpClient = HttpClient.newBuilder()
  55 + .connectTimeout(Duration.ofMillis(connectTimeoutMs))
  56 + .build();
  57 +
  58 + CompletableFuture<WebSocket> future = httpClient.newWebSocketBuilder()
  59 + .connectTimeout(Duration.ofMillis(connectTimeoutMs))
  60 + .buildAsync(URI.create(serverUrl), this);
  61 +
  62 + // 等待连接完成
  63 + webSocket = future.get(connectTimeoutMs, TimeUnit.MILLISECONDS);
  64 + connected.set(true);
  65 + log.info("[connect][WebSocket 客户端连接成功，服务器地址: {}]", serverUrl);
  66 + } catch (Exception e) {
  67 + close();
  68 + log.error("[connect][WebSocket 客户端连接失败，服务器地址: {}]", serverUrl, e);
  69 + throw e;
  70 + }
  71 + }
  72 +
  73 + @Override
  74 + public void onOpen(WebSocket webSocket) {
  75 + log.debug("[onOpen][WebSocket 连接已打开]");
  76 + webSocket.request(1);
  77 + }
  78 +
  79 + @Override
  80 + public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
  81 + messageBuffer.append(data);
  82 + if (last) {
  83 + log.debug("[onText][收到 WebSocket 消息: {}]", messageBuffer);
  84 + messageBuffer.setLength(0);
  85 + }
  86 + webSocket.request(1);
  87 + return null;
  88 + }
  89 +
  90 + @Override
  91 + public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
  92 + connected.set(false);
  93 + log.info("[onClose][WebSocket 连接已关闭，状态码: {}，原因: {}]", statusCode, reason);
  94 + return null;
  95 + }
  96 +
  97 + @Override
  98 + public void onError(WebSocket webSocket, Throwable error) {
  99 + connected.set(false);
  100 + log.error("[onError][WebSocket 发生错误]", error);
  101 + }
  102 +
  103 + /**
  104 + * 发送设备消息
  105 + *
  106 + * @param message 设备消息
  107 + * @throws Exception 发送异常
  108 + */
  109 + public void sendMessage(IotDeviceMessage message) throws Exception {
  110 + if (!connected.get() || webSocket == null) {
  111 + throw new IllegalStateException("WebSocket 客户端未连接");
  112 + }
  113 +
  114 + try {
  115 + String messageData;
  116 + if ("JSON".equalsIgnoreCase(dataFormat)) {
  117 + messageData = JsonUtils.toJsonString(message);
  118 + } else {
  119 + messageData = message.toString();
  120 + }
  121 +
  122 + // 发送消息并等待完成
  123 + CompletableFuture<WebSocket> future = webSocket.sendText(messageData, true);
  124 + future.get(sendTimeoutMs, TimeUnit.MILLISECONDS);
  125 + log.debug("[sendMessage][发送消息成功，设备 ID: {}，消息长度: {}]",
  126 + message.getDeviceId(), messageData.length());
  127 + } catch (Exception e) {
  128 + log.error("[sendMessage][发送消息失败，设备 ID: {}]", message.getDeviceId(), e);
  129 + throw e;
  130 + }
  131 + }
  132 +
  133 + /**
  134 + * 关闭连接
  135 + */
  136 + public void close() {
  137 + if (!connected.get() && webSocket == null) {
  138 + return;
  139 + }
  140 +
  141 + try {
  142 + if (webSocket != null) {
  143 + webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "客户端主动关闭")
  144 + .orTimeout(5, TimeUnit.SECONDS)
  145 + .exceptionally(e -> {
  146 + log.warn("[close][发送关闭帧失败]", e);
  147 + return null;
  148 + });
  149 + }
  150 + connected.set(false);
  151 + log.info("[close][WebSocket 客户端连接已关闭，服务器地址: {}]", serverUrl);
  152 + } catch (Exception e) {
  153 + log.error("[close][关闭 WebSocket 客户端连接异常]", e);
  154 + }
  155 + }
  156 +
  157 + /**
  158 + * 检查连接状态
  159 + *
  160 + * @return 是否已连接
  161 + */
  162 + public boolean isConnected() {
  163 + return connected.get() && webSocket != null;
  164 + }
  165 +
  166 + @Override
  167 + public String toString() {
  168 + return "IotWebSocketClient{" +
  169 + "serverUrl='" + serverUrl + '\'' +
  170 + ", dataFormat='" + dataFormat + '\'' +
  171 + ", connected=" + connected.get() +
  172 + '}';
  173 + }
  174 +
  175 +}