Websockets Implementation

application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java

@@ -18,7 +18,6 @@ package org.thingsboard.server.service.subscription;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
-import org.springframework.util.StringUtils;
 import org.thingsboard.common.util.DonAsynchron;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg;
@@ -53,7 +52,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.state.DefaultDeviceStateService;
 import org.thingsboard.server.service.state.DeviceStateService;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -147,6 +146,9 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
                 case ATTRIBUTES:
                     handleNewAttributeSubscription((TbAttributeSubscription) subscription);
                     break;
+                case ALARMS:
+                    handleNewAlarmsSubscription((TbAlarmsSubscription) subscription);
+                    break;
             }
         }
     }
@@ -290,7 +292,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
                 List<TsKvEntry> subscriptionUpdate = processFunction.apply(s);
                 if (subscriptionUpdate != null && !subscriptionUpdate.isEmpty()) {
                     if (serviceId.equals(s.getServiceId())) {
-                        SubscriptionUpdate update = new SubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate);
+                        TsSubscriptionUpdate update = new TsSubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate);
                         localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY);
                     } else {
                         TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId());
@@ -346,6 +348,12 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
                 e -> log.error("Failed to fetch missed updates.", e), tsCallBackExecutor);
     }
 
+    private void handleNewAlarmsSubscription(TbAlarmsSubscription subscription) {
+        log.trace("[{}][{}][{}] Processing remote alarm subscription for entity [{}]",
+                serviceId, subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId());
+        //TODO: @dlandiak search all new alarms for this entity.
+    }
+
     private void handleNewTelemetrySubscription(TbTimeseriesSubscription subscription) {
         log.trace("[{}][{}][{}] Processing remote telemetry subscription for entity [{}]",
                 serviceId, subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId());

application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java

@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -248,7 +248,8 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
             log.debug("[{}][{}] Creating new alarm subscription using: {}", session.getSessionId(), cmd.getCmdId(), cmd);
             ctx = createSubCtx(session, cmd);
         }
-        AlarmDataQuery adq = cmd.getQuery();
+        ctx.setQuery(cmd.getQuery());
+        AlarmDataQuery adq = ctx.getQuery();
         EntityDataSortOrder sortOrder = adq.getPageLink().getSortOrder();
         EntityDataSortOrder entitiesSortOrder;
         if (sortOrder == null || sortOrder.getKey().getType().equals(EntityKeyType.ALARM_FIELD)) {
@@ -265,12 +266,16 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
             AlarmDataUpdate update = new AlarmDataUpdate(cmd.getCmdId(), new PageData<>(Collections.emptyList(), 1, 0, false), null);
             wsService.sendWsMsg(ctx.getSessionId(), update);
         } else {
+            ctx.setLastFetchTs(System.currentTimeMillis());
             PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(ctx.getTenantId(), ctx.getCustomerId(),
                     ctx.getQuery().getPageLink(), ctx.getOrderedEntityIds());
-            ctx.setAlarmsData(alarms);
+            alarms = ctx.setAndMergeAlarmsData(alarms);
             AlarmDataUpdate update = new AlarmDataUpdate(cmd.getCmdId(), alarms, null);
             wsService.sendWsMsg(ctx.getSessionId(), update);
-            //TODO: Create WS subscription for alarms for this entities. If this is first page(?!) and new alarm matches the filter - invalidate alarms.
+            if (adq.getPageLink().getTimeWindow() > 0) {
+                //TODO: refresh list of entities periodically (similar to time-series subscription).
+                createAlarmSubscriptions(ctx);
+            }
         }
     }
 
@@ -389,7 +394,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
             }
             wsService.sendWsMsg(ctx.getSessionId(), update);
             if (subscribe) {
-                createSubscriptions(ctx, keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), false);
+                createTelemetrySubscriptions(ctx, keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), false);
             }
             ctx.getData().getData().forEach(ed -> ed.getTimeseries().clear());
             return ctx;
@@ -438,7 +443,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
                         update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData());
                     }
                     wsService.sendWsMsg(ctx.getSessionId(), update);
-                    createSubscriptions(ctx, latestCmd.getKeys());
+                    createTelemetrySubscriptions(ctx, latestCmd.getKeys());
                 }
 
                 @Override
@@ -454,16 +459,20 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
                 wsService.sendWsMsg(ctx.getSessionId(), update);
                 ctx.setInitialDataSent(true);
             }
-            createSubscriptions(ctx, latestCmd.getKeys());
+            createTelemetrySubscriptions(ctx, latestCmd.getKeys());
         }
     }
 
-    private void createSubscriptions(TbEntityDataSubCtx ctx, List<EntityKey> keys) {
+    private void createTelemetrySubscriptions(TbEntityDataSubCtx ctx, List<EntityKey> keys) {
-        createSubscriptions(ctx, keys, true);
+        createTelemetrySubscriptions(ctx, keys, true);
+    }
+
+    private void createAlarmSubscriptions(TbAlarmDataSubCtx ctx) {
+        List<TbSubscription> subscriptions = ctx.createSubscriptions();
+        subscriptions.forEach(localSubscriptionService::addSubscription);
     }
 
-    private void createSubscriptions(TbEntityDataSubCtx ctx, List<EntityKey> keys, boolean latest) {
+    private void createTelemetrySubscriptions(TbEntityDataSubCtx ctx, List<EntityKey> keys, boolean latest) {
-        //TODO: create context for this (session, cmdId) that contains query, latestCmd and update. Subscribe + periodic updates.
         List<TbSubscription> tbSubs = ctx.createSubscriptions(keys, latest);
         tbSubs.forEach(sub -> localSubscriptionService.addSubscription(sub));
     }

application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java

@@ -36,8 +36,7 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
-import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -138,7 +137,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
     }
 
     @Override
-    public void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback) {
+    public void onSubscriptionUpdate(String sessionId, TsSubscriptionUpdate update, TbCallback callback) {
         TbSubscription subscription = subscriptionsBySessionId
                 .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId());
         if (subscription != null) {

application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.service.subscription;
 
 import org.springframework.context.ApplicationListener;
+import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
@@ -36,4 +37,6 @@ public interface SubscriptionManagerService extends ApplicationListener<Partitio
     void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback);
 
     void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, TbCallback empty);
+
+    void onAlarmUpdate(TenantId tenantId, EntityId entityId, Alarm alarm);
 }

application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java

@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -21,14 +21,20 @@ import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.query.AlarmData;
+import org.thingsboard.server.common.data.query.AlarmDataPageLink;
 import org.thingsboard.server.common.data.query.AlarmDataQuery;
 import org.thingsboard.server.common.data.query.EntityData;
 import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
+import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
 @Slf4j
 public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
@@ -38,8 +44,15 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
     private final LinkedHashMap<EntityId, EntityData> entitiesMap;
     @Getter
     @Setter
+    private PageData<AlarmData> alarms;
+    @Getter
+    @Setter
     private boolean tooManyEntities;
 
+    private Map<Integer, EntityId> subToEntityIdMap;
+    @Setter
+    private long lastFetchTs;
+
     public TbAlarmDataSubCtx(String serviceId, TelemetryWebSocketService wsService, TelemetryWebSocketSessionRef sessionRef, int cmdId) {
         super(serviceId, wsService, sessionRef, cmdId);
         this.entitiesMap = new LinkedHashMap<>();
@@ -57,7 +70,46 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
         return entitiesMap.keySet();
     }
 
-    public void setAlarmsData(PageData<AlarmData> alarms) {
+    public PageData<AlarmData> setAndMergeAlarmsData(PageData<AlarmData> alarms) {
-//        TODO: implement
+        this.alarms = alarms;
+        for (AlarmData alarmData : alarms.getData()) {
+            EntityId entityId = alarmData.getEntityId();
+            if (entityId != null) {
+                EntityData entityData = entitiesMap.get(entityId);
+                if (entityData != null) {
+                    alarmData.getLatest().putAll(entityData.getLatest());
+                }
+            }
+        }
+        return this.alarms;
+    }
+
+    public List<TbSubscription> createSubscriptions() {
+        this.subToEntityIdMap = new HashMap<>();
+        AlarmDataPageLink pageLink = query.getPageLink();
+        List<TbSubscription> result = new ArrayList<>();
+        for (EntityData entityData : entitiesMap.values()) {
+            int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
+            subToEntityIdMap.put(subIdx, entityData.getEntityId());
+            log.trace("[{}][{}][{}] Creating alarms subscription for [{}] with query: {}", serviceId, cmdId, subIdx, entityData.getEntityId(), pageLink);
+            result.add(TbAlarmsSubscription.builder()
+                    .serviceId(serviceId)
+                    .sessionId(sessionRef.getSessionId())
+                    .subscriptionId(subIdx)
+                    .tenantId(sessionRef.getSecurityCtx().getTenantId())
+                    .entityId(entityData.getEntityId())
+                    .updateConsumer(this::sendWsMsg)
+                    .ts(lastFetchTs)
+                    .typeList(pageLink.getTypeList())
+                    .severityList(pageLink.getSeverityList())
+                    .statusList(pageLink.getStatusList())
+                    .searchPropagatedAlarms(pageLink.isSearchPropagatedAlarms())
+                    .build());
+        }
+        return result;
+    }
+
+    private void sendWsMsg(String sessionId, AlarmSubscriptionUpdate subscriptionUpdate) {
+
     }
 }

application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java

@@ -0,0 +1,59 @@
+/**
+ * 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.server.service.subscription;
+
+import lombok.Builder;
+import org.thingsboard.server.common.data.alarm.AlarmSearchStatus;
+import org.thingsboard.server.common.data.alarm.AlarmSeverity;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
+
+import java.util.List;
+import java.util.function.BiConsumer;
+
+public class TbAlarmsSubscription extends TbSubscription<AlarmSubscriptionUpdate> {
+
+    private final long ts;
+    private final List<String> typeList;
+    private final List<AlarmSearchStatus> statusList;
+    private final List<AlarmSeverity> severityList;
+    private final boolean searchPropagatedAlarms;
+
+    @Builder
+    public TbAlarmsSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId,
+                                TbSubscriptionType type, BiConsumer<String, AlarmSubscriptionUpdate> updateConsumer,
+                                long ts, List<String> typeList, List<AlarmSearchStatus> statusList,
+                                List<AlarmSeverity> severityList, boolean searchPropagatedAlarms) {
+        super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateConsumer);
+        this.ts = ts;
+        this.typeList = typeList;
+        this.statusList = statusList;
+        this.severityList = severityList;
+        this.searchPropagatedAlarms = searchPropagatedAlarms;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        return super.equals(o);
+    }
+
+    @Override
+    public int hashCode() {
+        return super.hashCode();
+    }
+}

application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java

@@ -16,16 +16,15 @@
 package org.thingsboard.server.service.subscription;
 
 import lombok.Builder;
-import lombok.Data;
 import lombok.Getter;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 import java.util.Map;
 import java.util.function.BiConsumer;
 
-public class TbAttributeSubscription extends TbSubscription {
+public class TbAttributeSubscription extends TbSubscription<TsSubscriptionUpdate> {
 
     @Getter private final boolean allKeys;
     @Getter private final Map<String, Long> keyStates;
@@ -33,7 +32,7 @@ public class TbAttributeSubscription extends TbSubscription {
 
     @Builder
     public TbAttributeSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId,
-                                   BiConsumer<String, SubscriptionUpdate> updateConsumer,
+                                   BiConsumer<String, TsSubscriptionUpdate> updateConsumer,
                                    boolean allKeys, Map<String, Long> keyStates, TbAttributeSubscriptionScope scope) {
         super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.ATTRIBUTES, updateConsumer);
         this.allKeys = allKeys;

application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java

@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -18,12 +18,9 @@ package org.thingsboard.server.service.subscription;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.query.EntityData;
 import org.thingsboard.server.common.data.query.EntityDataQuery;
@@ -36,7 +33,7 @@ import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
 import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -56,11 +53,13 @@ import java.util.stream.Collectors;
 @Slf4j
 public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
 
-    @Getter @Setter
+    @Getter
+    @Setter
     private TimeSeriesCmd tsCmd;
     @Getter
     private PageData<EntityData> data;
-    @Getter @Setter
+    @Getter
+    @Setter
     private boolean initialDataSent;
     private Map<Integer, EntityId> subToEntityIdMap;
     private volatile ScheduledFuture<?> refreshTask;
@@ -170,11 +169,11 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
         return keyStates;
     }
 
-    private void sendWsMsg(String sessionId, SubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
+    private void sendWsMsg(String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
         sendWsMsg(sessionId, subscriptionUpdate, keyType, true);
     }
 
-    private void sendWsMsg(String sessionId, SubscriptionUpdate subscriptionUpdate, EntityKeyType keyType, boolean resultToLatestValues) {
+    private void sendWsMsg(String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType, boolean resultToLatestValues) {
         EntityId entityId = subToEntityIdMap.get(subscriptionUpdate.getSubscriptionId());
         if (entityId != null) {
             log.trace("[{}][{}][{}][{}] Received subscription update: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), keyType, subscriptionUpdate);
@@ -188,7 +187,7 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
         }
     }
 
-    private void sendLatestWsMsg(EntityId entityId, String sessionId, SubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
+    private void sendLatestWsMsg(EntityId entityId, String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
         Map<String, TsValue> latestUpdate = new HashMap<>();
         subscriptionUpdate.getData().forEach((k, v) -> {
             Object[] data = (Object[]) v.get(0);
@@ -227,7 +226,7 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
         }
     }
 
-    private void sendTsWsMsg(EntityId entityId, String sessionId, SubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
+    private void sendTsWsMsg(EntityId entityId, String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
         Map<String, List<TsValue>> tsUpdate = new HashMap<>();
         subscriptionUpdate.getData().forEach((k, v) -> {
             Object[] data = (Object[]) v.get(0);

application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java

@@ -18,7 +18,7 @@ package org.thingsboard.server.service.subscription;
 import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.common.msg.queue.TbCallback;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 public interface TbLocalSubscriptionService {
 
@@ -28,7 +28,7 @@ public interface TbLocalSubscriptionService {
 
     void cancelAllSessionSubscriptions(String sessionId);
 
-    void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback);
+    void onSubscriptionUpdate(String sessionId, TsSubscriptionUpdate update, TbCallback callback);
 
     void onApplicationEvent(PartitionChangeEvent event);
 

application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java

@@ -19,14 +19,13 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
 
 import java.util.Objects;
 import java.util.function.BiConsumer;
 
 @Data
 @AllArgsConstructor
-public abstract class TbSubscription {
+public abstract class TbSubscription<T> {
 
     private final String serviceId;
     private final String sessionId;
@@ -34,7 +33,7 @@ public abstract class TbSubscription {
     private final TenantId tenantId;
     private final EntityId entityId;
     private final TbSubscriptionType type;
-    private final BiConsumer<String, SubscriptionUpdate> updateConsumer;
+    private final BiConsumer<String, T> updateConsumer;
 
     @Override
     public boolean equals(Object o) {

application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionType.java

@@ -16,5 +16,5 @@
 package org.thingsboard.server.service.subscription;
 
 public enum TbSubscriptionType {
-    TIMESERIES, ATTRIBUTES
+    TIMESERIES, ATTRIBUTES, ALARMS
 }

application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java

@@ -44,7 +44,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdatePr
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -137,9 +137,9 @@ public class TbSubscriptionUtils {
         return builder.build();
     }
 
-    public static SubscriptionUpdate fromProto(TbSubscriptionUpdateProto proto) {
+    public static TsSubscriptionUpdate fromProto(TbSubscriptionUpdateProto proto) {
         if (proto.getErrorCode() > 0) {
-            return new SubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg());
+            return new TsSubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg());
         } else {
             Map<String, List<Object>> data = new TreeMap<>();
             proto.getDataList().forEach(v -> {
@@ -151,7 +151,7 @@ public class TbSubscriptionUtils {
                     values.add(value);
                 }
             });
-            return new SubscriptionUpdate(proto.getSubscriptionId(), data);
+            return new TsSubscriptionUpdate(proto.getSubscriptionId(), data);
         }
     }
 

application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java

@@ -19,12 +19,12 @@ import lombok.Builder;
 import lombok.Getter;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 import java.util.Map;
 import java.util.function.BiConsumer;
 
-public class TbTimeseriesSubscription extends TbSubscription {
+public class TbTimeseriesSubscription extends TbSubscription<TsSubscriptionUpdate> {
 
     @Getter
     private final boolean allKeys;
@@ -37,7 +37,7 @@ public class TbTimeseriesSubscription extends TbSubscription {
 
     @Builder
     public TbTimeseriesSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId,
-                                    BiConsumer<String, SubscriptionUpdate> updateConsumer,
+                                    BiConsumer<String, TsSubscriptionUpdate> updateConsumer,
                                     boolean allKeys, Map<String, Long> keyStates, long startTime, long endTime) {
         super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.TIMESERIES, updateConsumer);
         this.allKeys = allKeys;

application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.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.server.service.telemetry;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BooleanDataEntry;
+import org.thingsboard.server.common.data.kv.DoubleDataEntry;
+import org.thingsboard.server.common.data.kv.LongDataEntry;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.service.queue.TbClusterService;
+import org.thingsboard.server.service.subscription.SubscriptionManagerService;
+import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
+
+import javax.annotation.Nullable;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+
+/**
+ * Created by ashvayka on 27.03.18.
+ */
+@Slf4j
+public abstract class AbstractSubscriptionService implements ApplicationListener<PartitionChangeEvent> {
+
+    protected final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
+
+    protected final TbClusterService clusterService;
+    protected final PartitionService partitionService;
+    protected Optional<SubscriptionManagerService> subscriptionManagerService;
+
+    protected ExecutorService wsCallBackExecutor;
+
+    public AbstractSubscriptionService(TbClusterService clusterService,
+                                       PartitionService partitionService) {
+        this.clusterService = clusterService;
+        this.partitionService = partitionService;
+    }
+
+    @Autowired(required = false)
+    public void setSubscriptionManagerService(Optional<SubscriptionManagerService> subscriptionManagerService) {
+        this.subscriptionManagerService = subscriptionManagerService;
+    }
+
+    abstract String getExecutorPrefix();
+
+    @PostConstruct
+    public void initExecutor() {
+        wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName(getExecutorPrefix() + "-service-ws-callback"));
+    }
+
+    @PreDestroy
+    public void shutdownExecutor() {
+        if (wsCallBackExecutor != null) {
+            wsCallBackExecutor.shutdownNow();
+        }
+    }
+
+    @Override
+    @EventListener(PartitionChangeEvent.class)
+    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+        if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
+            currentPartitions.clear();
+            currentPartitions.addAll(partitionChangeEvent.getPartitions());
+        }
+    }
+
+    protected void addWsCallback(ListenableFuture<List<Void>> saveFuture, Consumer<Void> callback) {
+        Futures.addCallback(saveFuture, new FutureCallback<List<Void>>() {
+            @Override
+            public void onSuccess(@Nullable List<Void> result) {
+                callback.accept(null);
+            }
+
+            @Override
+            public void onFailure(Throwable t) {
+            }
+        }, wsCallBackExecutor);
+    }
+}

application/src/main/java/org/thingsboard/server/service/telemetry/AlarmSubscriptionService.java

@@ -0,0 +1,28 @@
+/**
+ * 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.server.service.telemetry;
+
+import org.springframework.context.ApplicationListener;
+import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
+import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
+import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+
+/**
+ * Created by ashvayka on 27.03.18.
+ */
+public interface AlarmSubscriptionService extends RuleEngineAlarmService, ApplicationListener<PartitionChangeEvent> {
+
+}

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java

@@ -0,0 +1,155 @@
+/**
+ * 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.server.service.telemetry;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.alarm.AlarmInfo;
+import org.thingsboard.server.common.data.alarm.AlarmQuery;
+import org.thingsboard.server.common.data.alarm.AlarmSearchStatus;
+import org.thingsboard.server.common.data.alarm.AlarmSeverity;
+import org.thingsboard.server.common.data.alarm.AlarmStatus;
+import org.thingsboard.server.common.data.id.AlarmId;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BooleanDataEntry;
+import org.thingsboard.server.common.data.kv.DoubleDataEntry;
+import org.thingsboard.server.common.data.kv.LongDataEntry;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.query.AlarmData;
+import org.thingsboard.server.common.data.query.AlarmDataPageLink;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.dao.alarm.AlarmService;
+import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.service.queue.TbClusterService;
+import org.thingsboard.server.service.subscription.SubscriptionManagerService;
+import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
+import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
+
+import javax.annotation.Nullable;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+
+/**
+ * Created by ashvayka on 27.03.18.
+ */
+@Service
+@Slf4j
+public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService implements AlarmSubscriptionService {
+
+    private final AlarmService alarmService;
+
+    public DefaultAlarmSubscriptionService(TbClusterService clusterService,
+                                           PartitionService partitionService,
+                                           AlarmService alarmService) {
+        super(clusterService, partitionService);
+        this.alarmService = alarmService;
+    }
+
+    @Autowired(required = false)
+    public void setSubscriptionManagerService(Optional<SubscriptionManagerService> subscriptionManagerService) {
+        this.subscriptionManagerService = subscriptionManagerService;
+    }
+
+    @Override
+    String getExecutorPrefix() {
+        return "alarm";
+    }
+
+    @Override
+    public Alarm createOrUpdateAlarm(Alarm alarm) {
+        //TODO 3.1: we also need a list of related entities if this is propagated alarm;
+        Alarm result = alarmService.createOrUpdateAlarm(alarm);
+        List<EntityId> relatedEntities = Collections.singletonList(result.getOriginator());
+        pushAlarmToSubService(result, relatedEntities);
+        return result;
+    }
+
+    private void pushAlarmToSubService(Alarm result, List<EntityId> entityIds) {
+        wsCallBackExecutor.submit(() -> {
+            TenantId tenantId = result.getTenantId();
+            for (EntityId entityId : entityIds) {
+                TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
+                if (currentPartitions.contains(tpi)) {
+                    if (subscriptionManagerService.isPresent()) {
+                        subscriptionManagerService.get().onAlarmUpdate(tenantId, entityId, result);
+                    } else {
+                        log.warn("Possible misconfiguration because subscriptionManagerService is null!");
+                    }
+                } else {
+                    //TODO 3.1: cluster mode notification
+//                    TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toTimeseriesUpdateProto(tenantId, entityId, ts);
+//                    clusterService.pushMsgToCore(tpi, entityId.getId(), toCoreMsg, null);
+                }
+            }
+        });
+    }
+
+    @Override
+    public Boolean deleteAlarm(TenantId tenantId, AlarmId alarmId) {
+        return alarmService.deleteAlarm(tenantId, alarmId);
+    }
+
+    @Override
+    public ListenableFuture<Boolean> ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTs) {
+        return alarmService.ackAlarm(tenantId, alarmId, ackTs);
+    }
+
+    @Override
+    public ListenableFuture<Boolean> clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTs) {
+        return alarmService.clearAlarm(tenantId, alarmId, details, clearTs);
+    }
+
+    @Override
+    public ListenableFuture<Alarm> findAlarmByIdAsync(TenantId tenantId, AlarmId alarmId) {
+        return alarmService.findAlarmByIdAsync(tenantId, alarmId);
+    }
+
+    @Override
+    public ListenableFuture<Alarm> findLatestByOriginatorAndType(TenantId tenantId, EntityId originator, String type) {
+        return alarmService.findLatestByOriginatorAndType(tenantId, originator, type);
+    }
+
+
+}

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java

@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -61,38 +61,31 @@ import java.util.function.Consumer;
  */
 @Service
 @Slf4j
-public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptionService {
+public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionService implements TelemetrySubscriptionService {
-
-    private final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
 
     private final AttributesService attrService;
     private final TimeseriesService tsService;
-    private final TbClusterService clusterService;
-    private final PartitionService partitionService;
-    private Optional<SubscriptionManagerService> subscriptionManagerService;
 
     private ExecutorService tsCallBackExecutor;
-    private ExecutorService wsCallBackExecutor;
 
     public DefaultTelemetrySubscriptionService(AttributesService attrService,
                                                TimeseriesService tsService,
                                                TbClusterService clusterService,
                                                PartitionService partitionService) {
+        super(clusterService, partitionService);
         this.attrService = attrService;
         this.tsService = tsService;
-        this.clusterService = clusterService;
-        this.partitionService = partitionService;
-    }
-
-    @Autowired(required = false)
-    public void setSubscriptionManagerService(Optional<SubscriptionManagerService> subscriptionManagerService) {
-        this.subscriptionManagerService = subscriptionManagerService;
     }
 
     @PostConstruct
     public void initExecutor() {
+        super.initExecutor();
         tsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-service-ts-callback"));
-        wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-service-ws-callback"));
+    }
+
+    @Override
+    protected String getExecutorPrefix() {
+        return "ts";
     }
 
     @PreDestroy
@@ -100,18 +93,7 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
         if (tsCallBackExecutor != null) {
             tsCallBackExecutor.shutdownNow();
         }
-        if (wsCallBackExecutor != null) {
+        super.shutdownExecutor();
-            wsCallBackExecutor.shutdownNow();
-        }
-    }
-
-    @Override
-    @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
-            currentPartitions.clear();
-            currentPartitions.addAll(partitionChangeEvent.getPartitions());
-        }
     }
 
     @Override
@@ -219,17 +201,4 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
             }
         }, tsCallBackExecutor);
     }
-
-    private void addWsCallback(ListenableFuture<List<Void>> saveFuture, Consumer<Void> callback) {
-        Futures.addCallback(saveFuture, new FutureCallback<List<Void>>() {
-            @Override
-            public void onSuccess(@Nullable List<Void> result) {
-                callback.accept(null);
-            }
-
-            @Override
-            public void onFailure(Throwable t) {
-            }
-        }, wsCallBackExecutor);
-    }
 }

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java

@@ -49,7 +49,6 @@ import org.thingsboard.server.service.security.AccessValidator;
 import org.thingsboard.server.service.security.ValidationCallback;
 import org.thingsboard.server.service.security.ValidationResult;
 import org.thingsboard.server.service.security.ValidationResultCode;
-import org.thingsboard.server.service.security.model.SecurityUser;
 import org.thingsboard.server.service.security.model.UserPrincipal;
 import org.thingsboard.server.service.security.permission.Operation;
 import org.thingsboard.server.service.subscription.TbEntityDataSubscriptionService;
@@ -64,14 +63,13 @@ import org.thingsboard.server.service.telemetry.cmd.v1.TelemetryPluginCmd;
 import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
 import org.thingsboard.server.service.telemetry.cmd.v1.TimeseriesSubscriptionCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd;
-import org.thingsboard.server.service.telemetry.cmd.v2.DataCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.DataUpdate;
 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
 import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
 import org.thingsboard.server.service.telemetry.exception.UnauthorizedException;
 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
 
 import javax.annotation.Nullable;
 import javax.annotation.PostConstruct;
@@ -89,7 +87,6 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -222,7 +219,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
             }
         } catch (IOException e) {
             log.warn("Failed to decode subscription cmd: {}", e.getMessage(), e);
-            SubscriptionUpdate update = new SubscriptionUpdate(UNKNOWN_SUBSCRIPTION_ID, SubscriptionErrorCode.INTERNAL_ERROR, SESSION_META_DATA_NOT_FOUND);
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(UNKNOWN_SUBSCRIPTION_ID, SubscriptionErrorCode.INTERNAL_ERROR, SESSION_META_DATA_NOT_FOUND);
             sendWsMsg(sessionRef, update);
         }
     }
@@ -257,7 +254,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
     }
 
     @Override
-    public void sendWsMsg(String sessionId, SubscriptionUpdate update) {
+    public void sendWsMsg(String sessionId, TsSubscriptionUpdate update) {
         sendWsMsg(sessionId, update.getSubscriptionId(), update);
     }
 
@@ -400,7 +397,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
             @Override
             public void onSuccess(List<AttributeKvEntry> data) {
                 List<TsKvEntry> attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList());
-                sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData));
+                sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), attributesData));
 
                 Map<String, Long> subState = new HashMap<>(keys.size());
                 keys.forEach(key -> subState.put(key, 0L));
@@ -425,12 +422,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
             @Override
             public void onFailure(Throwable e) {
                 log.error(FAILED_TO_FETCH_ATTRIBUTES, e);
-                SubscriptionUpdate update;
+                TsSubscriptionUpdate update;
                 if (e instanceof UnauthorizedException) {
-                    update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED,
+                    update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED,
                             SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg());
                 } else {
-                    update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+                    update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
                             FAILED_TO_FETCH_ATTRIBUTES);
                 }
                 sendWsMsg(sessionRef, update);
@@ -449,19 +446,19 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId);
         if (sessionMD == null) {
             log.warn("[{}] Session meta data not found. ", sessionId);
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
                     SESSION_META_DATA_NOT_FOUND);
             sendWsMsg(sessionRef, update);
             return;
         }
         if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty() || cmd.getEntityType() == null || cmd.getEntityType().isEmpty()) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Device id is empty!");
             sendWsMsg(sessionRef, update);
             return;
         }
         if (cmd.getKeys() == null || cmd.getKeys().isEmpty()) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Keys are empty!");
             sendWsMsg(sessionRef, update);
             return;
@@ -474,17 +471,17 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         FutureCallback<List<TsKvEntry>> callback = new FutureCallback<List<TsKvEntry>>() {
             @Override
             public void onSuccess(List<TsKvEntry> data) {
-                sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), data));
+                sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), data));
             }
 
             @Override
             public void onFailure(Throwable e) {
-                SubscriptionUpdate update;
+                TsSubscriptionUpdate update;
                 if (UnauthorizedException.class.isInstance(e)) {
-                    update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED,
+                    update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED,
                             SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg());
                 } else {
-                    update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+                    update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
                             FAILED_TO_FETCH_DATA);
                 }
                 sendWsMsg(sessionRef, update);
@@ -500,7 +497,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
             @Override
             public void onSuccess(List<AttributeKvEntry> data) {
                 List<TsKvEntry> attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList());
-                sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData));
+                sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), attributesData));
 
                 Map<String, Long> subState = new HashMap<>(attributesData.size());
                 attributesData.forEach(v -> subState.put(v.getKey(), v.getTs()));
@@ -523,7 +520,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
             @Override
             public void onFailure(Throwable e) {
                 log.error(FAILED_TO_FETCH_ATTRIBUTES, e);
-                SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+                TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
                         FAILED_TO_FETCH_ATTRIBUTES);
                 sendWsMsg(sessionRef, update);
             }
@@ -586,7 +583,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         FutureCallback<List<TsKvEntry>> callback = new FutureCallback<List<TsKvEntry>>() {
             @Override
             public void onSuccess(List<TsKvEntry> data) {
-                sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), data));
+                sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), data));
                 Map<String, Long> subState = new HashMap<>(data.size());
                 data.forEach(v -> subState.put(v.getKey(), v.getTs()));
 
@@ -604,12 +601,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
 
             @Override
             public void onFailure(Throwable e) {
-                SubscriptionUpdate update;
+                TsSubscriptionUpdate update;
                 if (UnauthorizedException.class.isInstance(e)) {
-                    update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED,
+                    update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED,
                             SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg());
                 } else {
-                    update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+                    update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
                             FAILED_TO_FETCH_DATA);
                 }
                 sendWsMsg(sessionRef, update);
@@ -623,7 +620,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         return new FutureCallback<List<TsKvEntry>>() {
             @Override
             public void onSuccess(List<TsKvEntry> data) {
-                sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), data));
+                sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), data));
                 Map<String, Long> subState = new HashMap<>(keys.size());
                 keys.forEach(key -> subState.put(key, startTs));
                 data.forEach(v -> subState.put(v.getKey(), v.getTs()));
@@ -647,7 +644,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                 } else {
                     log.info(FAILED_TO_FETCH_DATA, e);
                 }
-                SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+                TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
                         FAILED_TO_FETCH_DATA);
                 sendWsMsg(sessionRef, update);
             }
@@ -664,12 +661,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
 
     private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) {
         if (cmd.getCmdId() < 0) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Cmd id is negative value!");
             sendWsMsg(sessionRef, update);
             return false;
         } else if (cmd.getQuery() == null && cmd.getLatestCmd() == null && cmd.getHistoryCmd() == null && cmd.getTsCmd() == null) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Query is empty!");
             sendWsMsg(sessionRef, update);
             return false;
@@ -679,12 +676,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
 
     private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) {
         if (cmd.getCmdId() < 0) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Cmd id is negative value!");
             sendWsMsg(sessionRef, update);
             return false;
         } else if (cmd.getQuery() == null) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Query is empty!");
             sendWsMsg(sessionRef, update);
             return false;
@@ -694,7 +691,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
 
     private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) {
         if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) {
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
                     "Device id is empty!");
             sendWsMsg(sessionRef, update);
             return false;
@@ -710,7 +707,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId);
         if (sessionMD == null) {
             log.warn("[{}] Session meta data not found. ", sessionId);
-            SubscriptionUpdate update = new SubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR,
+            TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR,
                     SESSION_META_DATA_NOT_FOUND);
             sendWsMsg(sessionRef, update);
             return false;
@@ -723,7 +720,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         sendWsMsg(sessionRef, update.getCmdId(), update);
     }
 
-    private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, SubscriptionUpdate update) {
+    private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, TsSubscriptionUpdate update) {
         sendWsMsg(sessionRef, update.getSubscriptionId(), update);
     }
 

application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java

@@ -16,8 +16,7 @@
 package org.thingsboard.server.service.telemetry;
 
 import org.thingsboard.server.service.telemetry.cmd.v2.DataUpdate;
-import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
+import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate;
-import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
 
 /**
  * Created by ashvayka on 27.03.18.
@@ -28,7 +27,7 @@ public interface TelemetryWebSocketService {
 
     void handleWebSocketMsg(TelemetryWebSocketSessionRef sessionRef, String msg);
 
-    void sendWsMsg(String sessionId, SubscriptionUpdate update);
+    void sendWsMsg(String sessionId, TsSubscriptionUpdate update);
 
     void sendWsMsg(String sessionId, DataUpdate update);
 

application/src/main/java/org/thingsboard/server/service/telemetry/sub/AlarmSubscriptionUpdate.java

@@ -0,0 +1,71 @@
+/**
+ * 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.server.service.telemetry.sub;
+
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.query.AlarmData;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+public class AlarmSubscriptionUpdate {
+
+    private int subscriptionId;
+    private int errorCode;
+    private String errorMsg;
+    private Alarm alarm;
+
+    public AlarmSubscriptionUpdate(int subscriptionId, Alarm alarm) {
+        super();
+        this.subscriptionId = subscriptionId;
+        this.alarm = alarm;
+    }
+
+    public AlarmSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode) {
+        this(subscriptionId, errorCode, null);
+    }
+
+    public AlarmSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode, String errorMsg) {
+        super();
+        this.subscriptionId = subscriptionId;
+        this.errorCode = errorCode.getCode();
+        this.errorMsg = errorMsg != null ? errorMsg : errorCode.getDefaultMsg();
+    }
+
+    public int getSubscriptionId() {
+        return subscriptionId;
+    }
+
+
+    public int getErrorCode() {
+        return errorCode;
+    }
+
+    public String getErrorMsg() {
+        return errorMsg;
+    }
+
+    @Override
+    public String toString() {
+        return "AlarmUpdate [subscriptionId=" + subscriptionId + ", errorCode=" + errorCode + ", errorMsg=" + errorMsg + ", alarm="
+                + alarm + "]";
+    }
+}

application/src/main/java/org/thingsboard/server/service/telemetry/sub/SubscriptionUpdate.java → application/src/main/java/org/thingsboard/server/service/telemetry/sub/TsSubscriptionUpdate.java

@@ -24,14 +24,14 @@ import java.util.Map;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 
-public class SubscriptionUpdate {
+public class TsSubscriptionUpdate {
 
     private int subscriptionId;
     private int errorCode;
     private String errorMsg;
     private Map<String, List<Object>> data;
 
-    public SubscriptionUpdate(int subscriptionId, List<TsKvEntry> data) {
+    public TsSubscriptionUpdate(int subscriptionId, List<TsKvEntry> data) {
         super();
         this.subscriptionId = subscriptionId;
         this.data = new TreeMap<>();
@@ -46,17 +46,17 @@ public class SubscriptionUpdate {
         }
     }
 
-    public SubscriptionUpdate(int subscriptionId, Map<String, List<Object>> data) {
+    public TsSubscriptionUpdate(int subscriptionId, Map<String, List<Object>> data) {
         super();
         this.subscriptionId = subscriptionId;
         this.data = data;
     }
 
-    public SubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode) {
+    public TsSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode) {
         this(subscriptionId, errorCode, null);
     }
 
-    public SubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode, String errorMsg) {
+    public TsSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode, String errorMsg) {
         super();
         this.subscriptionId = subscriptionId;
         this.errorCode = errorCode.getCode();
@@ -93,7 +93,7 @@ public class SubscriptionUpdate {
 
     @Override
     public String toString() {
-        return "SubscriptionUpdate [subscriptionId=" + subscriptionId + ", errorCode=" + errorCode + ", errorMsg=" + errorMsg + ", data="
+        return "TsSubscriptionUpdate [subscriptionId=" + subscriptionId + ", errorCode=" + errorCode + ", errorMsg=" + errorMsg + ", data="
                 + data + "]";
     }
 }

common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmOperationResult.java

@@ -0,0 +1,14 @@
+package org.thingsboard.server.dao.alarm;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.id.EntityId;
+
+import java.util.List;
+
+@Data
+public class AlarmOperationResult {
+    private final Alarm alarm;
+    private final boolean successful;
+    private final List<EntityId> propagatedEntitiesList;
+}

common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java

@@ -39,13 +39,13 @@ import java.util.Collection;
  */
 public interface AlarmService {
 
-    Alarm createOrUpdateAlarm(Alarm alarm);
+    AlarmOperationResult createOrUpdateAlarm(Alarm alarm);
 
-    Boolean deleteAlarm(TenantId tenantId, AlarmId alarmId);
+    AlarmOperationResult deleteAlarm(TenantId tenantId, AlarmId alarmId);
 
-    ListenableFuture<Boolean> ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTs);
+    ListenableFuture<AlarmOperationResult> ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTs);
 
-    ListenableFuture<Boolean> clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTs);
+    ListenableFuture<AlarmOperationResult> clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTs);
 
     ListenableFuture<Alarm> findAlarmByIdAsync(TenantId tenantId, AlarmId alarmId);
 

common/data/src/main/java/org/thingsboard/server/common/data/alarm/AlarmSearchStatus.java

@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -15,8 +15,26 @@
  */
 package org.thingsboard.server.common.data.alarm;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import lombok.Getter;
+
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
 public enum AlarmSearchStatus {
 
-    ANY, ACTIVE, CLEARED, ACK, UNACK
+    ANY(AlarmStatus.values()),
+    ACTIVE(AlarmStatus.ACTIVE_ACK, AlarmStatus.ACTIVE_UNACK),
+    CLEARED(AlarmStatus.CLEARED_ACK, AlarmStatus.CLEARED_UNACK),
+    ACK(AlarmStatus.ACTIVE_ACK, AlarmStatus.CLEARED_ACK),
+    UNACK(AlarmStatus.ACTIVE_UNACK, AlarmStatus.CLEARED_UNACK);
+
+    @JsonIgnore
+    @Getter
+    private Set<AlarmStatus> statuses;
 
+    AlarmSearchStatus(AlarmStatus... statuses) {
+        this.statuses = new LinkedHashSet<>(Arrays.asList(statuses));
+    }
 }

common/data/src/main/java/org/thingsboard/server/common/data/query/AlarmData.java

@@ -15,7 +15,7 @@
  */
 package org.thingsboard.server.common.data.query;
 
-import lombok.Data;
+import lombok.Getter;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmInfo;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -26,10 +26,12 @@ import java.util.UUID;
 
 public class AlarmData extends AlarmInfo {
 
-    private final UUID entityId;
+    @Getter
+    private final EntityId entityId;
+    @Getter
     private final Map<EntityKeyType, Map<String, TsValue>> latest;
 
-    public AlarmData(Alarm alarm, String originatorName, UUID entityId) {
+    public AlarmData(Alarm alarm, String originatorName, EntityId entityId) {
         super(alarm, originatorName);
         this.entityId = entityId;
         this.latest = new HashMap<>();

dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java

@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -59,7 +59,9 @@ import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Comparator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -76,7 +78,6 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
 
     public static final String INCORRECT_TENANT_ID = "Incorrect tenantId ";
     public static final String INCORRECT_CUSTOMER_ID = "Incorrect customerId ";
-    public static final String ALARM_RELATION_PREFIX = "ALARM_";
 
     @Autowired
     private AlarmDao alarmDao;
@@ -102,7 +103,7 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
     }
 
     @Override
-    public Alarm createOrUpdateAlarm(Alarm alarm) {
+    public AlarmOperationResult createOrUpdateAlarm(Alarm alarm) {
         alarmDataValidator.validate(alarm, Alarm::getTenantId);
         try {
             if (alarm.getStartTs() == 0L) {
@@ -153,25 +154,33 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
         }
     }
 
-    private Alarm createAlarm(Alarm alarm) throws InterruptedException, ExecutionException {
+    private AlarmOperationResult createAlarm(Alarm alarm) throws InterruptedException, ExecutionException {
         log.debug("New Alarm : {}", alarm);
         Alarm saved = alarmDao.save(alarm.getTenantId(), alarm);
-        createAlarmRelations(saved);
+        List<EntityId> propagatedEntitiesList = createAlarmRelations(saved);
-        return saved;
+        return new AlarmOperationResult(alarm, true, propagatedEntitiesList);
     }
 
-    private void createAlarmRelations(Alarm alarm) throws InterruptedException, ExecutionException {
+    private List<EntityId> createAlarmRelations(Alarm alarm) throws InterruptedException, ExecutionException {
+        List<EntityId> propagatedEntitiesList;
         if (alarm.isPropagate()) {
-            List<EntityId> parentEntities = getParentEntities(alarm);
+            Set<EntityId> parentEntities = getParentEntities(alarm);
+            propagatedEntitiesList = new ArrayList<>(parentEntities.size() + 1);
             for (EntityId parentId : parentEntities) {
+                propagatedEntitiesList.add(parentId);
                 createAlarmRelation(alarm.getTenantId(), parentId, alarm.getId(), alarm.getStatus(), true);
             }
+            propagatedEntitiesList.add(alarm.getOriginator());
+        } else {
+            propagatedEntitiesList = Collections.singletonList(alarm.getOriginator());
         }
         createAlarmRelation(alarm.getTenantId(), alarm.getOriginator(), alarm.getId(), alarm.getStatus(), true);
+        return propagatedEntitiesList;
     }
 
-    private List<EntityId> getParentEntities(Alarm alarm) throws InterruptedException, ExecutionException {
+    private Set<EntityId> getParentEntities(Alarm alarm) throws InterruptedException, ExecutionException {
         EntityRelationsQuery query = new EntityRelationsQuery();
+        //TODO 3.1: @dlandiak we need to fetch max 3 levels and then fetch more if needed and there is at least one non-duplicate.
         RelationsSearchParameters parameters = new RelationsSearchParameters(alarm.getOriginator(), EntitySearchDirection.TO, Integer.MAX_VALUE, false);
         query.setParameters(parameters);
         List<String> propagateRelationTypes = alarm.getPropagateRelationTypes();
@@ -179,15 +188,15 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
         if (!CollectionUtils.isEmpty(propagateRelationTypes)) {
             relations = relations.filter(entityRelation -> propagateRelationTypes.contains(entityRelation.getType()));
         }
-        return relations.map(EntityRelation::getFrom).collect(Collectors.toList());
+        return relations.map(EntityRelation::getFrom).collect(Collectors.toCollection(LinkedHashSet::new));
     }
 
-    private ListenableFuture<Alarm> updateAlarm(Alarm update) {
+    private ListenableFuture<AlarmOperationResult> updateAlarm(Alarm update) {
         alarmDataValidator.validate(update, Alarm::getTenantId);
-        return getAndUpdate(update.getTenantId(), update.getId(), new Function<Alarm, Alarm>() {
+        return getAndUpdate(update.getTenantId(), update.getId(), new Function<Alarm, AlarmOperationResult>() {
             @Nullable
             @Override
-            public Alarm apply(@Nullable Alarm alarm) {
+            public AlarmOperationResult apply(@Nullable Alarm alarm) {
                 if (alarm == null) {
                     return null;
                 } else {
@@ -197,23 +206,22 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
         });
     }
 
-    private Alarm updateAlarm(Alarm oldAlarm, Alarm newAlarm) {
+    private AlarmOperationResult updateAlarm(Alarm oldAlarm, Alarm newAlarm) {
-        AlarmStatus oldStatus = oldAlarm.getStatus();
-        AlarmStatus newStatus = newAlarm.getStatus();
         boolean oldPropagate = oldAlarm.isPropagate();
         boolean newPropagate = newAlarm.isPropagate();
         Alarm result = alarmDao.save(newAlarm.getTenantId(), merge(oldAlarm, newAlarm));
+        List<EntityId> propagatedEntitiesList;
         if (!oldPropagate && newPropagate) {
             try {
-                createAlarmRelations(result);
+                propagatedEntitiesList = createAlarmRelations(result);
             } catch (InterruptedException | ExecutionException e) {
                 log.warn("Failed to update alarm relations [{}]", result, e);
                 throw new RuntimeException(e);
             }
-        } else if (oldStatus != newStatus) {
+        } else {
-            updateRelations(oldAlarm, oldStatus, newStatus);
+            propagatedEntitiesList = new ArrayList<>(getPropagationEntityIds(result));
         }
-        return result;
+        return new AlarmOperationResult(result, true, propagatedEntitiesList);
     }
 
     @Override
@@ -383,37 +391,13 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
         return existing;
     }
 
-    private void updateRelations(Alarm alarm, AlarmStatus oldStatus, AlarmStatus newStatus) {
+    private Set<EntityId> getPropagationEntityIds(Alarm alarm) {
-        try {
+        List<EntityRelation> relations = relationService.findByTo(alarm.getTenantId(), alarm.getId(), RelationTypeGroup.ALARM);
-            List<EntityRelation> relations = relationService.findByToAsync(alarm.getTenantId(), alarm.getId(), RelationTypeGroup.ALARM).get();
+        return relations.stream().map(EntityRelation::getFrom).collect(Collectors.toSet());
-            Set<EntityId> parents = relations.stream().map(EntityRelation::getFrom).collect(Collectors.toSet());
-            for (EntityId parentId : parents) {
-                updateAlarmRelation(alarm.getTenantId(), parentId, alarm.getId(), oldStatus, newStatus);
-            }
-        } catch (ExecutionException | InterruptedException e) {
-            log.warn("[{}] Failed to update relations. Old status: [{}], New status: [{}]", alarm.getId(), oldStatus, newStatus);
-            throw new RuntimeException(e);
-        }
-    }
-
-    private void createAlarmRelation(TenantId tenantId, EntityId entityId, EntityId alarmId, AlarmStatus status, boolean createAnyRelation) {
-        if (createAnyRelation) {
-            createRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + AlarmSearchStatus.ANY.name(), RelationTypeGroup.ALARM));
-        }
-        createRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + status.name(), RelationTypeGroup.ALARM));
-        createRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + status.getClearSearchStatus().name(), RelationTypeGroup.ALARM));
-        createRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + status.getAckSearchStatus().name(), RelationTypeGroup.ALARM));
-    }
-
-    private void deleteAlarmRelation(TenantId tenantId, EntityId entityId, EntityId alarmId, AlarmStatus status) {
-        deleteRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + status.name(), RelationTypeGroup.ALARM));
-        deleteRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + status.getClearSearchStatus().name(), RelationTypeGroup.ALARM));
-        deleteRelation(tenantId, new EntityRelation(entityId, alarmId, ALARM_RELATION_PREFIX + status.getAckSearchStatus().name(), RelationTypeGroup.ALARM));
     }
 
-    private void updateAlarmRelation(TenantId tenantId, EntityId entityId, EntityId alarmId, AlarmStatus oldStatus, AlarmStatus newStatus) {
+    private void createAlarmRelation(TenantId tenantId, EntityId entityId, EntityId alarmId) {
-        deleteAlarmRelation(tenantId, entityId, alarmId, oldStatus);
+        createRelation(tenantId, new EntityRelation(entityId, alarmId, AlarmSearchStatus.ANY.name(), RelationTypeGroup.ALARM));
-        createAlarmRelation(tenantId, entityId, alarmId, newStatus, false);
     }
 
     private <T> ListenableFuture<T> getAndUpdate(TenantId tenantId, AlarmId alarmId, Function<Alarm, T> function) {

dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmDao.java

@@ -117,7 +117,7 @@ public class JpaAlarmDao extends JpaAbstractDao<AlarmEntity, Alarm> implements A
                     tenantId.getId(),
                     affectedEntity.getId(),
                     affectedEntity.getEntityType().name(),
-                    relationType,
+                    AlarmSearchStatus.ANY.name(),
                     query.getPageLink().getStartTime(),
                     query.getPageLink().getEndTime(),
                     Objects.toString(query.getPageLink().getTextSearch(), ""),

dao/src/main/java/org/thingsboard/server/dao/sql/query/AlarmDataAdapter.java

@@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmSeverity;
 import org.thingsboard.server.common.data.alarm.AlarmStatus;
 import org.thingsboard.server.common.data.id.AlarmId;
+import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.EntityIdFactory;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
@@ -32,10 +33,13 @@ import org.thingsboard.server.common.data.query.EntityDataPageLink;
 import org.thingsboard.server.dao.model.ModelConstants;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -45,19 +49,20 @@ public class AlarmDataAdapter {
 
     public static PageData<AlarmData> createAlarmData(EntityDataPageLink pageLink,
                                                       List<Map<String, Object>> rows,
-                                                      int totalElements) {
+                                                      int totalElements, Collection<EntityId> orderedEntityIds) {
+        Map<UUID, EntityId> entityIdMap = orderedEntityIds.stream().collect(Collectors.toMap(EntityId::getId, Function.identity()));
         int totalPages = pageLink.getPageSize() > 0 ? (int) Math.ceil((float) totalElements / pageLink.getPageSize()) : 1;
         int startIndex = pageLink.getPageSize() * pageLink.getPage();
         boolean hasNext = pageLink.getPageSize() > 0 && totalElements > startIndex + rows.size();
-        List<AlarmData> entitiesData = convertListToAlarmData(rows);
+        List<AlarmData> entitiesData = convertListToAlarmData(rows, entityIdMap);
         return new PageData<>(entitiesData, totalPages, totalElements, hasNext);
     }
 
-    private static List<AlarmData> convertListToAlarmData(List<Map<String, Object>> result) {
+    private static List<AlarmData> convertListToAlarmData(List<Map<String, Object>> result, Map<UUID, EntityId> entityIdMap) {
-        return result.stream().map(AlarmDataAdapter::toEntityData).collect(Collectors.toList());
+        return result.stream().map(tmp -> toEntityData(tmp, entityIdMap)).collect(Collectors.toList());
     }
 
-    private static AlarmData toEntityData(Map<String, Object> row) {
+    private static AlarmData toEntityData(Map<String, Object> row, Map<UUID, EntityId> entityIdMap) {
         Alarm alarm = new Alarm();
         alarm.setId(new AlarmId((UUID) row.get(ModelConstants.ID_PROPERTY)));
         alarm.setCreatedTime((long) row.get(ModelConstants.CREATED_TIME_PROPERTY));
@@ -91,7 +96,8 @@ public class AlarmDataAdapter {
         } else {
             alarm.setPropagateRelationTypes(Collections.emptyList());
         }
-        UUID entityId = (UUID) row.get(ModelConstants.ENTITY_ID_COLUMN);
+        UUID entityUuid = (UUID) row.get(ModelConstants.ENTITY_ID_COLUMN);
+        EntityId entityId = entityIdMap.get(entityUuid);
         Object originatorNameObj = row.get(ModelConstants.ALARM_ORIGINATOR_NAME_PROPERTY);
         String originatorName = originatorNameObj != null ? originatorNameObj.toString() : null;
         return new AlarmData(alarm, originatorName, entityId);

dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java

@@ -204,7 +204,7 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository {
             dataQuery = String.format("%s limit %s offset %s", dataQuery, pageLink.getPageSize(), startIndex);
         }
         List<Map<String, Object>> rows = jdbcTemplate.queryForList(dataQuery, ctx);
-        return AlarmDataAdapter.createAlarmData(pageLink, rows, totalElements);
+        return AlarmDataAdapter.createAlarmData(pageLink, rows, totalElements, orderedEntityIds);
     }
 
     private String buildPermissionsQuery(TenantId tenantId, CustomerId customerId, QueryContext ctx) {

dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java

@@ -24,7 +24,7 @@ import java.util.Arrays;
 
 @RunWith(ClasspathSuite.class)
 @ClassnameFilters({
-        "org.thingsboard.server.dao.service.sql.AlarmServiceSqlTest"
+        "org.thingsboard.server.dao.service.sql.*SqlTest"
 })
 public class SqlDaoServiceTestSuite {
 

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineAlarmService.java

@@ -0,0 +1,57 @@
+/**
+ * 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 com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.alarm.AlarmInfo;
+import org.thingsboard.server.common.data.alarm.AlarmQuery;
+import org.thingsboard.server.common.data.alarm.AlarmSearchStatus;
+import org.thingsboard.server.common.data.alarm.AlarmSeverity;
+import org.thingsboard.server.common.data.alarm.AlarmStatus;
+import org.thingsboard.server.common.data.id.AlarmId;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.query.AlarmData;
+import org.thingsboard.server.common.data.query.AlarmDataPageLink;
+
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Created by ashvayka on 02.04.18.
+ */
+public interface RuleEngineAlarmService {
+
+    Alarm createOrUpdateAlarm(Alarm alarm);
+
+    Boolean deleteAlarm(TenantId tenantId, AlarmId alarmId);
+
+    ListenableFuture<Boolean> ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTs);
+
+    ListenableFuture<Boolean> clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTs);
+
+    ListenableFuture<Alarm> findAlarmByIdAsync(TenantId tenantId, AlarmId alarmId);
+
+    ListenableFuture<Alarm> findLatestByOriginatorAndType(TenantId tenantId, EntityId originator, String type);
+
+}

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java

@@ -169,7 +169,7 @@ public interface TbContext {
 
     DashboardService getDashboardService();
 
-    AlarmService getAlarmService();
+    RuleEngineAlarmService getAlarmService();
 
     RuleChainService getRuleChainService();
 

ui-ngx/package-lock.json

@@ -8997,10 +8997,10 @@
       "integrity": "sha512-4O3GWAYJaauMCILm07weko2rHA8a4kjn7+8Lg4s1d7SxwS/3IpkVD/GljbRrIJ1c1W/XGJ3GbuK7RyYZEJChhw=="
     },
     "ngx-flowchart": {
-      "version": "git://github.com/thingsboard/ngx-flowchart.git#7a02f4748b5e7821a883c903107af5f20415d026",
+      "version": "git://github.com/thingsboard/ngx-flowchart.git#a4157b0eef2eb3646ef920447c7b06b39d54f87f",
       "from": "git://github.com/thingsboard/ngx-flowchart.git#master",
       "requires": {
-        "tslib": "^1.13.0"
+        "tslib": "^1.10.0"
       },
       "dependencies": {
         "tslib": {