diff --git a/servo/src/main/java/com/galaxis/rcs/ptr/PtrMqttClient.java b/servo/src/main/java/com/galaxis/rcs/ptr/PtrMqttClient.java index ed49e01..084def4 100644 --- a/servo/src/main/java/com/galaxis/rcs/ptr/PtrMqttClient.java +++ b/servo/src/main/java/com/galaxis/rcs/ptr/PtrMqttClient.java @@ -1,35 +1,35 @@ package com.galaxis.rcs.ptr; -import com.google.common.base.Splitter; import com.yvan.logisticsEnv.EnvPayload; -import com.yvan.logisticsModel.LogisticsRuntime; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.clever.core.BannerUtils; -import org.clever.core.json.JsonWrapper; 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 java.net.InetAddress; import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; @Slf4j public class PtrMqttClient implements MqttCallback { public final AmrMessageHandler amrMessageHandler; public final EnvPayload.MqttConfig mqttConfig; private final MqttClient client; + private CountDownLatch connectLatch = new CountDownLatch(1); @SneakyThrows public PtrMqttClient(AmrMessageHandler handler, EnvPayload.MqttConfig mqttConfig, String clientId) { this.amrMessageHandler = handler; this.mqttConfig = mqttConfig; - String brokerUrl = mqttConfig.getBrokerUrl(); - String username = mqttConfig.getUsername(); - String password = mqttConfig.getPassword(); + String brokerUrl = mqttConfig.getBrokerUrl(); // "tcp://10.10.203.239:1885" + String username = mqttConfig.getUsername(); // admin + String password = mqttConfig.getPassword(); // admin - client = new MqttClient(brokerUrl, clientId); + client = new MqttClient(brokerUrl, clientId); // String clientId = "LUOYIFAN-PC" MqttConnectionOptions options = new MqttConnectionOptions(); options.setServerURIs(new String[]{brokerUrl}); options.setAutomaticReconnect(true); @@ -40,20 +40,29 @@ public class PtrMqttClient implements MqttCallback { client.setCallback(this); + BannerUtils.printConfig(log, "MQTT 开启监听", new String[]{ + "brokerUrl: " + brokerUrl, + "userName: " + username, + "clientId: " + clientId, + "topic: /agv_robot/status"}); + client.connect(options); + connectLatch.await(); client.subscribe("/agv_robot/status", 0); } @SneakyThrows public void publish(String topic, String payload) { + // 这个一发送就阻塞 this.client.publish(topic, payload.getBytes(StandardCharsets.UTF_8), 0, false); } @SneakyThrows public static void main(String[] args) { + // 这个测试是好的 String brokerUrl = "tcp://10.10.203.239:1885"; - String clientId = "yvan-rcs-dev"; - String topic = "test/topic"; + String clientId = InetAddress.getLocalHost().getHostName(); // LUOYIFAN-PC + String topic = "/agv_robot/status"; String content = "Hello from Java"; String username = "admin"; String password = "admin"; @@ -74,7 +83,7 @@ public class PtrMqttClient implements MqttCallback { System.out.println("Publishing message..."); long start = System.currentTimeMillis(); - client.publish(topic, content.getBytes(), qos, false); + client.publish(topic, content.getBytes(), qos, false); // 这个发送是好的 System.out.println("Published in " + (System.currentTimeMillis() - start) + " ms"); client.disconnect(); @@ -113,7 +122,9 @@ public class PtrMqttClient implements MqttCallback { @Override public void connectComplete(boolean reconnect, String serverURI) { - BannerUtils.printConfig(log, "MQTT 开启监听", new String[]{serverURI + " topic: /agv_robot/status"}); + // BannerUtils.printConfig(log, "MQTT 开启监听", new String[]{serverURI + " topic: /agv_robot/status"}); + log.info("MQTT client connected to server: {}", serverURI); + connectLatch.countDown(); // 放行 } @Override