Process alarm rules on activity and inactivity events

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java

@@ -138,6 +138,8 @@ class DeviceState {
             stateChanged = processTelemetry(ctx, msg);
         } else if (msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) {
             stateChanged = processAttributesUpdateRequest(ctx, msg);
+        } else if (msg.getType().equals(DataConstants.ACTIVITY_EVENT) || msg.getType().equals(DataConstants.INACTIVITY_EVENT)) {
+            stateChanged = processDeviceActivityEvent(ctx, msg);
         } else if (msg.getType().equals(DataConstants.ATTRIBUTES_UPDATED)) {
             stateChanged = processAttributesUpdateNotification(ctx, msg);
         } else if (msg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) {
@@ -158,6 +160,11 @@ class DeviceState {
         }
     }
 
+    private boolean processDeviceActivityEvent(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
+        //TODO: Add handling a case when device state is saved in telemetry
+        return processAttributesUpdate(ctx, msg, msg.getMetaData().getValue("scope"));
+    }
+
     private boolean processAlarmClearNotification(TbContext ctx, TbMsg msg) {
         boolean stateChanged = false;
         Alarm alarmNf = JacksonUtil.fromString(msg.getData(), Alarm.class);
@@ -181,12 +188,11 @@ class DeviceState {
     }
 
     private boolean processAttributesUpdateNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
-        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
         String scope = msg.getMetaData().getValue("scope");
         if (StringUtils.isEmpty(scope)) {
             scope = DataConstants.CLIENT_SCOPE;
         }
-        return processAttributesUpdate(ctx, msg, attributes, scope);
+        return processAttributesUpdate(ctx, msg, scope);
     }
 
     private boolean processAttributesDeleteNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
@@ -211,12 +217,12 @@ class DeviceState {
     }
 
     protected boolean processAttributesUpdateRequest(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
-        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
-        return processAttributesUpdate(ctx, msg, attributes, DataConstants.CLIENT_SCOPE);
+        return processAttributesUpdate(ctx, msg, DataConstants.CLIENT_SCOPE);
     }
 
-    private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, Set<AttributeKvEntry> attributes, String scope) throws ExecutionException, InterruptedException {
+    private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException {
         boolean stateChanged = false;
+        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
         if (!attributes.isEmpty()) {
             SnapshotUpdate update = merge(latestValues, attributes, scope);
             for (DeviceProfileAlarm alarm : deviceProfile.getAlarmSettings()) {