feat(mqtt): 添加 MQTT 接收端和发送端功能

- 创建 MqttReceiveVerticle 类用于接收 MQTT 消息 - 重命名 MqttVerticle 为 MqttSendVerticle 用于发送 MQTT 消息 - 修改打印机主题常量从 mqtt.printer 为 mqtt.printer.send - 更新 VerticleLifeCycle 配置以部署 MqttSendVerticle - 实现 MQTT 客户端连接、订阅和消息处理功能

feat(mqtt): 添加 MQTT 接收端和发送端功能
- 创建 MqttReceiveVerticle 类用于接收 MQTT 消息 - 重命名 MqttVerticle 为 MqttSendVerticle 用于发送 MQTT 消息 - 修改打印机主题常量从 mqtt.printer 为 mqtt.printer.send - 更新 VerticleLifeCycle 配置以部署 MqttSendVerticle - 实现 MQTT 客户端连接、订阅和消息处理功能
zhangmeiyang
1 parent 81a45064
Showing 4 changed files with 72 additions and 9 deletions
mqtt-core/src/main/java/com/diligrp/mqtt/core/constant/MqttTopicConstant.java
mqtt-vertx/src/main/java/com/diligrp/mqtt/vertx/config/VerticleLifeCycle.java
mqtt-vertx/src/main/java/com/diligrp/mqtt/vertx/verticle/MqttReceiveVerticle.java
mqtt-vertx/src/main/java/com/diligrp/mqtt/vertx/verticle/MqttVerticle.java → mqtt-vertx/src/main/java/com/diligrp/mqtt/vertx/verticle/MqttSendVerticle.java
@@ -7,5 +7,5 @@ package com.diligrp.mqtt.core.constant;
  */
 public class MqttTopicConstant {
-    public static final String PRINTER_TOPIC = "mqtt.printer";
+    public static final String PRINTER_TOPIC = "mqtt.printer.send";
 }
 package com.diligrp.mqtt.vertx.config;
-import com.diligrp.mqtt.vertx.verticle.MqttVerticle;
+import com.diligrp.mqtt.vertx.verticle.MqttSendVerticle;
 import io.vertx.core.Vertx;
-import io.vertx.core.json.JsonObject;
-import io.vertx.ext.mongo.MongoClient;
-import io.vertx.redis.client.Redis;
 import jakarta.annotation.Resource;
 import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
@@ -25,7 +21,7 @@ public class VerticleLifeCycle implements CommandLineRunner, DisposableBean {
     @Override
     public void run(String... args) throws Exception {
-        vertx.deployVerticle(new MqttVerticle());
+        vertx.deployVerticle(new MqttSendVerticle());
     }
     @Override
+package com.diligrp.mqtt.vertx.verticle;
+
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Promise;
+import io.vertx.core.impl.logging.Logger;
+import io.vertx.core.impl.logging.LoggerFactory;
+import io.vertx.mqtt.MqttClient;
+import io.vertx.mqtt.MqttClientOptions;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-12-30 18:01
+ * @Version: todo
+ */
+public class MqttReceiveVerticle extends AbstractVerticle {
+
+    private static final Logger log = LoggerFactory.getLogger(MqttReceiveVerticle.class);
+
+    @Override
+    public void start(Promise<Void> startPromise) throws Exception {
+        // MQTT 服务器连接配置
+        MqttClientOptions options = new MqttClientOptions()
+                .setClientId("mqtt-receive-client-" + System.currentTimeMillis())
+                .setUsername("your-username") // 如果需要认证
+                .setPassword("your-password") // 如果需要认证
+                .setAutoKeepAlive(true)
+                .setSsl(false) // 根据服务器配置设置
+                .setKeepAliveInterval(30); // 保持连接间隔
+        // 创建 MQTT 客户端
+        var mqttClient = MqttClient.create(vertx, options);
+        mqttClient.connect(1883, "your-mqtt-server-host")
+                .onFailure(throwable -> {
+                    log.error("Failed to connect to MQTT server", throwable);
+                    startPromise.fail(throwable);
+                })
+                .onSuccess(ack -> {
+                    log.info("MQTT client connected to server");
+                    String topic = "your-topic-name"; // 替换为实际的主题
+                    mqttClient.subscribe(topic, 1)
+                            .onFailure(throwable -> {
+                                log.error("Failed to subscribe to topic,cause:", throwable);
+                                startPromise.fail(throwable);
+                            })
+                            .onSuccess(ack1 -> {
+                                log.info("Subscribed to topic: %s".formatted(topic));
+                                startPromise.complete();
+                            });
+                    mqttClient.publishHandler(message -> {
+                        String topicName = message.topicName();
+                        String payload = message.payload().toString();
+                        log.info("Received message - Topic: %s, Payload: %s".formatted(topicName, payload));
+                        processReceivedMessage(topicName, payload);
+                    });
+                    // 监听连接断开事件
+                    mqttClient.disconnect(v -> {
+                        log.warn("MQTT client disconnected from server");
+                    });
+                });
+    }
+
+    /**
+     * 处理接收到的 MQTT 消息
+     */
+    private void processReceivedMessage(String topic, String payload) {
+
+    }
+}
@@ -18,9 +18,9 @@ import static com.diligrp.mqtt.core.constant.MqttTopicConstant.PRINTER_TOPIC;
  * @CreateTime: 2025-12-26 17:05
  * @Version: todo
  */
-public class MqttVerticle extends AbstractVerticle {
+public class MqttSendVerticle extends AbstractVerticle {
-    private static final Logger LOGGER = LoggerFactory.getLogger(MqttVerticle.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(MqttSendVerticle.class);
     @Override