From 513d52437fcf6a9ee1268669a529c3b55c028465 Mon Sep 17 00:00:00 2001
From: yuliang <398780299@qq.com>
Date: Sat, 21 Jun 2025 18:03:29 +0800
Subject: [PATCH] mqtt

---
 .../java/com/galaxis/rcs/RcsRootController.java    |  1 +
 .../amrCommunication/AmrMessageHandler.java        | 13 +++++++
 .../communication/amrCommunication/MqttConfig.java | 43 ++++++++++++++++++++--
 .../amrCommunication/MqttService.java              | 13 ++-----
 servo/src/main/resources/application-dev.yml       |  6 +--
 5 files changed, 61 insertions(+), 15 deletions(-)
 create mode 100644 servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/AmrMessageHandler.java

diff --git a/servo/src/main/java/com/galaxis/rcs/RcsRootController.java b/servo/src/main/java/com/galaxis/rcs/RcsRootController.java
index e8389f5..db2029f 100644
--- a/servo/src/main/java/com/galaxis/rcs/RcsRootController.java
+++ b/servo/src/main/java/com/galaxis/rcs/RcsRootController.java
@@ -15,6 +15,7 @@ public class RcsRootController {
     @GetMapping("/ok")
     public String ok() throws MqttException {
         mqttService.subscribe("/agv_robot/status");
+        mqttService.publish("status", "OK");
 //        mqttService.publish("rcs/root/ok", "OK");
         return "OK";
     }
diff --git a/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/AmrMessageHandler.java b/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/AmrMessageHandler.java
new file mode 100644
index 0000000..577c0dc
--- /dev/null
+++ b/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/AmrMessageHandler.java
@@ -0,0 +1,13 @@
+package com.galaxis.rcs.communication.amrCommunication;
+
+import org.eclipse.paho.mqttv5.client.IMqttMessageListener;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.springframework.stereotype.Service;
+
+@Service
+public class AmrMessageHandler implements IMqttMessageListener {
+    @Override
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        System.out.println("Received message: " + message.toString());
+    }
+}
diff --git a/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttConfig.java b/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttConfig.java
index 3b2ef9e..0c3991b 100644
--- a/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttConfig.java
+++ b/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttConfig.java
@@ -1,13 +1,16 @@
 package com.galaxis.rcs.communication.amrCommunication;
 
-import org.eclipse.paho.mqttv5.client.MqttClient;
-import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.mqttv5.client.*;
 import org.eclipse.paho.mqttv5.common.MqttException;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-
+@Slf4j
 @Configuration
 public class MqttConfig {
     @Value("${mqtt.broker-url}")
@@ -32,10 +35,44 @@ public class MqttConfig {
         return options;
     }
 
+    @Autowired private AmrMessageHandler amrMessageHandler;
     @Bean(destroyMethod = "disconnect")
     public MqttClient mqttClient() throws MqttException {
         MqttClient client = new MqttClient(brokerUrl, clientId);
+        client.setCallback(new MqttCallback() {
+            @Override
+            public void disconnected(MqttDisconnectResponse disconnectResponse) {
+                log.info("mqtt disconnected");
+            }
+
+            @Override
+            public void mqttErrorOccurred(MqttException exception) {
+                log.error("mqtt error occurred");
+            }
+
+            @Override
+            public void messageArrived(String topic, MqttMessage message) throws Exception {
+                log.info("Received message: " + message);
+            }
+
+            @Override
+            public void deliveryComplete(IMqttToken token) {
+
+            }
+
+            @Override
+            public void connectComplete(boolean reconnect, String serverURI) {
+                log.info("mqtt Connect complete: " + serverURI);
+            }
+
+            @Override
+            public void authPacketArrived(int reasonCode, MqttProperties properties) {
+
+            }
+        });
         client.connect(mqttConnectOptions());
+        client.subscribe("/agv_robot/status", 0);
+
         return client;
     }
 }
diff --git a/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttService.java b/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttService.java
index c784e3a..3997530 100644
--- a/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttService.java
+++ b/servo/src/main/java/com/galaxis/rcs/communication/amrCommunication/MqttService.java
@@ -1,27 +1,22 @@
 package com.galaxis.rcs.communication.amrCommunication;
 
-import org.eclipse.paho.mqttv5.client.IMqttMessageListener;
-import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.*;
 import org.eclipse.paho.mqttv5.common.MqttException;
 import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 @Service
-public class MqttService implements IMqttMessageListener {
+public class MqttService {
     @Autowired
     private MqttClient mqttClient;
 
     public void subscribe(String topic) throws MqttException {
-        mqttClient.subscribe(topic, 0, this);
+        mqttClient.subscribe(topic, 0);
     }
 
     public void publish(String topic, String payload) throws MqttException {
         mqttClient.publish(topic, new MqttMessage(payload.getBytes()));
     }
-
-    @Override
-    public void messageArrived(String topic, MqttMessage message) throws Exception {
-        System.out.println("Received message: " + message.toString() + " from topic: " + topic);
-    }
 }
diff --git a/servo/src/main/resources/application-dev.yml b/servo/src/main/resources/application-dev.yml
index 8f4dc6b..923f8e5 100644
--- a/servo/src/main/resources/application-dev.yml
+++ b/servo/src/main/resources/application-dev.yml
@@ -38,10 +38,10 @@ redis:
         password: 'yxt123456'
 
 mqtt:
-  broker-url: 'tcp://10.10.8.31:1883'
+  broker-url: 'tcp://180.100.199.56:7097'
   client-id: 'yvan-rcs-dev'
-  username: 'admin'
-  password: 'admin'
+  username: 'galaxis'
+  password: 'Admin1234_'
 web:
   resources:
     enable: true