reset

apq-pc-client/src/main/java/com/priusis/client/service/MqttMessageSender.java

@@ -3,8 +3,11 @@ package com.priusis.client.service;
 import com.google.common.collect.Lists;
 import com.priusis.client.service.conf.PcConnectionConfiguration;
 import com.priusis.client.service.conf.PcPersistenceConfiguration;
+import com.priusis.client.service.core.MqttService;
+import com.priusis.client.service.core.MqttServiceImpl;
 import com.priusis.monitor.mqtt.MqttClient;
 import com.priusis.monitor.mqtt.MqttConnectResult;
+import com.priusis.monitor.mqtt.MqttHandler;
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import lombok.extern.slf4j.Slf4j;
@@ -23,20 +26,28 @@ public class MqttMessageSender implements Runnable {
     private PersistentFileService persistentFileService;
 
     private PcPersistenceConfiguration persistence;
+    private MqttHandler mqttService;
 
     private final PcConnectionConfiguration connection;
 
     private BlockingQueue<MessageFuturePair> incomingQueue;
     private Queue<Future<Void>> outgoingQueue;
 
+    private static final String DEVICE_RPC_RES_TOPIC = "devices/rpc/req";
+    private static final String DEVICE_RPC_TOPIC = DEVICE_RPC_RES_TOPIC + "/+";
+    private static final String DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC = "devices/attrs/res/+";
+    private static final String DEVICE_GET_ATTRIBUTES_RESPONSE_TOPIC = "devices/attrs/res/1";
+
     public MqttMessageSender(PcPersistenceConfiguration persistence,
                              PcConnectionConfiguration connection,
                              MqttClient tbClient,
+                             MqttHandler mqttService,
                              PersistentFileService persistentFileService,
                              BlockingQueue<MessageFuturePair> incomingQueue) {
         this.persistence = persistence;
         this.connection = connection;
         this.tbClient = tbClient;
+        this.mqttService = mqttService;
         this.persistentFileService = persistentFileService;
         this.incomingQueue = incomingQueue;
         outgoingQueue = new ConcurrentLinkedQueue();
@@ -87,6 +98,10 @@ public class MqttMessageSender implements Runnable {
                 log.info("Attempting to reconnect to Priusisiot.");
                 MqttConnectResult result = tbClient.reconnect().get(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
                 if (result.isSuccess()) {
+                    tbClient.on(DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC, mqttService).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
+                    tbClient.on(DEVICE_GET_ATTRIBUTES_RESPONSE_TOPIC, mqttService).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
+                    tbClient.on(DEVICE_RPC_TOPIC, mqttService).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
+
                     log.info("Successfully reconnected to Priusisiot.");
                 }
             } catch (TimeoutException e) {

apq-pc-client/src/main/java/com/priusis/client/service/core/MqttServiceImpl.java

@@ -356,8 +356,8 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
     private void onRpcCommand(String message) {
         JsonNode payload = fromString(message);
         MqttRpcDataMessage mqttRpcDataMessage = MqttRpcDataMessage.builder()
-                .method(payload.get("method").asText())
-                .params(Optional.ofNullable(payload.get("params")).map(JsonNode::asText).orElse(null)).build();
+                .method(payload.get("method").asText()).sendTime(System.currentTimeMillis())
+                .params(Optional.ofNullable(payload.get("params")).map(JsonNode::toString).orElse(null)).build();
 
 
         // 存储rpc下发的数据
@@ -503,7 +503,7 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
 
     private void initMqttSender(BlockingQueue<MessageFuturePair> incomingQueue) {
         mqttSenderExecutor = Executors.newSingleThreadExecutor();
-        mqttSenderExecutor.submit(new MqttMessageSender(persistence, connection, tbClient, persistentFileService, incomingQueue));
+        mqttSenderExecutor.submit(new MqttMessageSender(persistence, connection, tbClient, this, persistentFileService, incomingQueue));
     }
 
     private void initMqttReceiver(BlockingQueue<MessageFuturePair> incomingQueue) {
@@ -558,7 +558,6 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
 
 
             //tbClient.on(DEVICE_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
-            tbClient.on(DEVICE_EVENT_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
             tbClient.on(DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
             tbClient.on(DEVICE_GET_ATTRIBUTES_RESPONSE_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
             tbClient.on(DEVICE_RPC_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);

apq-pc-client/src/main/resources/application-dev.yml

-#apq:
-#  iot-gateway: 192.168.124.19:7002
-#PC_HOST: 192.168.124.29
 apq:
-  iot-gateway: 192.168.6.23:7033
-PC_HOST: 192.168.6.23
+  iot-gateway: 192.168.124.19:7002
+PC_HOST: 192.168.124.29
\ No newline at end of file
+#apq:
+#  iot-gateway: 192.168.6.23:7033
+#PC_HOST: 192.168.6.23
\ No newline at end of file