Merge branch 'master' of github.com:thingsboard/thingsboard

netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttChannelHandler.java

@@ -222,6 +222,9 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler<MqttMessage>
 
     private void handlePuback(MqttPubAckMessage message) {
         MqttPendingPublish pendingPublish = this.client.getPendingPublishes().get(message.variableHeader().messageId());
+        if (pendingPublish == null) {
+            return;
+        }
         pendingPublish.getFuture().setSuccess(null);
         pendingPublish.onPubackReceived();
         this.client.getPendingPublishes().remove(message.variableHeader().messageId());

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java

@@ -16,6 +16,7 @@
 package org.thingsboard.rule.engine.action;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Function;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -31,6 +32,8 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
 
+import java.io.IOException;
+
 @Slf4j
 @RuleNode(
         type = ComponentType.ACTION,
@@ -49,6 +52,8 @@ import org.thingsboard.server.common.msg.TbMsg;
 )
 public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConfiguration> {
 
+    private static ObjectMapper mapper = new ObjectMapper();
+
     @Override
     protected TbCreateAlarmNodeConfiguration loadAlarmNodeConfig(TbNodeConfiguration configuration) throws TbNodeException {
         return TbNodeUtils.convert(configuration, TbCreateAlarmNodeConfiguration.class);
@@ -56,32 +61,59 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
 
     @Override
     protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
-        ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType());
+        String alarmType;
-        return Futures.transformAsync(latest, a -> {
+        final Alarm msgAlarm;
-            if (a == null || a.getStatus().isCleared()) {
+
-                return createNewAlarm(ctx, msg);
+        if (!config.isUseMessageAlarmData()) {
+            alarmType = config.getAlarmType();
+            msgAlarm = null;
+        } else {
+            try {
+                msgAlarm = mapper.readValue(msg.getData(), Alarm.class);
+                msgAlarm.setTenantId(ctx.getTenantId());
+                alarmType = msgAlarm.getType();
+            } catch (IOException e) {
+                ctx.tellFailure(msg, e);
+                return null;
+            }
+        }
+
+        ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), alarmType);
+        return Futures.transformAsync(latest, existingAlarm -> {
+            if (existingAlarm == null || existingAlarm.getStatus().isCleared()) {
+                return createNewAlarm(ctx, msg, msgAlarm);
             } else {
-                return updateAlarm(ctx, msg, a);
+                return updateAlarm(ctx, msg, existingAlarm, msgAlarm);
             }
         }, ctx.getDbCallbackExecutor());
 
     }
 
-    private ListenableFuture<AlarmResult> createNewAlarm(TbContext ctx, TbMsg msg) {
+    private ListenableFuture<AlarmResult> createNewAlarm(TbContext ctx, TbMsg msg, Alarm msgAlarm) {
-        ListenableFuture<Alarm> asyncAlarm = Futures.transform(buildAlarmDetails(ctx, msg, null),
+        ListenableFuture<Alarm> asyncAlarm;
-                details -> buildAlarm(msg, details, ctx.getTenantId()));
+        if (msgAlarm != null ) {
+            asyncAlarm = Futures.immediateCheckedFuture(msgAlarm);
+        } else {
+            asyncAlarm = Futures.transform(buildAlarmDetails(ctx, msg, null),
+                    details -> buildAlarm(msg, details, ctx.getTenantId()));
+        }
         ListenableFuture<Alarm> asyncCreated = Futures.transform(asyncAlarm,
                 alarm -> ctx.getAlarmService().createOrUpdateAlarm(alarm), ctx.getDbCallbackExecutor());
         return Futures.transform(asyncCreated, alarm -> new AlarmResult(true, false, false, alarm));
     }
 
-    private ListenableFuture<AlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm alarm) {
+    private ListenableFuture<AlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm existingAlarm, Alarm msgAlarm) {
-        ListenableFuture<Alarm> asyncUpdated = Futures.transform(buildAlarmDetails(ctx, msg, alarm.getDetails()), (Function<JsonNode, Alarm>) details -> {
+        ListenableFuture<Alarm> asyncUpdated = Futures.transform(buildAlarmDetails(ctx, msg, existingAlarm.getDetails()), (Function<JsonNode, Alarm>) details -> {
-            alarm.setSeverity(config.getSeverity());
+            if (msgAlarm != null) {
-            alarm.setPropagate(config.isPropagate());
+                existingAlarm.setSeverity(msgAlarm.getSeverity());
-            alarm.setDetails(details);
+                existingAlarm.setPropagate(msgAlarm.isPropagate());
-            alarm.setEndTs(System.currentTimeMillis());
+            } else {
-            return ctx.getAlarmService().createOrUpdateAlarm(alarm);
+                existingAlarm.setSeverity(config.getSeverity());
+                existingAlarm.setPropagate(config.isPropagate());
+            }
+            existingAlarm.setDetails(details);
+            existingAlarm.setEndTs(System.currentTimeMillis());
+            return ctx.getAlarmService().createOrUpdateAlarm(existingAlarm);
         }, ctx.getDbCallbackExecutor());
 
         return Futures.transform(asyncUpdated, a -> new AlarmResult(false, true, false, a));

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNodeConfiguration.java

@@ -24,6 +24,7 @@ public class TbCreateAlarmNodeConfiguration extends TbAbstractAlarmNodeConfigura
 
     private AlarmSeverity severity;
     private boolean propagate;
+    private boolean useMessageAlarmData;
 
     @Override
     public TbCreateAlarmNodeConfiguration defaultConfiguration() {
@@ -36,6 +37,7 @@ public class TbCreateAlarmNodeConfiguration extends TbAbstractAlarmNodeConfigura
         configuration.setAlarmType("General Alarm");
         configuration.setSeverity(AlarmSeverity.CRITICAL);
         configuration.setPropagate(false);
+        configuration.setUseMessageAlarmData(false);
         return configuration;
     }
 }

ui/src/app/extension/extensions-forms/extension-form-mqtt.tpl.html

@@ -261,7 +261,7 @@
                                                                                 </md-input-container>
                                                                                 <md-input-container flex="60" class="md-block" md-is-error="theForm['mqttFilterExpression' + brokerIndex + mapIndex].$touched && theForm['mqttFilterExpression' + brokerIndex + mapIndex].$invalid">
                                                                                     <label translate>extension.filter-expression</label>
-                                                                                    <input required name="mqttFilterExpression{{brokerIndex}}{{mapIndex}}" ng-model="map.converter.filterExpression">
+                                                                                    <input name="mqttFilterExpression{{brokerIndex}}{{mapIndex}}" ng-model="map.converter.filterExpression">
                                                                                     <div ng-messages="theForm['mqttFilterExpression' + brokerIndex + mapIndex].$error">
                                                                                         <div translate ng-message="required">extension.field-required</div>
                                                                                     </div>