diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java deleted file mode 100644 index c7a0500030..0000000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java +++ /dev/null @@ -1,40 +0,0 @@ -package cn.iocoder.yudao.module.iot.mqttrpc.config; - -import lombok.Data; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.context.annotation.Configuration; - -@Data -@Configuration -@ConfigurationProperties(prefix = "mqtt") -public class MqttConfig { - /** - * MQTT 代理地址 - */ - private String broker; - - /** - * MQTT 用户名 - */ - private String username; - - /** - * MQTT 密码 - */ - private String password; - - /** - * MQTT 客户端 ID - */ - private String clientId; - - /** - * MQTT 请求主题 - */ - private String requestTopic; - - /** - * MQTT 响应主题前缀 - */ - private String responseTopicPrefix; -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/server/RpcServer.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/server/RpcServer.java deleted file mode 100644 index 90ce2a3875..0000000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/server/RpcServer.java +++ /dev/null @@ -1,100 +0,0 @@ -package cn.iocoder.yudao.module.iot.mqttrpc.server; - -import cn.hutool.core.lang.UUID; -import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcRequest; -import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcResponse; -import cn.iocoder.yudao.module.iot.mqttrpc.common.SerializationUtils; -import cn.iocoder.yudao.module.iot.mqttrpc.config.MqttConfig; -import lombok.extern.slf4j.Slf4j; -import org.eclipse.paho.client.mqttv3.*; -import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; -import org.springframework.stereotype.Service; - -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; -import java.util.HashMap; -import java.util.Map; - -// TODO @芋艿：server 逻辑，再瞅瞅； -// TODO @haohao：如果只写在 iot biz 里，那么后续 server => client 貌似不方便？微信再讨论下~； -@Service -@Slf4j -public class RpcServer { - - private final MqttConfig mqttConfig; - private final MqttClient mqttClient; - private final Map methodRegistry = new HashMap<>(); - - public RpcServer(MqttConfig mqttConfig) throws MqttException { - this.mqttConfig = mqttConfig; - this.mqttClient = new MqttClient(mqttConfig.getBroker(), "rpc-server-" + UUID.randomUUID(), new MemoryPersistence()); - MqttConnectOptions options = new MqttConnectOptions(); - options.setAutomaticReconnect(true); - options.setCleanSession(true); - options.setUserName(mqttConfig.getUsername()); - options.setPassword(mqttConfig.getPassword().toCharArray()); - this.mqttClient.connect(options); - } - - @PostConstruct - public void init() throws MqttException { - mqttClient.subscribe(mqttConfig.getRequestTopic(), this::handleRequest); - log.info("RPC Server subscribed to topic: {}", mqttConfig.getRequestTopic()); - } - - private void handleRequest(String topic, MqttMessage message) { - RpcRequest request = SerializationUtils.deserialize(new String(message.getPayload()), RpcRequest.class); - RpcResponse response = new RpcResponse(); - response.setCorrelationId(request.getCorrelationId()); - - try { - MethodInvoker invoker = methodRegistry.get(request.getMethod()); - if (invoker == null) { - throw new NoSuchMethodException("Unknown method: " + request.getMethod()); - } - Object result = invoker.invoke(request.getParams()); - response.setResult(result); - } catch (Exception e) { - response.setError(e.getMessage()); - log.error("Error processing RPC request: {}", e.getMessage(), e); - } - - String replyPayload = SerializationUtils.serialize(response); - MqttMessage replyMessage = new MqttMessage(replyPayload.getBytes()); - replyMessage.setQos(1); - try { - mqttClient.publish(request.getReplyTo(), replyMessage); - log.info("Published response to {}", request.getReplyTo()); - } catch (MqttException e) { - log.error("Failed to publish response: {}", e.getMessage(), e); - } - } - - /** - * 注册可调用的方法 - * - * @param methodName 方法名称 - * @param invoker 方法调用器 - */ - public void registerMethod(String methodName, MethodInvoker invoker) { - methodRegistry.put(methodName, invoker); - log.info("Registered method: {}", methodName); - } - - @PreDestroy - public void cleanup() throws MqttException { - mqttClient.disconnect(); - log.info("RPC Server disconnected"); - } - - /** - * 方法调用器接口 - */ - @FunctionalInterface - public interface MethodInvoker { - - Object invoke(Object[] params) throws Exception; - - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/controller/RpcController.java b/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/controller/RpcController.java deleted file mode 100644 index 4615dcf96f..0000000000 --- a/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/controller/RpcController.java +++ /dev/null @@ -1,38 +0,0 @@ - -package cn.iocoder.yudao.module.iot.controller; - -import cn.iocoder.yudao.module.iot.mqttrpc.client.RpcClient; -import lombok.RequiredArgsConstructor; -import org.springframework.web.bind.annotation.PostMapping; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RequestParam; -import org.springframework.web.bind.annotation.RestController; - -import javax.annotation.Resource; -import java.util.concurrent.CompletableFuture; - -// TODO 芋艿：后续 review 下 -/** - * 插件实例 RPC 接口 - * - * @author 芋道源码 - */ -@RestController -@RequestMapping("/rpc") -@RequiredArgsConstructor -public class RpcController { - - @Resource - private RpcClient rpcClient; - - @PostMapping("/add") - public CompletableFuture add(@RequestParam int a, @RequestParam int b) throws Exception { - return rpcClient.call("add", new Object[]{a, b}, 10); - } - - @PostMapping("/concat") - public CompletableFuture concat(@RequestParam String str1, @RequestParam String str2) throws Exception { - return rpcClient.call("concat", new Object[]{str1, str2}, 10); - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/client/RpcClient.java b/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/client/RpcClient.java deleted file mode 100644 index b73f88c537..0000000000 --- a/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/client/RpcClient.java +++ /dev/null @@ -1,93 +0,0 @@ -package cn.iocoder.yudao.module.iot.mqttrpc.client; - -import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcRequest; -import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcResponse; -import cn.iocoder.yudao.module.iot.mqttrpc.common.SerializationUtils; -import cn.iocoder.yudao.module.iot.mqttrpc.config.MqttConfig; -import lombok.extern.slf4j.Slf4j; -import org.eclipse.paho.client.mqttv3.MqttClient; -import org.eclipse.paho.client.mqttv3.MqttConnectOptions; -import org.eclipse.paho.client.mqttv3.MqttException; -import org.eclipse.paho.client.mqttv3.MqttMessage; -import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; -import org.springframework.stereotype.Service; - -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; -import java.util.UUID; -import java.util.concurrent.*; - -// TODO @芋艿：需要考虑，怎么公用！ -@Service -@Slf4j -public class RpcClient { - - private final MqttConfig mqttConfig; - private final MqttClient mqttClient; - private final ConcurrentMap> pendingRequests = new ConcurrentHashMap<>(); - private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); - - public RpcClient(MqttConfig mqttConfig) throws MqttException { - this.mqttConfig = mqttConfig; - this.mqttClient = new MqttClient(mqttConfig.getBroker(), mqttConfig.getClientId(), new MemoryPersistence()); - MqttConnectOptions options = new MqttConnectOptions(); - options.setAutomaticReconnect(true); - options.setCleanSession(true); - options.setUserName(mqttConfig.getUsername()); - options.setPassword(mqttConfig.getPassword().toCharArray()); - this.mqttClient.connect(options); - } - - @PostConstruct - public void init() throws MqttException { - mqttClient.subscribe(mqttConfig.getResponseTopicPrefix() + "#", this::handleResponse); - log.info("RPC Client subscribed to topics: {}", mqttConfig.getResponseTopicPrefix() + "#"); - } - - private void handleResponse(String topic, MqttMessage message) { - String correlationId = topic.substring(mqttConfig.getResponseTopicPrefix().length()); - RpcResponse response = SerializationUtils.deserialize(new String(message.getPayload()), RpcResponse.class); - CompletableFuture future = pendingRequests.remove(correlationId); - if (future != null) { - if (response.getError() != null) { - future.completeExceptionally(new RuntimeException(response.getError())); - } else { - future.complete(response); - } - } else { - log.warn("Received response for unknown correlationId: {}", correlationId); - } - } - - public CompletableFuture call(String method, Object[] params, int timeoutSeconds) throws MqttException { - String correlationId = UUID.randomUUID().toString(); - String replyTo = mqttConfig.getResponseTopicPrefix() + correlationId; - - RpcRequest request = new RpcRequest(method, params, correlationId, replyTo); - String payload = SerializationUtils.serialize(request); - MqttMessage message = new MqttMessage(payload.getBytes()); - message.setQos(1); - mqttClient.publish(mqttConfig.getRequestTopic(), message); - - CompletableFuture futureResponse = new CompletableFuture<>(); - pendingRequests.put(correlationId, futureResponse); - - // 设置超时 - scheduler.schedule(() -> { - CompletableFuture removed = pendingRequests.remove(correlationId); - if (removed != null) { - removed.completeExceptionally(new TimeoutException("RPC call timed out")); - } - }, timeoutSeconds, TimeUnit.SECONDS); - - // 返回最终的结果 - return futureResponse.thenApply(RpcResponse::getResult); - } - - @PreDestroy - public void cleanup() throws MqttException { - mqttClient.disconnect(); - scheduler.shutdown(); - log.info("RPC Client disconnected"); - } -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java b/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java deleted file mode 100644 index 89569b0c3d..0000000000 --- a/yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java +++ /dev/null @@ -1,41 +0,0 @@ -package cn.iocoder.yudao.module.iot.mqttrpc.config; - -import lombok.Data; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.context.annotation.Configuration; - -@Data -@Configuration -@ConfigurationProperties(prefix = "mqtt") -public class MqttConfig { - - /** - * MQTT 代理地址 - */ - private String broker; - - /** - * MQTT 用户名 - */ - private String username; - - /** - * MQTT 密码 - */ - private String password; - - /** - * MQTT 客户端 ID - */ - private String clientId; - - /** - * MQTT 请求主题 - */ - private String requestTopic; - - /** - * MQTT 响应主题前缀 - */ - private String responseTopicPrefix; -}