diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 0b6198ebbb..c45e9f8406 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -32,6 +32,7 @@ import org.springframework.data.redis.core.RedisTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.thingsboard.rule.engine.api.MailService; +import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.server.actors.service.ActorService; import org.thingsboard.server.actors.tenant.DebugTbRateLimits; import org.thingsboard.server.common.data.DataConstants; @@ -69,6 +70,7 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.executors.ExternalCallExecutorService; import org.thingsboard.server.service.executors.SharedEventLoopGroupService; import org.thingsboard.server.service.mail.MailExecutorService; +import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.rpc.TbCoreDeviceRpcService; import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; @@ -124,6 +126,10 @@ public class ActorSystemContext { @Getter private DeviceService deviceService; + @Autowired + @Getter + private TbDeviceProfileCache deviceProfileCache; + @Autowired @Getter private AssetService assetService; @@ -530,4 +536,5 @@ public class ActorSystemContext { log.debug("Scheduling msg {} with delay {} ms", msg, delayInMs); getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS); } + } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 4e2ab430b7..cdbf322bf2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -23,6 +23,7 @@ import org.springframework.data.redis.core.RedisTemplate; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; +import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.rule.engine.api.RuleEngineRpcService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.ScriptEngine; @@ -390,6 +391,11 @@ class DefaultTbContext implements TbContext { return mainCtx.getEntityViewService(); } + @Override + public RuleEngineDeviceProfileCache getDeviceProfileCache() { + return mainCtx.getDeviceProfileCache(); + } + @Override public EventLoopGroup getSharedEventLoop() { return mainCtx.getSharedEventLoopGroupService().getSharedEventLoopGroup(); diff --git a/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java b/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java index ea14a71072..ec19eb1da6 100644 --- a/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java +++ b/application/src/main/java/org/thingsboard/server/service/profile/TbDeviceProfileCache.java @@ -15,16 +15,12 @@ */ package org.thingsboard.server.service.profile; +import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; -import org.thingsboard.server.common.data.id.TenantId; -public interface TbDeviceProfileCache { - - DeviceProfile get(TenantId tenantId, DeviceProfileId deviceProfileId); - - DeviceProfile get(TenantId tenantId, DeviceId deviceId); +public interface TbDeviceProfileCache extends RuleEngineDeviceProfileCache { void put(DeviceProfile profile); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 47beb9e796..3f6595f81e 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -159,13 +159,10 @@ public abstract class AbstractConsumerService> convertToTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException { - Map> result = new HashMap<>(); + return convertToTelemetry(jsonElement, systemTs, false); + } + + public static Map> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException { + return convertToTelemetry(jsonElement, systemTs, true); + } + + public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException { + Map> result = sorted ? new TreeMap<>() : new HashMap<>(); convertToTelemetry(jsonElement, systemTs, result, null); return result; } + private static void parseObject(Map> result, long systemTs, JsonObject jo) { if (jo.has("ts") && jo.has("values")) { parseWithTs(result, jo); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java b/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java index e11e7859cc..525483f0a0 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java @@ -54,24 +54,24 @@ public class EntityKeyMapping { private static final Map entityFieldColumnMap = new HashMap<>(); private static final Map> aliases = new HashMap<>(); - private static final String CREATED_TIME = "createdTime"; - private static final String ENTITY_TYPE = "entityType"; - private static final String NAME = "name"; - private static final String TYPE = "type"; - private static final String LABEL = "label"; - private static final String FIRST_NAME = "firstName"; - private static final String LAST_NAME = "lastName"; - private static final String EMAIL = "email"; - private static final String TITLE = "title"; - private static final String REGION = "region"; - private static final String COUNTRY = "country"; - private static final String STATE = "state"; - private static final String CITY = "city"; - private static final String ADDRESS = "address"; - private static final String ADDRESS_2 = "address2"; - private static final String ZIP = "zip"; - private static final String PHONE = "phone"; - private static final String ADDITIONAL_INFO = "additionalInfo"; + public static final String CREATED_TIME = "createdTime"; + public static final String ENTITY_TYPE = "entityType"; + public static final String NAME = "name"; + public static final String TYPE = "type"; + public static final String LABEL = "label"; + public static final String FIRST_NAME = "firstName"; + public static final String LAST_NAME = "lastName"; + public static final String EMAIL = "email"; + public static final String TITLE = "title"; + public static final String REGION = "region"; + public static final String COUNTRY = "country"; + public static final String STATE = "state"; + public static final String CITY = "city"; + public static final String ADDRESS = "address"; + public static final String ADDRESS_2 = "address2"; + public static final String ZIP = "zip"; + public static final String PHONE = "phone"; + public static final String ADDITIONAL_INFO = "additionalInfo"; public static final List typedEntityFields = Arrays.asList(CREATED_TIME, ENTITY_TYPE, NAME, TYPE, ADDITIONAL_INFO); public static final List widgetEntityFields = Arrays.asList(CREATED_TIME, ENTITY_TYPE, NAME); diff --git a/dao/src/main/java/org/thingsboard/server/dao/util/mapping/JacksonUtil.java b/dao/src/main/java/org/thingsboard/server/dao/util/mapping/JacksonUtil.java index 09c6fff1c1..d17fbe1e83 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/util/mapping/JacksonUtil.java +++ b/dao/src/main/java/org/thingsboard/server/dao/util/mapping/JacksonUtil.java @@ -18,6 +18,7 @@ package org.thingsboard.server.dao.util.mapping; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import org.thingsboard.server.common.data.alarm.Alarm; import java.io.IOException; @@ -69,4 +70,8 @@ public class JacksonUtil { public static T clone(T value) { return fromString(toString(value), (Class) value.getClass()); } + + public static JsonNode valueToTree(Alarm alarm) { + return OBJECT_MAPPER.valueToTree(alarm); + } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceProfileCache.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceProfileCache.java new file mode 100644 index 0000000000..c398131143 --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceProfileCache.java @@ -0,0 +1,32 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.api; + +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.TenantId; + +/** + * Created by ashvayka on 02.04.18. + */ +public interface RuleEngineDeviceProfileCache { + + DeviceProfile get(TenantId tenantId, DeviceProfileId deviceProfileId); + + DeviceProfile get(TenantId tenantId, DeviceId deviceId); + +} diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 8894ba222d..6ad8bc719a 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -183,6 +183,8 @@ public interface TbContext { EntityViewService getEntityViewService(); + RuleEngineDeviceProfileCache getDeviceProfileCache(); + ListeningExecutor getJsExecutor(); ListeningExecutor getMailExecutor(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java index b3344105bb..103da4706a 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java @@ -25,9 +25,10 @@ import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; -import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.dao.util.mapping.JacksonUtil; import static org.thingsboard.common.util.DonAsynchron.withCallback; @@ -37,10 +38,6 @@ public abstract class TbAbstractAlarmNode ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); } - protected abstract ListenableFuture processAlarm(TbContext ctx, TbMsg msg); + protected abstract ListenableFuture processAlarm(TbContext ctx, TbMsg msg); protected ListenableFuture buildAlarmDetails(TbContext ctx, TbMsg msg, JsonNode previousDetails) { try { @@ -91,21 +88,20 @@ public abstract class TbAbstractAlarmNode processAlarm(TbContext ctx, TbMsg msg) { + protected ListenableFuture processAlarm(TbContext ctx, TbMsg msg) { String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg.getMetaData()); ListenableFuture alarmFuture; if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) { @@ -67,11 +67,11 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) { + private ListenableFuture clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) { ctx.logJsEvalRequest(); ListenableFuture asyncDetails = buildAlarmDetails(ctx, msg, alarm.getDetails()); return Futures.transformAsync(asyncDetails, details -> { @@ -86,7 +86,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode processAlarm(TbContext ctx, TbMsg msg) { + protected ListenableFuture processAlarm(TbContext ctx, TbMsg msg) { String alarmType; final Alarm msgAlarm; @@ -106,7 +106,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode createNewAlarm(TbContext ctx, TbMsg msg, Alarm msgAlarm) { + private ListenableFuture createNewAlarm(TbContext ctx, TbMsg msg, Alarm msgAlarm) { ListenableFuture asyncAlarm; if (msgAlarm != null) { asyncAlarm = Futures.immediateFuture(msgAlarm); @@ -120,10 +120,10 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode asyncCreated = Futures.transform(asyncAlarm, alarm -> ctx.getAlarmService().createOrUpdateAlarm(alarm), ctx.getDbCallbackExecutor()); - return Futures.transform(asyncCreated, alarm -> new AlarmResult(true, false, false, alarm), MoreExecutors.directExecutor()); + return Futures.transform(asyncCreated, alarm -> new TbAlarmResult(true, false, false, alarm), MoreExecutors.directExecutor()); } - private ListenableFuture updateAlarm(TbContext ctx, TbMsg msg, Alarm existingAlarm, Alarm msgAlarm) { + private ListenableFuture updateAlarm(TbContext ctx, TbMsg msg, Alarm existingAlarm, Alarm msgAlarm) { ctx.logJsEvalRequest(); ListenableFuture asyncUpdated = Futures.transform(buildAlarmDetails(ctx, msg, existingAlarm.getDetails()), (Function) details -> { ctx.logJsEvalResponse(); @@ -141,7 +141,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode new AlarmResult(false, true, false, a), MoreExecutors.directExecutor()); + return Futures.transform(asyncUpdated, a -> new TbAlarmResult(false, true, false, a), MoreExecutors.directExecutor()); } private Alarm buildAlarm(TbMsg msg, JsonNode details, TenantId tenantId) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmStateUpdateResult.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmStateUpdateResult.java new file mode 100644 index 0000000000..de9708dc58 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmStateUpdateResult.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +public enum AlarmStateUpdateResult { + + NONE, CREATED, UPDATED, SEVERITY_UPDATED, CLEARED; + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceDataSnapshot.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceDataSnapshot.java new file mode 100644 index 0000000000..34f31cfb1d --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceDataSnapshot.java @@ -0,0 +1,49 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import lombok.Getter; +import lombok.Setter; +import org.thingsboard.server.common.data.query.EntityKey; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +public class DeviceDataSnapshot { + + private volatile boolean ready; + @Getter @Setter + private long ts; + private final Map values = new ConcurrentHashMap<>(); + + public DeviceDataSnapshot(Set entityKeySet) { + entityKeySet.forEach(key -> values.put(key, new EntityKeyValue())); + this.ready = false; + } + + void putValue(EntityKey key, EntityKeyValue value) { + values.put(key, value); + } + + EntityKeyValue getValue(EntityKey key) { + return values.get(key); + } + + boolean isReady() { + return ready; + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceProfileAlarmState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceProfileAlarmState.java new file mode 100644 index 0000000000..8b086f942f --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceProfileAlarmState.java @@ -0,0 +1,302 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import com.fasterxml.jackson.databind.JsonNode; +import lombok.Data; +import org.thingsboard.rule.engine.action.TbAlarmResult; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.alarm.AlarmSeverity; +import org.thingsboard.server.common.data.device.profile.AlarmCondition; +import org.thingsboard.server.common.data.device.profile.AlarmRule; +import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.query.BooleanFilterPredicate; +import org.thingsboard.server.common.data.query.ComplexFilterPredicate; +import org.thingsboard.server.common.data.query.KeyFilter; +import org.thingsboard.server.common.data.query.KeyFilterPredicate; +import org.thingsboard.server.common.data.query.NumericFilterPredicate; +import org.thingsboard.server.common.data.query.StringFilterPredicate; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.dao.util.mapping.JacksonUtil; + +import java.util.Comparator; +import java.util.Map; +import java.util.TreeMap; + +@Data +class DeviceProfileAlarmState { + + private final EntityId originator; + private final DeviceProfileAlarm alarmDefinition; + private volatile Map createRulesSortedBySeverityDesc; + private volatile Alarm currentAlarm; + + public DeviceProfileAlarmState(EntityId originator, DeviceProfileAlarm alarmDefinition) { + this.originator = originator; + this.alarmDefinition = alarmDefinition; + this.createRulesSortedBySeverityDesc = new TreeMap<>(Comparator.comparingInt(AlarmSeverity::ordinal)); + this.createRulesSortedBySeverityDesc.putAll(alarmDefinition.getCreateRules()); + } + + public void process(TbContext ctx, TbMsg msg, DeviceDataSnapshot data) { + AlarmSeverity resultSeverity = null; + for (Map.Entry kv : createRulesSortedBySeverityDesc.entrySet()) { + AlarmRule alarmRule = kv.getValue(); + if (eval(alarmRule.getCondition(), data)) { + resultSeverity = kv.getKey(); + break; + } + } + if (resultSeverity != null) { + pushMsg(ctx, calculateAlarmResult(ctx, resultSeverity), msg); + } else if (currentAlarm != null) { + AlarmRule clearRule = alarmDefinition.getClearRule(); + if (eval(clearRule.getCondition(), data)) { + pushMsg(ctx, new TbAlarmResult(false, false, true, currentAlarm), msg); + currentAlarm = null; + } + } + } + + public void pushMsg(TbContext ctx, TbAlarmResult alarmResult, TbMsg originalMsg) { + JsonNode jsonNodes = JacksonUtil.valueToTree(alarmResult.getAlarm()); + String data = jsonNodes.toString(); + TbMsgMetaData metaData = originalMsg.getMetaData().copy(); + String relationType; + if (alarmResult.isCreated()) { + relationType = "Alarm Created"; + metaData.putValue(DataConstants.IS_NEW_ALARM, Boolean.TRUE.toString()); + } else if (alarmResult.isUpdated()) { + relationType = "Alarm Updated"; + metaData.putValue(DataConstants.IS_EXISTING_ALARM, Boolean.TRUE.toString()); + } else if (alarmResult.isSeverityUpdated()) { + relationType = "Alarm Severity Updated"; + metaData.putValue(DataConstants.IS_EXISTING_ALARM, Boolean.TRUE.toString()); + metaData.putValue(DataConstants.IS_SEVERITY_UPDATED_ALARM, Boolean.TRUE.toString()); + } else { + relationType = "Alarm Cleared"; + metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString()); + } + TbMsg newMsg = ctx.newMsg(originalMsg.getQueueName(), "ALARM", originalMsg.getOriginator(), metaData, data); + ctx.tellNext(newMsg, relationType); + } + + private TbAlarmResult calculateAlarmResult(TbContext ctx, AlarmSeverity severity) { + if (currentAlarm != null) { + currentAlarm.setEndTs(System.currentTimeMillis()); + AlarmSeverity oldSeverity = currentAlarm.getSeverity(); + if (!oldSeverity.equals(severity)) { + currentAlarm.setSeverity(severity); + currentAlarm = ctx.getAlarmService().createOrUpdateAlarm(currentAlarm); + return new TbAlarmResult(false, false, true, false, currentAlarm); + } else { + currentAlarm = ctx.getAlarmService().createOrUpdateAlarm(currentAlarm); + return new TbAlarmResult(false, true, false, false, currentAlarm); + } + } else { + currentAlarm = new Alarm(); + currentAlarm.setSeverity(severity); + currentAlarm.setStartTs(System.currentTimeMillis()); + currentAlarm.setEndTs(currentAlarm.getStartTs()); + currentAlarm.setDetails(JacksonUtil.OBJECT_MAPPER.createObjectNode()); + currentAlarm.setOriginator(originator); + currentAlarm.setTenantId(ctx.getTenantId()); + currentAlarm.setPropagate(alarmDefinition.isPropagate()); + if (alarmDefinition.getPropagateRelationTypes() != null) { + currentAlarm.setPropagateRelationTypes(alarmDefinition.getPropagateRelationTypes()); + } + currentAlarm = ctx.getAlarmService().createOrUpdateAlarm(currentAlarm); + boolean updated = currentAlarm.getStartTs() != currentAlarm.getEndTs(); + return new TbAlarmResult(!updated, updated, false, false, currentAlarm); + } + } + + private boolean eval(AlarmCondition condition, DeviceDataSnapshot data) { + boolean eval = true; + for (KeyFilter keyFilter : condition.getCondition()) { + EntityKeyValue value = data.getValue(keyFilter.getKey()); + if (value == null) { + return false; + } + eval = eval && eval(value, keyFilter.getPredicate()); + } + //TODO: use condition duration; + return eval; + } + + private boolean eval(EntityKeyValue value, KeyFilterPredicate predicate) { + switch (predicate.getType()) { + case STRING: + return evalStrPredicate(value, (StringFilterPredicate) predicate); + case NUMERIC: + return evalNumPredicate(value, (NumericFilterPredicate) predicate); + case COMPLEX: + return evalComplexPredicate(value, (ComplexFilterPredicate) predicate); + case BOOLEAN: + return evalBoolPredicate(value, (BooleanFilterPredicate) predicate); + default: + return false; + } + } + + private boolean evalComplexPredicate(EntityKeyValue ekv, ComplexFilterPredicate predicate) { + switch (predicate.getOperation()) { + case OR: + for (KeyFilterPredicate kfp : predicate.getPredicates()) { + if (eval(ekv, kfp)) { + return true; + } + } + return false; + case AND: + for (KeyFilterPredicate kfp : predicate.getPredicates()) { + if (!eval(ekv, kfp)) { + return false; + } + } + return true; + default: + throw new RuntimeException("Operation not supported: " + predicate.getOperation()); + } + } + + + private boolean evalBoolPredicate(EntityKeyValue ekv, BooleanFilterPredicate predicate) { + Boolean value; + switch (ekv.getDataType()) { + case LONG: + value = ekv.getLngValue() > 0; + break; + case DOUBLE: + value = ekv.getDblValue() > 0; + break; + case BOOLEAN: + value = ekv.getBoolValue(); + break; + case STRING: + try { + value = Boolean.parseBoolean(ekv.getStrValue()); + break; + } catch (RuntimeException e) { + return false; + } + case JSON: + try { + value = Boolean.parseBoolean(ekv.getJsonValue()); + break; + } catch (RuntimeException e) { + return false; + } + default: + return false; + } + if (value == null) { + return false; + } + switch (predicate.getOperation()) { + case EQUAL: + return value.equals(predicate.getValue().getDefaultValue()); + case NOT_EQUAL: + return !value.equals(predicate.getValue().getDefaultValue()); + default: + throw new RuntimeException("Operation not supported: " + predicate.getOperation()); + } + } + + private boolean evalNumPredicate(EntityKeyValue ekv, NumericFilterPredicate predicate) { + Double value; + switch (ekv.getDataType()) { + case LONG: + value = ekv.getLngValue().doubleValue(); + break; + case DOUBLE: + value = ekv.getDblValue(); + break; + case BOOLEAN: + value = ekv.getBoolValue() ? 1.0 : 0.0; + break; + case STRING: + try { + value = Double.parseDouble(ekv.getStrValue()); + break; + } catch (RuntimeException e) { + return false; + } + case JSON: + try { + value = Double.parseDouble(ekv.getJsonValue()); + break; + } catch (RuntimeException e) { + return false; + } + default: + return false; + } + if (value == null) { + return false; + } + + Double predicateValue = predicate.getValue().getDefaultValue(); + switch (predicate.getOperation()) { + case NOT_EQUAL: + return !value.equals(predicateValue); + case EQUAL: + return value.equals(predicateValue); + case GREATER: + return value > predicateValue; + case GREATER_OR_EQUAL: + return value >= predicateValue; + case LESS: + return value < predicateValue; + case LESS_OR_EQUAL: + return value <= predicateValue; + default: + throw new RuntimeException("Operation not supported: " + predicate.getOperation()); + } + } + + private boolean evalStrPredicate(EntityKeyValue ekv, StringFilterPredicate predicate) { + String val; + String predicateValue; + if (predicate.isIgnoreCase()) { + val = ekv.getStrValue().toLowerCase(); + predicateValue = predicate.getValue().getDefaultValue().toLowerCase(); + } else { + val = ekv.getStrValue(); + predicateValue = predicate.getValue().getDefaultValue(); + } + switch (predicate.getOperation()) { + case CONTAINS: + return val.contains(predicateValue); + case EQUAL: + return val.equals(predicateValue); + case STARTS_WITH: + return val.startsWith(predicateValue); + case ENDS_WITH: + return val.endsWith(predicateValue); + case NOT_EQUAL: + return !val.equals(predicateValue); + case NOT_CONTAINS: + return !val.contains(predicateValue); + default: + throw new RuntimeException("Operation not supported: " + predicate.getOperation()); + } + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceProfileState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceProfileState.java new file mode 100644 index 0000000000..8a23091c2b --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceProfileState.java @@ -0,0 +1,58 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import lombok.AccessLevel; +import lombok.Getter; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.device.profile.AlarmRule; +import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm; +import org.thingsboard.server.common.data.query.EntityKey; +import org.thingsboard.server.common.data.query.KeyFilter; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; + + +class DeviceProfileState { + + private DeviceProfile deviceProfile; + @Getter(AccessLevel.PACKAGE) + private final List alarmSettings = new CopyOnWriteArrayList<>(); + @Getter(AccessLevel.PACKAGE) + private final Set entityKeys = ConcurrentHashMap.newKeySet(); + + DeviceProfileState(DeviceProfile deviceProfile) { + updateDeviceProfile(deviceProfile); + } + + void updateDeviceProfile(DeviceProfile deviceProfile) { + this.deviceProfile = deviceProfile; + alarmSettings.clear(); + if (deviceProfile.getProfileData().getAlarms() != null) { + alarmSettings.addAll(deviceProfile.getProfileData().getAlarms()); + for (DeviceProfileAlarm alarm : deviceProfile.getProfileData().getAlarms()) { + for (AlarmRule alarmRule : alarm.getCreateRules().values()) { + for (KeyFilter keyFilter : alarmRule.getCondition().getCondition()) { + entityKeys.add(keyFilter.getKey()); + } + } + } + } + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java new file mode 100644 index 0000000000..2442778cf0 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java @@ -0,0 +1,189 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import com.google.gson.JsonParser; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.common.data.query.EntityKey; +import org.thingsboard.server.common.data.query.EntityKeyType; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.session.SessionMsgType; +import org.thingsboard.server.common.transport.adaptor.JsonConverter; +import org.thingsboard.server.dao.sql.query.EntityKeyMapping; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; + +class DeviceState { + + private DeviceProfileState deviceProfile; + private DeviceDataSnapshot latestValues; + private final ConcurrentMap alarmStates = new ConcurrentHashMap<>(); + + public DeviceState(DeviceProfileState deviceProfile) { + this.deviceProfile = deviceProfile; + } + + public void process(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { + if (latestValues == null) { + latestValues = fetchLatestValues(ctx, msg.getOriginator()); + } + if (msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) { + processTelemetry(ctx, msg); + } else { + ctx.tellSuccess(msg); + } + } + + private void processTelemetry(TbContext ctx, TbMsg msg) { + Map> tsKvMap = JsonConverter.convertToSortedTelemetry(new JsonParser().parse(msg.getData()), TbMsgTimeseriesNode.getTs(msg)); + tsKvMap.forEach((ts, data) -> { + latestValues = merge(latestValues, ts, data); + for (DeviceProfileAlarm alarm : deviceProfile.getAlarmSettings()) { + DeviceProfileAlarmState alarmState = alarmStates.computeIfAbsent(alarm.getId(), a -> new DeviceProfileAlarmState(msg.getOriginator(), alarm)); + alarmState.process(ctx, msg, latestValues); + } + }); + ctx.tellSuccess(msg); + } + + private DeviceDataSnapshot merge(DeviceDataSnapshot latestValues, Long ts, List data) { + latestValues.setTs(ts); + for (KvEntry entry : data) { + latestValues.putValue(new EntityKey(EntityKeyType.TIME_SERIES, entry.getKey()), toEntityValue(entry)); + } + return latestValues; + } + + private DeviceDataSnapshot fetchLatestValues(TbContext ctx, EntityId originator) throws ExecutionException, InterruptedException { + DeviceDataSnapshot result = new DeviceDataSnapshot(deviceProfile.getEntityKeys()); + + Set serverAttributeKeys = new HashSet<>(); + Set clientAttributeKeys = new HashSet<>(); + Set sharedAttributeKeys = new HashSet<>(); + Set commonAttributeKeys = new HashSet<>(); + Set latestTsKeys = new HashSet<>(); + + Device device = null; + for (EntityKey entityKey : deviceProfile.getEntityKeys()) { + String key = entityKey.getKey(); + switch (entityKey.getType()) { + case SERVER_ATTRIBUTE: + serverAttributeKeys.add(key); + break; + case CLIENT_ATTRIBUTE: + clientAttributeKeys.add(key); + break; + case SHARED_ATTRIBUTE: + sharedAttributeKeys.add(key); + break; + case ATTRIBUTE: + serverAttributeKeys.add(key); + clientAttributeKeys.add(key); + sharedAttributeKeys.add(key); + commonAttributeKeys.add(key); + break; + case TIME_SERIES: + latestTsKeys.add(key); + break; + case ENTITY_FIELD: + if (device == null) { + device = ctx.getDeviceService().findDeviceById(ctx.getTenantId(), new DeviceId(originator.getId())); + } + if (device != null) { + switch (key) { + case EntityKeyMapping.NAME: + result.putValue(entityKey, EntityKeyValue.fromString(device.getName())); + break; + case EntityKeyMapping.TYPE: + result.putValue(entityKey, EntityKeyValue.fromString(device.getType())); + break; + case EntityKeyMapping.CREATED_TIME: + result.putValue(entityKey, EntityKeyValue.fromLong(device.getCreatedTime())); + break; + case EntityKeyMapping.LABEL: + result.putValue(entityKey, EntityKeyValue.fromString(device.getLabel())); + break; + } + } + break; + } + } + + if (!latestTsKeys.isEmpty()) { + List data = ctx.getTimeseriesService().findLatest(ctx.getTenantId(), originator, latestTsKeys).get(); + for (TsKvEntry entry : data) { + result.putValue(new EntityKey(EntityKeyType.TIME_SERIES, entry.getKey()), toEntityValue(entry)); + } + } + if (!clientAttributeKeys.isEmpty()) { + addToSnapshot(result, commonAttributeKeys, + ctx.getAttributesService().find(ctx.getTenantId(), originator, DataConstants.CLIENT_SCOPE, clientAttributeKeys).get()); + } + if (!sharedAttributeKeys.isEmpty()) { + addToSnapshot(result, commonAttributeKeys, + ctx.getAttributesService().find(ctx.getTenantId(), originator, DataConstants.SHARED_SCOPE, sharedAttributeKeys).get()); + } + if (!serverAttributeKeys.isEmpty()) { + addToSnapshot(result, commonAttributeKeys, + ctx.getAttributesService().find(ctx.getTenantId(), originator, DataConstants.SERVER_SCOPE, serverAttributeKeys).get()); + } + + return result; + } + + private void addToSnapshot(DeviceDataSnapshot snapshot, Set commonAttributeKeys, List data) { + for (AttributeKvEntry entry : data) { + EntityKeyValue value = toEntityValue(entry); + snapshot.putValue(new EntityKey(EntityKeyType.CLIENT_ATTRIBUTE, entry.getKey()), value); + if (commonAttributeKeys.contains(entry.getKey())) { + snapshot.putValue(new EntityKey(EntityKeyType.ATTRIBUTE, entry.getKey()), value); + } + } + } + + private EntityKeyValue toEntityValue(KvEntry entry) { + switch (entry.getDataType()) { + case STRING: + return EntityKeyValue.fromString(entry.getStrValue().get()); + case LONG: + return EntityKeyValue.fromLong(entry.getLongValue().get()); + case DOUBLE: + return EntityKeyValue.fromDouble(entry.getDoubleValue().get()); + case BOOLEAN: + return EntityKeyValue.fromBool(entry.getBooleanValue().get()); + case JSON: + return EntityKeyValue.fromJson(entry.getJsonValue().get()); + default: + throw new RuntimeException("Can't parse entry: " + entry.getDataType()); + } + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/EntityKeyState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/EntityKeyState.java new file mode 100644 index 0000000000..08929bd2a2 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/EntityKeyState.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +public class EntityKeyState { + + + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/EntityKeyValue.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/EntityKeyValue.java new file mode 100644 index 0000000000..40ca323307 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/EntityKeyValue.java @@ -0,0 +1,109 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import lombok.Getter; +import org.thingsboard.server.common.data.kv.DataType; + +class EntityKeyValue { + + @Getter + private DataType dataType; + private Long lngValue; + private Double dblValue; + private Boolean boolValue; + private String strValue; + + public Long getLngValue() { + return dataType == DataType.LONG ? lngValue : null; + } + + public void setLngValue(Long lngValue) { + this.dataType = DataType.LONG; + this.lngValue = lngValue; + } + + public Double getDblValue() { + return dataType == DataType.DOUBLE ? dblValue : null; + } + + public void setDblValue(Double dblValue) { + this.dataType = DataType.DOUBLE; + this.dblValue = dblValue; + } + + public Boolean getBoolValue() { + return dataType == DataType.BOOLEAN ? boolValue : null; + } + + public void setBoolValue(Boolean boolValue) { + this.dataType = DataType.BOOLEAN; + this.boolValue = boolValue; + } + + public String getStrValue() { + return dataType == DataType.STRING ? strValue : null; + } + + public void setStrValue(String strValue) { + this.dataType = DataType.STRING; + this.strValue = strValue; + } + + public void setJsonValue(String jsonValue) { + this.dataType = DataType.JSON; + this.strValue = jsonValue; + } + + public String getJsonValue() { + return dataType == DataType.JSON ? strValue : null; + } + + boolean isSet() { + return dataType != null; + } + + static EntityKeyValue fromString(String s) { + EntityKeyValue result = new EntityKeyValue(); + result.setStrValue(s); + return result; + } + + static EntityKeyValue fromBool(boolean b) { + EntityKeyValue result = new EntityKeyValue(); + result.setBoolValue(b); + return result; + } + + static EntityKeyValue fromLong(long l) { + EntityKeyValue result = new EntityKeyValue(); + result.setLngValue(l); + return result; + } + + static EntityKeyValue fromDouble(double d) { + EntityKeyValue result = new EntityKeyValue(); + result.setDblValue(d); + return result; + } + + static EntityKeyValue fromJson(String s) { + EntityKeyValue result = new EntityKeyValue(); + result.setJsonValue(s); + return result; + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java new file mode 100644 index 0000000000..52b6d2aabd --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -0,0 +1,120 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.BooleanUtils; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; +import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; +import org.thingsboard.rule.engine.api.RuleNode; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNode; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TbRelationTypes; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.rule.engine.kafka.TbKafkaNodeConfiguration; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; + +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; + +@Slf4j +@RuleNode( + type = ComponentType.ACTION, + name = "device profile", + customRelations = true, + relationTypes = {"Alarm Created", "Alarm Updated", "Alarm Severity Updated", "Alarm Cleared", "Success", "Failure"}, + configClazz = EmptyNodeConfiguration.class, + nodeDescription = "Process device messages based on device profile settings", + nodeDetails = "Create and clear alarms based on alarm rules defined in device profile. Generates ", + uiResources = {"static/rulenode/rulenode-core-config.js"}, + configDirective = "tbNodeEmptyConfig" +) +public class TbDeviceProfileNode implements TbNode { + + private RuleEngineDeviceProfileCache cache; + private Map deviceStates = new ConcurrentHashMap<>(); + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + cache = ctx.getDeviceProfileCache(); + } + + /** + * TODO: + * 1. Duration in the alarm conditions; + * 2. Update of the Profile (rules); + * 3. Update of the Device attributes (client, server and shared); + * 4. Dynamic values evaluation; + */ + + @Override + public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { + EntityType originatorType = msg.getOriginator().getEntityType(); + if (EntityType.DEVICE.equals(originatorType)) { + DeviceId deviceId = new DeviceId(msg.getOriginator().getId()); + DeviceState deviceState = getOrCreateDeviceState(ctx, msg, deviceId); + if (deviceState != null) { + deviceState.process(ctx, msg); + } else { + ctx.tellFailure(msg, new IllegalStateException("Device profile for device [" + deviceId + "] not found!")); + } + } else if (EntityType.DEVICE_PROFILE.equals(originatorType)) { + //TODO: check that the profile rule set was changed. If yes - invalidate the rules. + ctx.tellSuccess(msg); + } else { + ctx.tellSuccess(msg); + } + } + + private DeviceState getOrCreateDeviceState(TbContext ctx, TbMsg msg, DeviceId deviceId) { + DeviceState deviceState = deviceStates.get(deviceId); + if (deviceState == null) { + DeviceProfile deviceProfile = cache.get(ctx.getTenantId(), deviceId); + if (deviceProfile != null) { + deviceState = new DeviceState(new DeviceProfileState(deviceProfile)); + deviceStates.put(deviceId, deviceState); + } + } + return deviceState; + } + + @Override + public void destroy() { + + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index 9ed810863f..a415e08695 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java @@ -62,16 +62,7 @@ public class TbMsgTimeseriesNode implements TbNode { ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); return; } - long ts = -1; - String tsStr = msg.getMetaData().getValue("ts"); - if (!StringUtils.isEmpty(tsStr)) { - try { - ts = Long.parseLong(tsStr); - } catch (NumberFormatException e) { - } - } else { - ts = msg.getTs(); - } + long ts = getTs(msg); String src = msg.getData(); Map> tsKvMap = JsonConverter.convertToTelemetry(new JsonParser().parse(src), ts); if (tsKvMap.isEmpty()) { @@ -89,6 +80,20 @@ public class TbMsgTimeseriesNode implements TbNode { ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getOriginator(), tsKvEntryList, ttl, new TelemetryNodeCallback(ctx, msg)); } + public static long getTs(TbMsg msg) { + long ts = -1; + String tsStr = msg.getMetaData().getValue("ts"); + if (!StringUtils.isEmpty(tsStr)) { + try { + ts = Long.parseLong(tsStr); + } catch (NumberFormatException e) { + } + } else { + ts = msg.getTs(); + } + return ts; + } + @Override public void destroy() { } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java index ae4fb3df1e..721d053b9c 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java @@ -45,8 +45,6 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.dao.alarm.AlarmOperationResult; -import org.thingsboard.server.dao.alarm.AlarmService; import javax.script.ScriptException; import java.io.IOException; @@ -65,9 +63,9 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; -import static org.thingsboard.rule.engine.action.TbAbstractAlarmNode.IS_CLEARED_ALARM; -import static org.thingsboard.rule.engine.action.TbAbstractAlarmNode.IS_EXISTING_ALARM; -import static org.thingsboard.rule.engine.action.TbAbstractAlarmNode.IS_NEW_ALARM; +import static org.thingsboard.server.common.data.DataConstants.IS_CLEARED_ALARM; +import static org.thingsboard.server.common.data.DataConstants.IS_EXISTING_ALARM; +import static org.thingsboard.server.common.data.DataConstants.IS_NEW_ALARM; import static org.thingsboard.server.common.data.alarm.AlarmSeverity.CRITICAL; import static org.thingsboard.server.common.data.alarm.AlarmSeverity.WARNING; import static org.thingsboard.server.common.data.alarm.AlarmStatus.ACTIVE_UNACK; diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java index f835cfd506..2b71d5d14f 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java @@ -24,6 +24,7 @@ import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; import org.mockito.Matchers; import org.mockito.Mock; +import org.mockito.Mockito; import org.mockito.runners.MockitoJUnitRunner; import org.mockito.stubbing.Answer; import org.thingsboard.common.util.ListeningExecutor; @@ -90,7 +91,7 @@ public class TbJsFilterNodeTest { public void metadataConditionCanBeTrue() throws TbNodeException, ScriptException { initWithScript(); TbMsgMetaData metaData = new TbMsgMetaData(); - TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, "{}", ruleChainId, ruleNodeId); + TbMsg msg = TbMsg.newMsg("USER", null, metaData, TbMsgDataType.JSON, "{}", ruleChainId, ruleNodeId); mockJsExecutor(); when(scriptEngine.executeFilterAsync(msg)).thenReturn(Futures.immediateFuture(true)); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java new file mode 100644 index 0000000000..b0612e5c2a --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java @@ -0,0 +1,183 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.profile; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.util.concurrent.Futures; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.AdditionalAnswers; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.runners.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; +import org.springframework.util.StringUtils; +import org.thingsboard.rule.engine.api.RuleEngineAlarmService; +import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.alarm.AlarmSeverity; +import org.thingsboard.server.common.data.device.profile.AlarmCondition; +import org.thingsboard.server.common.data.device.profile.AlarmRule; +import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm; +import org.thingsboard.server.common.data.device.profile.DeviceProfileData; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.query.EntityKey; +import org.thingsboard.server.common.data.query.EntityKeyType; +import org.thingsboard.server.common.data.query.EntityKeyValueType; +import org.thingsboard.server.common.data.query.FilterPredicateValue; +import org.thingsboard.server.common.data.query.KeyFilter; +import org.thingsboard.server.common.data.query.NumericFilterPredicate; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgDataType; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.session.SessionMsgType; +import org.thingsboard.server.dao.alarm.AlarmService; +import org.thingsboard.server.dao.timeseries.TimeseriesService; + +import java.util.Collections; +import java.util.UUID; + +import static org.mockito.Mockito.verify; + +@RunWith(MockitoJUnitRunner.class) +public class TbDeviceProfileNodeTest { + + private static final ObjectMapper mapper = new ObjectMapper(); + + private TbDeviceProfileNode node; + + @Mock + private TbContext ctx; + @Mock + private RuleEngineDeviceProfileCache cache; + @Mock + private TimeseriesService timeseriesService; + @Mock + private RuleEngineAlarmService alarmService; + + private TenantId tenantId = new TenantId(UUID.randomUUID()); + private DeviceId deviceId = new DeviceId(UUID.randomUUID()); + private DeviceProfileId deviceProfileId = new DeviceProfileId(UUID.randomUUID()); + + @Test + public void testRandomMessageType() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + deviceProfileData.setAlarms(Collections.emptyList()); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 42); + TbMsg msg = TbMsg.newMsg("123456789", deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + } + + @Test + public void testEmptyProfile() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + deviceProfileData.setAlarms(Collections.emptyList()); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 42); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + } + + @Test + public void testAlarmCreate() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + KeyFilter highTempFilter = new KeyFilter(); + highTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature")); + highTempFilter.setValueType(EntityKeyValueType.NUMERIC); + NumericFilterPredicate highTemperaturePredicate = new NumericFilterPredicate(); + highTemperaturePredicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER); + highTemperaturePredicate.setValue(new FilterPredicateValue<>(30.0)); + highTempFilter.setPredicate(highTemperaturePredicate); + AlarmCondition alarmCondition = new AlarmCondition(); + alarmCondition.setCondition(Collections.singletonList(highTempFilter)); + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setCondition(alarmCondition); + DeviceProfileAlarm dpa = new DeviceProfileAlarm(); + dpa.setId("highTemperatureAlarmID"); + dpa.setCreateRules(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule)); + deviceProfileData.setAlarms(Collections.singletonList(dpa)); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature"))) + .thenReturn(Futures.immediateFuture(Collections.emptyList())); + Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())).thenAnswer(AdditionalAnswers.returnsFirstArg()); + + TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), ""); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())).thenReturn(theMsg); + + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 42); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + + TbMsg theMsg2 = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "2"); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())).thenReturn(theMsg2); + + + TbMsg msg2 = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + node.onMsg(ctx, msg2); + verify(ctx).tellSuccess(msg2); + verify(ctx).tellNext(theMsg2, "Alarm Updated"); + + } + + private void init() throws TbNodeException { + Mockito.when(ctx.getTenantId()).thenReturn(tenantId); + Mockito.when(ctx.getDeviceProfileCache()).thenReturn(cache); + Mockito.when(ctx.getTimeseriesService()).thenReturn(timeseriesService); + Mockito.when(ctx.getAlarmService()).thenReturn(alarmService); + TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.createObjectNode()); + node = new TbDeviceProfileNode(); + node.init(ctx, nodeConfiguration); + } + +}