DeviceProfile rule node draft

@@ -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;
@@ -126,6 +128,10 @@ public class ActorSystemContext {
 
     @Autowired
     @Getter
+    private TbDeviceProfileCache deviceProfileCache;
+
+    @Autowired
+    @Getter
     private AssetService assetService;
 
     @Autowired
@@ -530,4 +536,5 @@ public class ActorSystemContext {
         log.debug("Scheduling msg {} with delay {} ms", msg, delayInMs);
         getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS);
     }
+
 }

@@ -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;
@@ -391,6 +392,11 @@ class DefaultTbContext implements TbContext {
     }
 
     @Override
+    public RuleEngineDeviceProfileCache getDeviceProfileCache() {
+        return mainCtx.getDeviceProfileCache();
+    }
+
+    @Override
     public EventLoopGroup getSharedEventLoop() {
         return mainCtx.getSharedEventLoopGroupService().getSharedEventLoopGroup();
     }

@@ -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);
 

@@ -159,13 +159,10 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
     @PreDestroy
     public void destroy() {
         stopped = true;
-
         stopMainConsumers();
-
         if (nfConsumer != null) {
             nfConsumer.unsubscribe();
         }
-
         if (consumersExecutor != null) {
             consumersExecutor.shutdownNow();
         }

@@ -28,6 +28,10 @@ public class DataConstants {
     public static final String SERVER_SCOPE = "SERVER_SCOPE";
     public static final String SHARED_SCOPE = "SHARED_SCOPE";
     public static final String LATEST_TS = "LATEST_TS";
+    public static final String IS_NEW_ALARM = "isNewAlarm";
+    public static final String IS_EXISTING_ALARM = "isExistingAlarm";
+    public static final String IS_SEVERITY_UPDATED_ALARM = "isSeverityUpdated";
+    public static final String IS_CLEARED_ALARM = "isClearedAlarm";
 
     public static final String[] allScopes() {
         return new String[]{CLIENT_SCOPE, SHARED_SCOPE, SERVER_SCOPE};

@@ -26,7 +26,6 @@ import org.apache.commons.lang3.math.NumberUtils;
 import org.springframework.util.StringUtils;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.kv.AttributeKey;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BooleanDataEntry;
@@ -53,6 +52,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -454,11 +454,20 @@ public class JsonConverter {
     }
 
     public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException {
-        Map<Long, List<KvEntry>> result = new HashMap<>();
+        return convertToTelemetry(jsonElement, systemTs, false);
+    }
+
+    public static Map<Long, List<KvEntry>> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException {
+        return convertToTelemetry(jsonElement, systemTs, true);
+    }
+
+    public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException {
+        Map<Long, List<KvEntry>> result = sorted ? new TreeMap<>() : new HashMap<>();
         convertToTelemetry(jsonElement, systemTs, result, null);
         return result;
     }
 
+
     private static void parseObject(Map<Long, List<KvEntry>> result, long systemTs, JsonObject jo) {
         if (jo.has("ts") && jo.has("values")) {
             parseWithTs(result, jo);

@@ -54,24 +54,24 @@ public class EntityKeyMapping {
     private static final Map<String, String> entityFieldColumnMap = new HashMap<>();
     private static final Map<EntityType, Map<String, String>> 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<String> typedEntityFields = Arrays.asList(CREATED_TIME, ENTITY_TYPE, NAME, TYPE, ADDITIONAL_INFO);
     public static final List<String> widgetEntityFields = Arrays.asList(CREATED_TIME, ENTITY_TYPE, NAME);

@@ -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> T clone(T value) {
         return fromString(toString(value), (Class<T>) value.getClass());
     }
+
+    public static JsonNode valueToTree(Alarm alarm) {
+        return OBJECT_MAPPER.valueToTree(alarm);
+    }
 }

+/**
+ * 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);
+
+}

@@ -183,6 +183,8 @@ public interface TbContext {
 
     EntityViewService getEntityViewService();
 
+    RuleEngineDeviceProfileCache getDeviceProfileCache();
+
     ListeningExecutor getJsExecutor();
 
     ListeningExecutor getMailExecutor();

@@ -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<C extends TbAbstractAlarmNodeConfigura
 
     static final String PREV_ALARM_DETAILS = "prevAlarmDetails";
 
-    static final String IS_NEW_ALARM = "isNewAlarm";
-    static final String IS_EXISTING_ALARM = "isExistingAlarm";
-    static final String IS_CLEARED_ALARM = "isClearedAlarm";
-
     private final ObjectMapper mapper = new ObjectMapper();
 
     protected C config;
@@ -75,7 +72,7 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura
                 t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
     }
 
-    protected abstract ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg);
+    protected abstract ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg);
 
     protected ListenableFuture<JsonNode> buildAlarmDetails(TbContext ctx, TbMsg msg, JsonNode previousDetails) {
         try {
@@ -91,21 +88,20 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura
         }
     }
 
-    private TbMsg toAlarmMsg(TbContext ctx, AlarmResult alarmResult, TbMsg originalMsg) {
-        JsonNode jsonNodes = mapper.valueToTree(alarmResult.alarm);
+    public static TbMsg toAlarmMsg(TbContext ctx, TbAlarmResult alarmResult, TbMsg originalMsg) {
+        JsonNode jsonNodes = JacksonUtil.valueToTree(alarmResult.alarm);
         String data = jsonNodes.toString();
         TbMsgMetaData metaData = originalMsg.getMetaData().copy();
         if (alarmResult.isCreated) {
-            metaData.putValue(IS_NEW_ALARM, Boolean.TRUE.toString());
+            metaData.putValue(DataConstants.IS_NEW_ALARM, Boolean.TRUE.toString());
         } else if (alarmResult.isUpdated) {
-            metaData.putValue(IS_EXISTING_ALARM, Boolean.TRUE.toString());
+            metaData.putValue(DataConstants.IS_EXISTING_ALARM, Boolean.TRUE.toString());
         } else if (alarmResult.isCleared) {
-            metaData.putValue(IS_CLEARED_ALARM, Boolean.TRUE.toString());
+            metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString());
         }
         return ctx.transformMsg(originalMsg, "ALARM", originalMsg.getOriginator(), metaData, data);
     }
 
-
     @Override
     public void destroy() {
         if (buildDetailsJsEngine != null) {
@@ -113,17 +109,4 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura
         }
     }
 
-    protected static class AlarmResult {
-        boolean isCreated;
-        boolean isUpdated;
-        boolean isCleared;
-        Alarm alarm;
-
-        AlarmResult(boolean isCreated, boolean isUpdated, boolean isCleared, Alarm alarm) {
-            this.isCreated = isCreated;
-            this.isUpdated = isUpdated;
-            this.isCleared = isCleared;
-            this.alarm = alarm;
-        }
-    }
 }

+/**
+ * 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.action;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.thingsboard.server.common.data.alarm.Alarm;
+
+@Data
+@AllArgsConstructor
+public class TbAlarmResult {
+    boolean isCreated;
+    boolean isUpdated;
+    boolean isSeverityUpdated;
+    boolean isCleared;
+    Alarm alarm;
+
+    public TbAlarmResult(boolean isCreated, boolean isUpdated, boolean isCleared, Alarm alarm) {
+        this.isCreated = isCreated;
+        this.isUpdated = isUpdated;
+        this.isCleared = isCleared;
+        this.alarm = alarm;
+    }
+}

@@ -55,7 +55,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
     }
 
     @Override
-    protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
+    protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
         String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg.getMetaData());
         ListenableFuture<Alarm> alarmFuture;
         if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) {
@@ -67,11 +67,11 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
             if (a != null && !a.getStatus().isCleared()) {
                 return clearAlarm(ctx, msg, a);
             }
-            return Futures.immediateFuture(new AlarmResult(false, false, false, null));
+            return Futures.immediateFuture(new TbAlarmResult(false, false, false, null));
         }, ctx.getDbCallbackExecutor());
     }
 
-    private ListenableFuture<AlarmResult> clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) {
+    private ListenableFuture<TbAlarmResult> clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) {
         ctx.logJsEvalRequest();
         ListenableFuture<JsonNode> asyncDetails = buildAlarmDetails(ctx, msg, alarm.getDetails());
         return Futures.transformAsync(asyncDetails, details -> {
@@ -86,7 +86,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
                         alarm.setClearTs(savedAlarm.getClearTs());
                     }
                     alarm.setStatus(alarm.getStatus().isAck() ? AlarmStatus.CLEARED_ACK : AlarmStatus.CLEARED_UNACK);
-                    return Futures.immediateFuture(new AlarmResult(false, false, true, alarm));
+                    return Futures.immediateFuture(new TbAlarmResult(false, false, true, alarm));
                 }, ctx.getDbCallbackExecutor());
             }, ctx.getDbCallbackExecutor());
         }, ctx.getDbCallbackExecutor());

@@ -65,7 +65,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
     }
 
     @Override
-    protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
+    protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
         String alarmType;
         final Alarm msgAlarm;
 
@@ -106,7 +106,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
         return msgAlarm;
     }
 
-    private ListenableFuture<AlarmResult> createNewAlarm(TbContext ctx, TbMsg msg, Alarm msgAlarm) {
+    private ListenableFuture<TbAlarmResult> createNewAlarm(TbContext ctx, TbMsg msg, Alarm msgAlarm) {
         ListenableFuture<Alarm> asyncAlarm;
         if (msgAlarm != null) {
             asyncAlarm = Futures.immediateFuture(msgAlarm);
@@ -120,10 +120,10 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
         }
         ListenableFuture<Alarm> 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<AlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm existingAlarm, Alarm msgAlarm) {
+    private ListenableFuture<TbAlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm existingAlarm, Alarm msgAlarm) {
         ctx.logJsEvalRequest();
         ListenableFuture<Alarm> asyncUpdated = Futures.transform(buildAlarmDetails(ctx, msg, existingAlarm.getDetails()), (Function<JsonNode, Alarm>) details -> {
             ctx.logJsEvalResponse();
@@ -141,7 +141,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
             return ctx.getAlarmService().createOrUpdateAlarm(existingAlarm);
         }, ctx.getDbCallbackExecutor());
 
-        return Futures.transform(asyncUpdated, a -> 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) {

+/**
+ * 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;
+
+}

+/**
+ * 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<EntityKey, EntityKeyValue> values = new ConcurrentHashMap<>();
+
+    public DeviceDataSnapshot(Set<EntityKey> 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;
+    }
+}

+/**
+ * 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<AlarmSeverity, AlarmRule> 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<AlarmSeverity, AlarmRule> 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());
+        }
+    }
+}

+/**
+ * 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<DeviceProfileAlarm> alarmSettings = new CopyOnWriteArrayList<>();
+    @Getter(AccessLevel.PACKAGE)
+    private final Set<EntityKey> 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());
+                    }
+                }
+            }
+        }
+    }
+}

+/**
+ * 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<String, DeviceProfileAlarmState> 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<Long, List<KvEntry>> 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<KvEntry> 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<String> serverAttributeKeys = new HashSet<>();
+        Set<String> clientAttributeKeys = new HashSet<>();
+        Set<String> sharedAttributeKeys = new HashSet<>();
+        Set<String> commonAttributeKeys = new HashSet<>();
+        Set<String> 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<TsKvEntry> 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<String> commonAttributeKeys, List<AttributeKvEntry> 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());
+        }
+    }
+
+}

+/**
+ * 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 {
+
+
+
+}

+/**
+ * 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;
+    }
+
+}

+/**
+ * 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<DeviceId, DeviceState> 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() {
+
+    }
+
+}

@@ -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<Long, List<KvEntry>> 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() {
     }

@@ -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;

@@ -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));
 

+/**
+ * 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);
+    }
+
+}