mqtt

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";
     }

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());
+    }
+}

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;
     }
 }

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);
-    }
 }

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