BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 552432455415142c5b57f403b52bba7eacc008e8

Authored by 安浩浩 2025-01-19 09:53:57 +0800

【功能完善】IoT: 删除不再使用的 MQTT RPC 相关配置和实现，包括 MqttConfig、RpcServer、RpcClient 和 RpcContr…

…oller 类，优化代码结构以清理未使用的组件。

Inline Side-by-side

Showing 5 changed files with 0 additions and 312 deletions

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java deleted

View file @890d304

1	-package cn.iocoder.yudao.module.iot.mqttrpc.config;
2	-
3	-import lombok.Data;
4	-import org.springframework.boot.context.properties.ConfigurationProperties;
5	-import org.springframework.context.annotation.Configuration;
6	-
7	-@Data
8	-@Configuration
9	-@ConfigurationProperties(prefix = "mqtt")
10	-public class MqttConfig {
11	- /**
12	- * MQTT 代理地址
13	- */
14	- private String broker;
15	-
16	- /**
17	- * MQTT 用户名
18	- */
19	- private String username;
20	-
21	- /**
22	- * MQTT 密码
23	- */
24	- private String password;
25	-
26	- /**
27	- * MQTT 客户端 ID
28	- */
29	- private String clientId;
30	-
31	- /**
32	- * MQTT 请求主题
33	- */
34	- private String requestTopic;
35	-
36	- /**
37	- * MQTT 响应主题前缀
38	- */
39	- private String responseTopicPrefix;
40	-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/server/RpcServer.java deleted

View file @890d304

1	-package cn.iocoder.yudao.module.iot.mqttrpc.server;
2	-
3	-import cn.hutool.core.lang.UUID;
4	-import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcRequest;
5	-import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcResponse;
6	-import cn.iocoder.yudao.module.iot.mqttrpc.common.SerializationUtils;
7	-import cn.iocoder.yudao.module.iot.mqttrpc.config.MqttConfig;
8	-import lombok.extern.slf4j.Slf4j;
9	-import org.eclipse.paho.client.mqttv3.*;
10	-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
11	-import org.springframework.stereotype.Service;
12	-
13	-import javax.annotation.PostConstruct;
14	-import javax.annotation.PreDestroy;
15	-import java.util.HashMap;
16	-import java.util.Map;
17	-
18	-// TODO @芋艿：server 逻辑，再瞅瞅；
19	-// TODO @haohao：如果只写在 iot biz 里，那么后续 server => client 貌似不方便？微信再讨论下~；
20	-@Service
21	-@Slf4j
22	-public class RpcServer {
23	-
24	- private final MqttConfig mqttConfig;
25	- private final MqttClient mqttClient;
26	- private final Map<String, MethodInvoker> methodRegistry = new HashMap<>();
27	-
28	- public RpcServer(MqttConfig mqttConfig) throws MqttException {
29	- this.mqttConfig = mqttConfig;
30	- this.mqttClient = new MqttClient(mqttConfig.getBroker(), "rpc-server-" + UUID.randomUUID(), new MemoryPersistence());
31	- MqttConnectOptions options = new MqttConnectOptions();
32	- options.setAutomaticReconnect(true);
33	- options.setCleanSession(true);
34	- options.setUserName(mqttConfig.getUsername());
35	- options.setPassword(mqttConfig.getPassword().toCharArray());
36	- this.mqttClient.connect(options);
37	- }
38	-
39	- @PostConstruct
40	- public void init() throws MqttException {
41	- mqttClient.subscribe(mqttConfig.getRequestTopic(), this::handleRequest);
42	- log.info("RPC Server subscribed to topic: {}", mqttConfig.getRequestTopic());
43	- }
44	-
45	- private void handleRequest(String topic, MqttMessage message) {
46	- RpcRequest request = SerializationUtils.deserialize(new String(message.getPayload()), RpcRequest.class);
47	- RpcResponse response = new RpcResponse();
48	- response.setCorrelationId(request.getCorrelationId());
49	-
50	- try {
51	- MethodInvoker invoker = methodRegistry.get(request.getMethod());
52	- if (invoker == null) {
53	- throw new NoSuchMethodException("Unknown method: " + request.getMethod());
54	- }
55	- Object result = invoker.invoke(request.getParams());
56	- response.setResult(result);
57	- } catch (Exception e) {
58	- response.setError(e.getMessage());
59	- log.error("Error processing RPC request: {}", e.getMessage(), e);
60	- }
61	-
62	- String replyPayload = SerializationUtils.serialize(response);
63	- MqttMessage replyMessage = new MqttMessage(replyPayload.getBytes());
64	- replyMessage.setQos(1);
65	- try {
66	- mqttClient.publish(request.getReplyTo(), replyMessage);
67	- log.info("Published response to {}", request.getReplyTo());
68	- } catch (MqttException e) {
69	- log.error("Failed to publish response: {}", e.getMessage(), e);
70	- }
71	- }
72	-
73	- /**
74	- * 注册可调用的方法
75	- *
76	- * @param methodName 方法名称
77	- * @param invoker 方法调用器
78	- */
79	- public void registerMethod(String methodName, MethodInvoker invoker) {
80	- methodRegistry.put(methodName, invoker);
81	- log.info("Registered method: {}", methodName);
82	- }
83	-
84	- @PreDestroy
85	- public void cleanup() throws MqttException {
86	- mqttClient.disconnect();
87	- log.info("RPC Server disconnected");
88	- }
89	-
90	- /**
91	- * 方法调用器接口
92	- */
93	- @FunctionalInterface
94	- public interface MethodInvoker {
95	-
96	- Object invoke(Object[] params) throws Exception;
97	-
98	- }
99	-
100	-}
101	\ No newline at end of file	0	\ No newline at end of file

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/controller/RpcController.java deleted

View file @890d304

1	-
2	-package cn.iocoder.yudao.module.iot.controller;
3	-
4	-import cn.iocoder.yudao.module.iot.mqttrpc.client.RpcClient;
5	-import lombok.RequiredArgsConstructor;
6	-import org.springframework.web.bind.annotation.PostMapping;
7	-import org.springframework.web.bind.annotation.RequestMapping;
8	-import org.springframework.web.bind.annotation.RequestParam;
9	-import org.springframework.web.bind.annotation.RestController;
10	-
11	-import javax.annotation.Resource;
12	-import java.util.concurrent.CompletableFuture;
13	-
14	-// TODO 芋艿：后续 review 下
15	-/**
16	- * 插件实例 RPC 接口
17	- *
18	- * @author 芋道源码
19	- */
20	-@RestController
21	-@RequestMapping("/rpc")
22	-@RequiredArgsConstructor
23	-public class RpcController {
24	-
25	- @Resource
26	- private RpcClient rpcClient;
27	-
28	- @PostMapping("/add")
29	- public CompletableFuture<Object> add(@RequestParam int a, @RequestParam int b) throws Exception {
30	- return rpcClient.call("add", new Object[]{a, b}, 10);
31	- }
32	-
33	- @PostMapping("/concat")
34	- public CompletableFuture<Object> concat(@RequestParam String str1, @RequestParam String str2) throws Exception {
35	- return rpcClient.call("concat", new Object[]{str1, str2}, 10);
36	- }
37	-
38	-}
39	\ No newline at end of file	0	\ No newline at end of file

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

View file @890d304

1	-package cn.iocoder.yudao.module.iot.mqttrpc.client;
2	-
3	-import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcRequest;
4	-import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcResponse;
5	-import cn.iocoder.yudao.module.iot.mqttrpc.common.SerializationUtils;
6	-import cn.iocoder.yudao.module.iot.mqttrpc.config.MqttConfig;
7	-import lombok.extern.slf4j.Slf4j;
8	-import org.eclipse.paho.client.mqttv3.MqttClient;
9	-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
10	-import org.eclipse.paho.client.mqttv3.MqttException;
11	-import org.eclipse.paho.client.mqttv3.MqttMessage;
12	-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
13	-import org.springframework.stereotype.Service;
14	-
15	-import javax.annotation.PostConstruct;
16	-import javax.annotation.PreDestroy;
17	-import java.util.UUID;
18	-import java.util.concurrent.*;
19	-
20	-// TODO @芋艿：需要考虑，怎么公用！
21	-@Service
22	-@Slf4j
23	-public class RpcClient {
24	-
25	- private final MqttConfig mqttConfig;
26	- private final MqttClient mqttClient;
27	- private final ConcurrentMap<String, CompletableFuture<RpcResponse>> pendingRequests = new ConcurrentHashMap<>();
28	- private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
29	-
30	- public RpcClient(MqttConfig mqttConfig) throws MqttException {
31	- this.mqttConfig = mqttConfig;
32	- this.mqttClient = new MqttClient(mqttConfig.getBroker(), mqttConfig.getClientId(), new MemoryPersistence());
33	- MqttConnectOptions options = new MqttConnectOptions();
34	- options.setAutomaticReconnect(true);
35	- options.setCleanSession(true);
36	- options.setUserName(mqttConfig.getUsername());
37	- options.setPassword(mqttConfig.getPassword().toCharArray());
38	- this.mqttClient.connect(options);
39	- }
40	-
41	- @PostConstruct
42	- public void init() throws MqttException {
43	- mqttClient.subscribe(mqttConfig.getResponseTopicPrefix() + "#", this::handleResponse);
44	- log.info("RPC Client subscribed to topics: {}", mqttConfig.getResponseTopicPrefix() + "#");
45	- }
46	-
47	- private void handleResponse(String topic, MqttMessage message) {
48	- String correlationId = topic.substring(mqttConfig.getResponseTopicPrefix().length());
49	- RpcResponse response = SerializationUtils.deserialize(new String(message.getPayload()), RpcResponse.class);
50	- CompletableFuture<RpcResponse> future = pendingRequests.remove(correlationId);
51	- if (future != null) {
52	- if (response.getError() != null) {
53	- future.completeExceptionally(new RuntimeException(response.getError()));
54	- } else {
55	- future.complete(response);
56	- }
57	- } else {
58	- log.warn("Received response for unknown correlationId: {}", correlationId);
59	- }
60	- }
61	-
62	- public CompletableFuture<Object> call(String method, Object[] params, int timeoutSeconds) throws MqttException {
63	- String correlationId = UUID.randomUUID().toString();
64	- String replyTo = mqttConfig.getResponseTopicPrefix() + correlationId;
65	-
66	- RpcRequest request = new RpcRequest(method, params, correlationId, replyTo);
67	- String payload = SerializationUtils.serialize(request);
68	- MqttMessage message = new MqttMessage(payload.getBytes());
69	- message.setQos(1);
70	- mqttClient.publish(mqttConfig.getRequestTopic(), message);
71	-
72	- CompletableFuture<RpcResponse> futureResponse = new CompletableFuture<>();
73	- pendingRequests.put(correlationId, futureResponse);
74	-
75	- // 设置超时
76	- scheduler.schedule(() -> {
77	- CompletableFuture<RpcResponse> removed = pendingRequests.remove(correlationId);
78	- if (removed != null) {
79	- removed.completeExceptionally(new TimeoutException("RPC call timed out"));
80	- }
81	- }, timeoutSeconds, TimeUnit.SECONDS);
82	-
83	- // 返回最终的结果
84	- return futureResponse.thenApply(RpcResponse::getResult);
85	- }
86	-
87	- @PreDestroy
88	- public void cleanup() throws MqttException {
89	- mqttClient.disconnect();
90	- scheduler.shutdown();
91	- log.info("RPC Client disconnected");
92	- }
93	-}
94	\ No newline at end of file	0	\ No newline at end of file

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

View file @890d304

1	-package cn.iocoder.yudao.module.iot.mqttrpc.config;
2	-
3	-import lombok.Data;
4	-import org.springframework.boot.context.properties.ConfigurationProperties;
5	-import org.springframework.context.annotation.Configuration;
6	-
7	-@Data
8	-@Configuration
9	-@ConfigurationProperties(prefix = "mqtt")
10	-public class MqttConfig {
11	-
12	- /**
13	- * MQTT 代理地址
14	- */
15	- private String broker;
16	-
17	- /**
18	- * MQTT 用户名
19	- */
20	- private String username;
21	-
22	- /**
23	- * MQTT 密码
24	- */
25	- private String password;
26	-
27	- /**
28	- * MQTT 客户端 ID
29	- */
30	- private String clientId;
31	-
32	- /**
33	- * MQTT 请求主题
34	- */
35	- private String requestTopic;
36	-
37	- /**
38	- * MQTT 响应主题前缀
39	- */
40	- private String responseTopicPrefix;
41	-}