MQTT 处理参数修改

src/main/java/com/iflytop/digester/DigestionTaskTheadManager.java

@@ -113,13 +113,13 @@ public class DigestionTaskTheadManager {
         assert actionJsonTree != null;
         String actionName = actionJsonTree.get("action").asText();
         if ( "StartNewDigestion".equals(actionName) ) {
-            this.startNewDigestion(actionJsonTree.get("params"));
+            this.startNewDigestion(actionJsonTree.get("parameters"));
             return;
         }
 
         Map<String, Object> actionParams = new HashMap<>();
-        if ( actionJsonTree.has("params") ) {
-            var paramsJsonTree = actionJsonTree.get("params");
+        if ( actionJsonTree.has("parameters") ) {
+            var paramsJsonTree = actionJsonTree.get("parameters");
             for ( JsonNode paramNode : paramsJsonTree ) {
                 actionParams.put(paramNode.get("key").asText(), paramNode.get("value").asText());
             }
@@ -148,7 +148,11 @@ public class DigestionTaskTheadManager {
         List<Map<String,Object>> tubes = new ArrayList<>();
         for ( JsonNode node : params.get("tubes") ) {
             Map<String, Object> tube = new HashMap<>();
+            if ( node.isNull() ) {
+                tube.put("no", "");
+            } else {
                 tube.put("no", node.get("no").asText());
+            }
             tubes.add(tube);
         }
 
@@ -181,7 +185,7 @@ public class DigestionTaskTheadManager {
             var message = Map.of(
                 "src", myTopic,
                 "action", action,
-                "params", params
+                "parameters", params
             );
             var messageJson = UfJsonHelper.objectToJson(message);
             MqttMessage mqttMessage = new MqttMessage(messageJson.getBytes());