diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index 76e88104d2..89c79fcb8b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -103,6 +103,7 @@ import org.thingsboard.server.service.security.permission.AccessControlService; import org.thingsboard.server.service.security.permission.Operation; import org.thingsboard.server.service.security.permission.Resource; import org.thingsboard.server.service.state.DeviceStateService; +import org.thingsboard.server.service.telemetry.AlarmSubscriptionService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import javax.mail.MessagingException; @@ -145,7 +146,7 @@ public abstract class BaseController { protected AssetService assetService; @Autowired - protected AlarmService alarmService; + protected AlarmSubscriptionService alarmService; @Autowired protected DeviceCredentialsService deviceCredentialsService; diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index 41778e7c5e..b6e46e30ab 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -23,6 +23,7 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -52,7 +53,8 @@ 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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -195,7 +197,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer @Override public void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts, TbCallback callback) { - onLocalSubUpdate(entityId, + onLocalTelemetrySubUpdate(entityId, s -> { if (TbSubscriptionType.TIMESERIES.equals(s.getType())) { return (TbTimeseriesSubscription) s; @@ -219,7 +221,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer @Override public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbCallback callback) { - onLocalSubUpdate(entityId, + onLocalTelemetrySubUpdate(entityId, s -> { if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) { return (TbAttributeSubscription) s; @@ -256,9 +258,43 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer callback.onSuccess(); } + @Override + public void onAlarmUpdate(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback) { + onLocalAlarmSubUpdate(entityId, + s -> { + if (TbSubscriptionType.ALARMS.equals(s.getType())) { + return (TbAlarmsSubscription) s; + } else { + return null; + } + }, + s -> alarm.getCreatedTime() >= s.getTs(), + s -> alarm, + false + ); + callback.onSuccess(); + } + + @Override + public void onAlarmDeleted(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback) { + onLocalAlarmSubUpdate(entityId, + s -> { + if (TbSubscriptionType.ALARMS.equals(s.getType())) { + return (TbAlarmsSubscription) s; + } else { + return null; + } + }, + s -> alarm.getCreatedTime() >= s.getTs(), + s -> alarm, + true + ); + callback.onSuccess(); + } + @Override public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys, TbCallback callback) { - onLocalSubUpdate(entityId, + onLocalTelemetrySubUpdate(entityId, s -> { if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) { return (TbAttributeSubscription) s; @@ -282,17 +318,17 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer callback.onSuccess(); } - private void onLocalSubUpdate(EntityId entityId, - Function castFunction, - Predicate filterFunction, - Function> processFunction) { + private void onLocalTelemetrySubUpdate(EntityId entityId, + Function castFunction, + Predicate filterFunction, + Function> processFunction) { Set entitySubscriptions = subscriptionsByEntityId.get(entityId); if (entitySubscriptions != null) { entitySubscriptions.stream().map(castFunction).filter(Objects::nonNull).filter(filterFunction).forEach(s -> { List subscriptionUpdate = processFunction.apply(s); if (subscriptionUpdate != null && !subscriptionUpdate.isEmpty()) { if (serviceId.equals(s.getServiceId())) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate); + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate); localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY); } else { TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); @@ -305,6 +341,29 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } } + private void onLocalAlarmSubUpdate(EntityId entityId, + Function castFunction, + Predicate filterFunction, + Function processFunction, boolean deleted) { + Set entitySubscriptions = subscriptionsByEntityId.get(entityId); + if (entitySubscriptions != null) { + entitySubscriptions.stream().map(castFunction).filter(Objects::nonNull).filter(filterFunction).forEach(s -> { + Alarm alarm = processFunction.apply(s); + if (alarm != null) { + if (serviceId.equals(s.getServiceId())) { + AlarmSubscriptionUpdate update = new AlarmSubscriptionUpdate(s.getSubscriptionId(), alarm, deleted); + localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY); + } else { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); + toCoreNotificationsProducer.send(tpi, toProto(s, alarm), null); + } + } + }); + } else { + log.debug("[{}] No device subscriptions to process!", entityId); + } + } + private boolean isInTimeRange(TbTimeseriesSubscription subscription, long kvTime) { return (subscription.getStartTime() == 0 || subscription.getStartTime() <= kvTime) && (subscription.getEndTime() == 0 || subscription.getEndTime() >= kvTime); @@ -412,4 +471,19 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg); } + private TbProtoQueueMsg toProto(TbSubscription subscription, Alarm alarm) { + TbSubscriptionUpdateProto.Builder builder = TbSubscriptionUpdateProto.newBuilder(); + + builder.setSessionId(subscription.getSessionId()); + builder.setSubscriptionId(subscription.getSubscriptionId()); + + //TODO 3.1 + throw new RuntimeException("Not implemented!"); +// +// ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg( +// LocalSubscriptionServiceMsgProto.newBuilder().setSubUpdate(builder.build()).build()) +// .build(); +// return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index 439ed9be21..58957b01bb 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/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, diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index 53e84fc7cf..4614247f7c 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -36,7 +36,8 @@ 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.sub.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -137,7 +138,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer } @Override - public void onSubscriptionUpdate(String sessionId, TsSubscriptionUpdate update, TbCallback callback) { + public void onSubscriptionUpdate(String sessionId, TelemetrySubscriptionUpdate update, TbCallback callback) { TbSubscription subscription = subscriptionsBySessionId .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId()); if (subscription != null) { @@ -156,6 +157,16 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer callback.onSuccess(); } + @Override + public void onSubscriptionUpdate(String sessionId, AlarmSubscriptionUpdate update, TbCallback callback) { + TbSubscription subscription = subscriptionsBySessionId + .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId()); + if (subscription != null && subscription.getType() == TbSubscriptionType.ALARMS) { + subscription.getUpdateConsumer().accept(sessionId, update); + } + callback.onSuccess(); + } + @Override public void cancelSubscription(String sessionId, int subscriptionId) { log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java index ad02518719..5ca6b4f82e 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java @@ -17,6 +17,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.AlarmId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; @@ -38,5 +39,7 @@ public interface SubscriptionManagerService extends ApplicationListener keys, TbCallback empty); - void onAlarmUpdate(TenantId tenantId, EntityId entityId, Alarm alarm); + void onAlarmUpdate(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback); + + void onAlarmDeleted(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback); } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java index f76aa654d2..852c616d22 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java +++ b/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, @@ -18,6 +18,9 @@ package org.thingsboard.server.service.subscription; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.alarm.AlarmSearchStatus; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.AlarmData; @@ -26,15 +29,18 @@ 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.cmd.v2.AlarmDataUpdate; 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.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; @Slf4j public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx { @@ -44,6 +50,9 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx { private final LinkedHashMap entitiesMap; @Getter @Setter + private final HashMap alarmsMap; + @Getter + @Setter private PageData alarms; @Getter @Setter @@ -56,6 +65,7 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx { public TbAlarmDataSubCtx(String serviceId, TelemetryWebSocketService wsService, TelemetryWebSocketSessionRef sessionRef, int cmdId) { super(serviceId, wsService, sessionRef, cmdId); this.entitiesMap = new LinkedHashMap<>(); + this.alarmsMap = new HashMap<>(); } public void setEntitiesData(PageData entitiesData) { @@ -81,6 +91,8 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx { } } } + alarmsMap.clear(); + alarmsMap.putAll(alarms.getData().stream().collect(Collectors.toMap(AlarmData::getId, Function.identity()))); return this.alarms; } @@ -100,16 +112,64 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx { .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) { + Alarm alarm = subscriptionUpdate.getAlarm(); + AlarmId alarmId = alarm.getId(); + if (subscriptionUpdate.isAlarmDeleted()) { + Alarm deleted = alarmsMap.remove(alarmId); + if (deleted != null) { + //TODO: invalidate current page; + } + } else { + AlarmData current = alarmsMap.get(alarmId); + boolean onCurrentPage = current != null; + boolean matchesFilter = filter(alarm); + if (onCurrentPage) { + if (matchesFilter) { + AlarmData updated = new AlarmData(alarm, current.getName(), current.getEntityId()); + alarmsMap.put(alarmId, updated); + wsService.sendWsMsg(sessionId, new AlarmDataUpdate(cmdId, null, Collections.singletonList(updated))); + } else { + //TODO: invalidate current page; + } + } else if (matchesFilter && query.getPageLink().getPage() == 0) { + //TODO: invalidate current page; + } + } + } + private boolean filter(Alarm alarm) { + AlarmDataPageLink filter = query.getPageLink(); + long startTs = System.currentTimeMillis() - filter.getTimeWindow(); + if (alarm.getCreatedTime() < startTs) { + //Skip update that does not match time window. + return false; + } + if (filter.getTypeList() != null && !filter.getTypeList().isEmpty() && !filter.getTypeList().contains(alarm.getType())) { + return false; + } + if (filter.getSeverityList() != null && !filter.getSeverityList().isEmpty()) { + if (!filter.getSeverityList().contains(alarm.getSeverity())) { + return false; + } + } + if (filter.getStatusList() != null && !filter.getStatusList().isEmpty()) { + boolean matches = false; + for (AlarmSearchStatus status : filter.getStatusList()) { + if (status.getStatuses().contains(alarm.getStatus())) { + matches = true; + break; + } + } + if (!matches) { + return false; + } + } + return true; } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java index 8c76d14003..664467459a 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.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, @@ -16,35 +16,26 @@ package org.thingsboard.server.service.subscription; import lombok.Builder; +import lombok.Getter; 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 { + @Getter private final long ts; - private final List typeList; - private final List statusList; - private final List severityList; - private final boolean searchPropagatedAlarms; @Builder public TbAlarmsSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, - TbSubscriptionType type, BiConsumer updateConsumer, - long ts, List typeList, List statusList, - List severityList, boolean searchPropagatedAlarms) { + TbSubscriptionType type, BiConsumer updateConsumer, long ts) { super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateConsumer); this.ts = ts; - this.typeList = typeList; - this.statusList = statusList; - this.severityList = severityList; - this.searchPropagatedAlarms = searchPropagatedAlarms; } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java index b401d8400a..52a90c9bbf 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAttributeSubscription.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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import java.util.Map; import java.util.function.BiConsumer; -public class TbAttributeSubscription extends TbSubscription { +public class TbAttributeSubscription extends TbSubscription { @Getter private final boolean allKeys; @Getter private final Map keyStates; @@ -32,7 +32,7 @@ public class TbAttributeSubscription extends TbSubscription updateConsumer, + BiConsumer updateConsumer, boolean allKeys, Map keyStates, TbAttributeSubscriptionScope scope) { super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.ATTRIBUTES, updateConsumer); this.allKeys = allKeys; diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java index bc89bf8d58..85318efe26 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java +++ b/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, @@ -33,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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import java.util.ArrayList; import java.util.Arrays; @@ -169,11 +169,11 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx { return keyStates; } - private void sendWsMsg(String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { + private void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { sendWsMsg(sessionId, subscriptionUpdate, keyType, true); } - private void sendWsMsg(String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType, boolean resultToLatestValues) { + private void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate 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); @@ -187,7 +187,7 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx { } } - private void sendLatestWsMsg(EntityId entityId, String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { + private void sendLatestWsMsg(EntityId entityId, String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { Map latestUpdate = new HashMap<>(); subscriptionUpdate.getData().forEach((k, v) -> { Object[] data = (Object[]) v.get(0); @@ -226,7 +226,7 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx { } } - private void sendTsWsMsg(EntityId entityId, String sessionId, TsSubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { + private void sendTsWsMsg(EntityId entityId, String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { Map> tsUpdate = new HashMap<>(); subscriptionUpdate.getData().forEach((k, v) -> { Object[] data = (Object[]) v.get(0); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java index e16b70f74c..e8c0c46505 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java @@ -18,7 +18,8 @@ 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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; public interface TbLocalSubscriptionService { @@ -28,7 +29,9 @@ public interface TbLocalSubscriptionService { void cancelAllSessionSubscriptions(String sessionId); - void onSubscriptionUpdate(String sessionId, TsSubscriptionUpdate update, TbCallback callback); + void onSubscriptionUpdate(String sessionId, TelemetrySubscriptionUpdate update, TbCallback callback); + + void onSubscriptionUpdate(String sessionId, AlarmSubscriptionUpdate update, TbCallback callback); void onApplicationEvent(PartitionChangeEvent event); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java index dfb655b9c9..fbb9957969 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.subscription; +import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; @@ -44,7 +45,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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import java.util.ArrayList; import java.util.HashMap; @@ -137,9 +138,9 @@ public class TbSubscriptionUtils { return builder.build(); } - public static TsSubscriptionUpdate fromProto(TbSubscriptionUpdateProto proto) { + public static TelemetrySubscriptionUpdate fromProto(TbSubscriptionUpdateProto proto) { if (proto.getErrorCode() > 0) { - return new TsSubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg()); + return new TelemetrySubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg()); } else { Map> data = new TreeMap<>(); proto.getDataList().forEach(v -> { @@ -151,7 +152,7 @@ public class TbSubscriptionUtils { values.add(value); } }); - return new TsSubscriptionUpdate(proto.getSubscriptionId(), data); + return new TelemetrySubscriptionUpdate(proto.getSubscriptionId(), data); } } @@ -261,4 +262,14 @@ public class TbSubscriptionUtils { } return entry; } + + public static ToCoreMsg toAlarmUpdateProto(TenantId tenantId, EntityId entityId, Alarm alarm) { +// TODO: 3.1 + throw new RuntimeException("Not implemented!"); + } + + public static ToCoreMsg toAlarmDeletedProto(TenantId tenantId, EntityId entityId, Alarm alarm) { +// TODO: 3.1 + throw new RuntimeException("Not implemented!"); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java index c0d3a63379..3ee55da7a5 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java +++ b/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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import java.util.Map; import java.util.function.BiConsumer; -public class TbTimeseriesSubscription extends TbSubscription { +public class TbTimeseriesSubscription extends TbSubscription { @Getter private final boolean allKeys; @@ -37,7 +37,7 @@ public class TbTimeseriesSubscription extends TbSubscription updateConsumer, + BiConsumer updateConsumer, boolean allKeys, Map keyStates, long startTime, long endTime) { super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.TIMESERIES, updateConsumer); this.allKeys = allKeys; diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java index bb049400bd..bc4099e81b 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.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, diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java index 71a4c843b9..1182848669 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.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, @@ -20,6 +20,7 @@ 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.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.event.EventListener; import org.springframework.stereotype.Service; @@ -47,6 +48,7 @@ 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.AlarmOperationResult; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; @@ -58,7 +60,6 @@ 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; @@ -99,46 +100,32 @@ public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService @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 relatedEntities = Collections.singletonList(result.getOriginator()); - pushAlarmToSubService(result, relatedEntities); - return result; - } - - private void pushAlarmToSubService(Alarm result, List 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); - } - } - }); + AlarmOperationResult result = alarmService.createOrUpdateAlarm(alarm); + if (result.isSuccessful()) { + onAlarmUpdated(result); + } + return result.getAlarm(); } @Override public Boolean deleteAlarm(TenantId tenantId, AlarmId alarmId) { - return alarmService.deleteAlarm(tenantId, alarmId); + AlarmOperationResult result = alarmService.deleteAlarm(tenantId, alarmId); + onAlarmDeleted(result); + return result.isSuccessful(); } @Override public ListenableFuture ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTs) { - return alarmService.ackAlarm(tenantId, alarmId, ackTs); + ListenableFuture result = alarmService.ackAlarm(tenantId, alarmId, ackTs); + Futures.addCallback(result, new AlarmUpdateCallback(), wsCallBackExecutor); + return Futures.transform(result, AlarmOperationResult::isSuccessful, wsCallBackExecutor); } @Override public ListenableFuture clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTs) { - return alarmService.clearAlarm(tenantId, alarmId, details, clearTs); + ListenableFuture result = alarmService.clearAlarm(tenantId, alarmId, details, clearTs); + Futures.addCallback(result, new AlarmUpdateCallback(), wsCallBackExecutor); + return Futures.transform(result, AlarmOperationResult::isSuccessful, wsCallBackExecutor); } @Override @@ -146,10 +133,81 @@ public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService return alarmService.findAlarmByIdAsync(tenantId, alarmId); } + @Override + public ListenableFuture findAlarmInfoByIdAsync(TenantId tenantId, AlarmId alarmId) { + return alarmService.findAlarmInfoByIdAsync(tenantId, alarmId); + } + + @Override + public ListenableFuture> findAlarms(TenantId tenantId, AlarmQuery query) { + return alarmService.findAlarms(tenantId, query); + } + + @Override + public AlarmSeverity findHighestAlarmSeverity(TenantId tenantId, EntityId entityId, AlarmSearchStatus alarmSearchStatus, AlarmStatus alarmStatus) { + return alarmService.findHighestAlarmSeverity(tenantId, entityId, alarmSearchStatus, alarmStatus); + } + + @Override + public PageData findAlarmDataByQueryForEntities(TenantId tenantId, CustomerId customerId, AlarmDataPageLink pageLink, Collection orderedEntityIds) { + return alarmService.findAlarmDataByQueryForEntities(tenantId, customerId, pageLink, orderedEntityIds); + } + @Override public ListenableFuture findLatestByOriginatorAndType(TenantId tenantId, EntityId originator, String type) { return alarmService.findLatestByOriginatorAndType(tenantId, originator, type); } + private void onAlarmUpdated(AlarmOperationResult result) { + wsCallBackExecutor.submit(() -> { + Alarm alarm = result.getAlarm(); + TenantId tenantId = result.getAlarm().getTenantId(); + for (EntityId entityId : result.getPropagatedEntitiesList()) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + if (currentPartitions.contains(tpi)) { + if (subscriptionManagerService.isPresent()) { + subscriptionManagerService.get().onAlarmUpdate(tenantId, entityId, alarm); + } else { + log.warn("Possible misconfiguration because subscriptionManagerService is null!"); + } + } else { + TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toAlarmUpdateProto(tenantId, entityId, alarm); + clusterService.pushMsgToCore(tpi, entityId.getId(), toCoreMsg, null); + } + } + }); + } + + private void onAlarmDeleted(AlarmOperationResult result) { + wsCallBackExecutor.submit(() -> { + Alarm alarm = result.getAlarm(); + TenantId tenantId = result.getAlarm().getTenantId(); + for (EntityId entityId : result.getPropagatedEntitiesList()) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + if (currentPartitions.contains(tpi)) { + if (subscriptionManagerService.isPresent()) { + subscriptionManagerService.get().onAlarmDeleted(tenantId, entityId, alarm); + } else { + log.warn("Possible misconfiguration because subscriptionManagerService is null!"); + } + } else { + TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toAlarmDeletedProto(tenantId, entityId, alarm); + clusterService.pushMsgToCore(tpi, entityId.getId(), toCoreMsg, null); + } + } + }); + } + + private class AlarmUpdateCallback implements FutureCallback { + @Override + public void onSuccess(@Nullable AlarmOperationResult result) { + onAlarmUpdated(result); + } + + @Override + public void onFailure(Throwable t) { + log.warn("Failed to update alarm", t); + } + } } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index fb6494dc64..d40ce8bfa9 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/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, diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java index 66dea5abfe..7cdb911ae2 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java @@ -69,7 +69,7 @@ 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.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; import javax.annotation.Nullable; import javax.annotation.PostConstruct; @@ -219,7 +219,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi } } catch (IOException e) { log.warn("Failed to decode subscription cmd: {}", e.getMessage(), e); - TsSubscriptionUpdate update = new TsSubscriptionUpdate(UNKNOWN_SUBSCRIPTION_ID, SubscriptionErrorCode.INTERNAL_ERROR, SESSION_META_DATA_NOT_FOUND); + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(UNKNOWN_SUBSCRIPTION_ID, SubscriptionErrorCode.INTERNAL_ERROR, SESSION_META_DATA_NOT_FOUND); sendWsMsg(sessionRef, update); } } @@ -254,7 +254,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi } @Override - public void sendWsMsg(String sessionId, TsSubscriptionUpdate update) { + public void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate update) { sendWsMsg(sessionId, update.getSubscriptionId(), update); } @@ -397,7 +397,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Override public void onSuccess(List data) { List attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList()); - sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), attributesData)); + sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), attributesData)); Map subState = new HashMap<>(keys.size()); keys.forEach(key -> subState.put(key, 0L)); @@ -422,12 +422,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Override public void onFailure(Throwable e) { log.error(FAILED_TO_FETCH_ATTRIBUTES, e); - TsSubscriptionUpdate update; + TelemetrySubscriptionUpdate update; if (e instanceof UnauthorizedException) { - update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg()); } else { - update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, FAILED_TO_FETCH_ATTRIBUTES); } sendWsMsg(sessionRef, update); @@ -446,19 +446,19 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId); if (sessionMD == null) { log.warn("[{}] Session meta data not found. ", sessionId); - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(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()) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Device id is empty!"); sendWsMsg(sessionRef, update); return; } if (cmd.getKeys() == null || cmd.getKeys().isEmpty()) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Keys are empty!"); sendWsMsg(sessionRef, update); return; @@ -471,17 +471,17 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi FutureCallback> callback = new FutureCallback>() { @Override public void onSuccess(List data) { - sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), data)); + sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), data)); } @Override public void onFailure(Throwable e) { - TsSubscriptionUpdate update; + TelemetrySubscriptionUpdate update; if (UnauthorizedException.class.isInstance(e)) { - update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg()); } else { - update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, FAILED_TO_FETCH_DATA); } sendWsMsg(sessionRef, update); @@ -497,7 +497,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Override public void onSuccess(List data) { List attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList()); - sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), attributesData)); + sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), attributesData)); Map subState = new HashMap<>(attributesData.size()); attributesData.forEach(v -> subState.put(v.getKey(), v.getTs())); @@ -520,7 +520,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Override public void onFailure(Throwable e) { log.error(FAILED_TO_FETCH_ATTRIBUTES, e); - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, FAILED_TO_FETCH_ATTRIBUTES); sendWsMsg(sessionRef, update); } @@ -583,7 +583,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi FutureCallback> callback = new FutureCallback>() { @Override public void onSuccess(List data) { - sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), data)); + sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), data)); Map subState = new HashMap<>(data.size()); data.forEach(v -> subState.put(v.getKey(), v.getTs())); @@ -601,12 +601,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi @Override public void onFailure(Throwable e) { - TsSubscriptionUpdate update; + TelemetrySubscriptionUpdate update; if (UnauthorizedException.class.isInstance(e)) { - update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg()); } else { - update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, FAILED_TO_FETCH_DATA); } sendWsMsg(sessionRef, update); @@ -620,7 +620,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi return new FutureCallback>() { @Override public void onSuccess(List data) { - sendWsMsg(sessionRef, new TsSubscriptionUpdate(cmd.getCmdId(), data)); + sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), data)); Map subState = new HashMap<>(keys.size()); keys.forEach(key -> subState.put(key, startTs)); data.forEach(v -> subState.put(v.getKey(), v.getTs())); @@ -644,7 +644,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi } else { log.info(FAILED_TO_FETCH_DATA, e); } - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, FAILED_TO_FETCH_DATA); sendWsMsg(sessionRef, update); } @@ -661,12 +661,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) { if (cmd.getCmdId() < 0) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(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) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Query is empty!"); sendWsMsg(sessionRef, update); return false; @@ -676,12 +676,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) { if (cmd.getCmdId() < 0) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Cmd id is negative value!"); sendWsMsg(sessionRef, update); return false; } else if (cmd.getQuery() == null) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Query is empty!"); sendWsMsg(sessionRef, update); return false; @@ -691,7 +691,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) { if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) { - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, "Device id is empty!"); sendWsMsg(sessionRef, update); return false; @@ -707,7 +707,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId); if (sessionMD == null) { log.warn("[{}] Session meta data not found. ", sessionId); - TsSubscriptionUpdate update = new TsSubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR, + TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR, SESSION_META_DATA_NOT_FOUND); sendWsMsg(sessionRef, update); return false; @@ -720,7 +720,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi sendWsMsg(sessionRef, update.getCmdId(), update); } - private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, TsSubscriptionUpdate update) { + private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, TelemetrySubscriptionUpdate update) { sendWsMsg(sessionRef, update.getSubscriptionId(), update); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java index 5308b539ba..45c061c40a 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java @@ -16,7 +16,7 @@ package org.thingsboard.server.service.telemetry; import org.thingsboard.server.service.telemetry.cmd.v2.DataUpdate; -import org.thingsboard.server.service.telemetry.sub.TsSubscriptionUpdate; +import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; /** * Created by ashvayka on 27.03.18. @@ -27,7 +27,7 @@ public interface TelemetryWebSocketService { void handleWebSocketMsg(TelemetryWebSocketSessionRef sessionRef, String msg); - void sendWsMsg(String sessionId, TsSubscriptionUpdate update); + void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate update); void sendWsMsg(String sessionId, DataUpdate update); diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/AlarmSubscriptionUpdate.java b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/AlarmSubscriptionUpdate.java index 340228f389..87a94dae96 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/AlarmSubscriptionUpdate.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/AlarmSubscriptionUpdate.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.telemetry.sub; +import lombok.Getter; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.query.AlarmData; @@ -28,15 +29,26 @@ import java.util.stream.Collectors; public class AlarmSubscriptionUpdate { + @Getter private int subscriptionId; + @Getter private int errorCode; + @Getter private String errorMsg; + @Getter private Alarm alarm; + @Getter + private boolean alarmDeleted; public AlarmSubscriptionUpdate(int subscriptionId, Alarm alarm) { + this(subscriptionId, alarm, false); + } + + public AlarmSubscriptionUpdate(int subscriptionId, Alarm alarm, boolean alarmDeleted) { super(); this.subscriptionId = subscriptionId; this.alarm = alarm; + this.alarmDeleted = alarmDeleted; } public AlarmSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode) { @@ -50,19 +62,6 @@ public class AlarmSubscriptionUpdate { 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=" diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/TsSubscriptionUpdate.java b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/TelemetrySubscriptionUpdate.java similarity index 86% rename from application/src/main/java/org/thingsboard/server/service/telemetry/sub/TsSubscriptionUpdate.java rename to application/src/main/java/org/thingsboard/server/service/telemetry/sub/TelemetrySubscriptionUpdate.java index 5ed5fca1f1..bd5359f642 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/TsSubscriptionUpdate.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/TelemetrySubscriptionUpdate.java @@ -24,14 +24,14 @@ import java.util.Map; import java.util.TreeMap; import java.util.stream.Collectors; -public class TsSubscriptionUpdate { +public class TelemetrySubscriptionUpdate { private int subscriptionId; private int errorCode; private String errorMsg; private Map> data; - public TsSubscriptionUpdate(int subscriptionId, List data) { + public TelemetrySubscriptionUpdate(int subscriptionId, List data) { super(); this.subscriptionId = subscriptionId; this.data = new TreeMap<>(); @@ -46,17 +46,17 @@ public class TsSubscriptionUpdate { } } - public TsSubscriptionUpdate(int subscriptionId, Map> data) { + public TelemetrySubscriptionUpdate(int subscriptionId, Map> data) { super(); this.subscriptionId = subscriptionId; this.data = data; } - public TsSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode) { + public TelemetrySubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode) { this(subscriptionId, errorCode, null); } - public TsSubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode, String errorMsg) { + public TelemetrySubscriptionUpdate(int subscriptionId, SubscriptionErrorCode errorCode, String errorMsg) { super(); this.subscriptionId = subscriptionId; this.errorCode = errorCode.getCode(); diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmOperationResult.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmOperationResult.java index f0f18b894c..291d358829 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmOperationResult.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmOperationResult.java @@ -1,9 +1,26 @@ +/** + * 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.dao.alarm; +import lombok.AllArgsConstructor; import lombok.Data; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.id.EntityId; +import java.util.Collections; import java.util.List; @Data @@ -11,4 +28,16 @@ public class AlarmOperationResult { private final Alarm alarm; private final boolean successful; private final List propagatedEntitiesList; + + public AlarmOperationResult(Alarm alarm, boolean successful) { + this.alarm = alarm; + this.successful = successful; + this.propagatedEntitiesList = Collections.emptyList(); + } + + public AlarmOperationResult(Alarm alarm, boolean successful, List propagatedEntitiesList) { + this.alarm = alarm; + this.successful = successful; + this.propagatedEntitiesList = propagatedEntitiesList; + } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/AlarmSearchStatus.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/AlarmSearchStatus.java index 6095cf7caa..4231170d69 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/AlarmSearchStatus.java +++ b/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, diff --git a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java index 1984892563..b4691c4683 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java +++ b/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, @@ -140,15 +140,17 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ } @Override - public Boolean deleteAlarm(TenantId tenantId, AlarmId alarmId) { + public AlarmOperationResult deleteAlarm(TenantId tenantId, AlarmId alarmId) { try { log.debug("Deleting Alarm Id: {}", alarmId); Alarm alarm = alarmDao.findAlarmByIdAsync(tenantId, alarmId.getId()).get(); if (alarm == null) { - return false; + return new AlarmOperationResult(alarm, false); } + AlarmOperationResult result = new AlarmOperationResult(alarm, true, new ArrayList<>(getPropagationEntityIds(alarm))); deleteEntityRelations(tenantId, alarm.getId()); - return alarmDao.deleteAlarm(tenantId, alarm); + alarmDao.deleteAlarm(tenantId, alarm); + return result; } catch (ExecutionException | InterruptedException e) { throw new RuntimeException(e); } @@ -158,7 +160,7 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ log.debug("New Alarm : {}", alarm); Alarm saved = alarmDao.save(alarm.getTenantId(), alarm); List propagatedEntitiesList = createAlarmRelations(saved); - return new AlarmOperationResult(alarm, true, propagatedEntitiesList); + return new AlarmOperationResult(saved, true, propagatedEntitiesList); } private List createAlarmRelations(Alarm alarm) throws InterruptedException, ExecutionException { @@ -168,13 +170,13 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ propagatedEntitiesList = new ArrayList<>(parentEntities.size() + 1); for (EntityId parentId : parentEntities) { propagatedEntitiesList.add(parentId); - createAlarmRelation(alarm.getTenantId(), parentId, alarm.getId(), alarm.getStatus(), true); + createAlarmRelation(alarm.getTenantId(), parentId, alarm.getId()); } propagatedEntitiesList.add(alarm.getOriginator()); } else { propagatedEntitiesList = Collections.singletonList(alarm.getOriginator()); } - createAlarmRelation(alarm.getTenantId(), alarm.getOriginator(), alarm.getId(), alarm.getStatus(), true); + createAlarmRelation(alarm.getTenantId(), alarm.getOriginator(), alarm.getId()); return propagatedEntitiesList; } @@ -225,34 +227,33 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ } @Override - public ListenableFuture ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTime) { - return getAndUpdate(tenantId, alarmId, new Function() { + public ListenableFuture ackAlarm(TenantId tenantId, AlarmId alarmId, long ackTime) { + return getAndUpdate(tenantId, alarmId, new Function() { @Nullable @Override - public Boolean apply(@Nullable Alarm alarm) { + public AlarmOperationResult apply(@Nullable Alarm alarm) { if (alarm == null || alarm.getStatus().isAck()) { - return false; + return new AlarmOperationResult(alarm, false); } else { AlarmStatus oldStatus = alarm.getStatus(); AlarmStatus newStatus = oldStatus.isCleared() ? AlarmStatus.CLEARED_ACK : AlarmStatus.ACTIVE_ACK; alarm.setStatus(newStatus); alarm.setAckTs(ackTime); - alarmDao.save(alarm.getTenantId(), alarm); - updateRelations(alarm, oldStatus, newStatus); - return true; + alarm = alarmDao.save(alarm.getTenantId(), alarm); + return new AlarmOperationResult(alarm, true, new ArrayList<>(getPropagationEntityIds(alarm))); } } }); } @Override - public ListenableFuture clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTime) { - return getAndUpdate(tenantId, alarmId, new Function() { + public ListenableFuture clearAlarm(TenantId tenantId, AlarmId alarmId, JsonNode details, long clearTime) { + return getAndUpdate(tenantId, alarmId, new Function() { @Nullable @Override - public Boolean apply(@Nullable Alarm alarm) { + public AlarmOperationResult apply(@Nullable Alarm alarm) { if (alarm == null || alarm.getStatus().isCleared()) { - return false; + return new AlarmOperationResult(alarm, false); } else { AlarmStatus oldStatus = alarm.getStatus(); AlarmStatus newStatus = oldStatus.isAck() ? AlarmStatus.CLEARED_ACK : AlarmStatus.CLEARED_UNACK; @@ -261,9 +262,8 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ if (details != null) { alarm.setDetails(details); } - alarmDao.save(alarm.getTenantId(), alarm); - updateRelations(alarm, oldStatus, newStatus); - return true; + alarm = alarmDao.save(alarm.getTenantId(), alarm); + return new AlarmOperationResult(alarm, true, new ArrayList<>(getPropagationEntityIds(alarm))); } } }); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmDao.java index 630f09f8c1..20fff1fd5b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmDao.java @@ -102,15 +102,17 @@ public class JpaAlarmDao extends JpaAbstractDao implements A public PageData findAlarms(TenantId tenantId, AlarmQuery query) { log.trace("Try to find alarms by entity [{}], status [{}] and pageLink [{}]", query.getAffectedEntityId(), query.getStatus(), query.getPageLink()); EntityId affectedEntity = query.getAffectedEntityId(); - String searchStatusName; - if (query.getSearchStatus() == null && query.getStatus() == null) { - searchStatusName = AlarmSearchStatus.ANY.name(); - } else if (query.getSearchStatus() != null) { - searchStatusName = query.getSearchStatus().name(); - } else { - searchStatusName = query.getStatus().name(); - } - String relationType = BaseAlarmService.ALARM_RELATION_PREFIX + searchStatusName; + + //TODO 3.1: add search by statuses +// String searchStatusName; +// if (query.getSearchStatus() == null && query.getStatus() == null) { +// searchStatusName = AlarmSearchStatus.ANY.name(); +// } else if (query.getSearchStatus() != null) { +// searchStatusName = query.getSearchStatus().name(); +// } else { +// searchStatusName = query.getStatus().name(); +// } +// String relationType = BaseAlarmService.ALARM_RELATION_PREFIX; return DaoUtil.toPageData( alarmRepository.findAlarms( diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java index 00b6414b3f..a6b188a3fb 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java @@ -64,9 +64,9 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository { } public static final String SELECT_ORIGINATOR_NAME = " CASE" + - " WHEN a.originator_type = "+ EntityType.TENANT.ordinal() + + " WHEN a.originator_type = " + EntityType.TENANT.ordinal() + " THEN (select title from tenant where id = a.originator_id)" + - " WHEN a.originator_type = "+ EntityType.CUSTOMER.ordinal() + + " WHEN a.originator_type = " + EntityType.CUSTOMER.ordinal() + " THEN (select title from customer where id = a.originator_id)" + " WHEN a.originator_type = " + EntityType.USER.ordinal() + " THEN (select CONCAT (first_name, ' ', last_name) from tb_user where id = a.originator_id)" + @@ -156,17 +156,27 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository { sortPart.append("e.priority"); } - if (pageLink.getStartTs() > 0) { + long startTs; + long endTs; + if (pageLink.getTimeWindow() > 0) { + endTs = System.currentTimeMillis(); + startTs = endTs - pageLink.getTimeWindow(); + } else { + startTs = pageLink.getStartTs(); + endTs = pageLink.getEndTs(); + } + + if (startTs > 0) { addAndIfNeeded(wherePart, addAnd); addAnd = true; - ctx.addLongParameter("startTime", pageLink.getStartTs()); + ctx.addLongParameter("startTime", startTs); wherePart.append("a.created_time >= :startTime"); } - if (pageLink.getEndTs() > 0) { + if (endTs > 0) { addAndIfNeeded(wherePart, addAnd); addAnd = true; - ctx.addLongParameter("endTime", pageLink.getEndTs()); + ctx.addLongParameter("endTime", endTs); wherePart.append("a.created_time <= :endTime"); } diff --git a/dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java b/dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java index a6ef3935b0..db87deab93 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java +++ b/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.*SqlTest" + "org.thingsboard.server.dao.service.sql.AlarmServiceSqlTest" }) public class SqlDaoServiceTestSuite { diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java index ec0d10286b..ddb49b4b4e 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java @@ -43,6 +43,7 @@ import org.thingsboard.server.common.data.query.EntityKey; import org.thingsboard.server.common.data.query.EntityKeyType; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.dao.alarm.AlarmOperationResult; import java.util.Arrays; import java.util.Collections; @@ -84,7 +85,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { .severity(AlarmSeverity.CRITICAL).status(AlarmStatus.ACTIVE_UNACK) .startTs(ts).build(); - Alarm created = alarmService.createOrUpdateAlarm(alarm); + AlarmOperationResult result = alarmService.createOrUpdateAlarm(alarm); + Alarm created = result.getAlarm(); Assert.assertNotNull(created); Assert.assertNotNull(created.getId()); @@ -122,7 +124,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { .severity(AlarmSeverity.CRITICAL).status(AlarmStatus.ACTIVE_UNACK) .startTs(ts).build(); - Alarm created = alarmService.createOrUpdateAlarm(alarm); + AlarmOperationResult result = alarmService.createOrUpdateAlarm(alarm); + Alarm created = result.getAlarm(); // Check child relation PageData alarms = alarmService.findAlarms(tenantId, AlarmQuery.builder() @@ -146,7 +149,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { Assert.assertEquals(0, alarms.getData().size()); created.setPropagate(true); - created = alarmService.createOrUpdateAlarm(created); + result = alarmService.createOrUpdateAlarm(created); + created = result.getAlarm(); // Check child relation alarms = alarmService.findAlarms(tenantId, AlarmQuery.builder() @@ -234,7 +238,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { .propagate(true) .severity(AlarmSeverity.CRITICAL).status(AlarmStatus.ACTIVE_UNACK) .startTs(ts).build(); - tenantAlarm = alarmService.createOrUpdateAlarm(tenantAlarm); + AlarmOperationResult result = alarmService.createOrUpdateAlarm(tenantAlarm); + tenantAlarm = result.getAlarm(); Alarm deviceAlarm = Alarm.builder().tenantId(tenantId) .originator(customerDevice.getId()) @@ -242,7 +247,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { .propagate(true) .severity(AlarmSeverity.CRITICAL).status(AlarmStatus.ACTIVE_UNACK) .startTs(ts).build(); - deviceAlarm = alarmService.createOrUpdateAlarm(deviceAlarm); + result = alarmService.createOrUpdateAlarm(deviceAlarm); + deviceAlarm = result.getAlarm(); AlarmDataPageLink pageLink = new AlarmDataPageLink(); pageLink.setPage(0); @@ -281,7 +287,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { .status(AlarmStatus.ACTIVE_UNACK) .startTs(ts).build(); - Alarm created = alarmService.createOrUpdateAlarm(alarm); + AlarmOperationResult result = alarmService.createOrUpdateAlarm(alarm); + Alarm created = result.getAlarm(); AlarmDataPageLink pageLink = new AlarmDataPageLink(); pageLink.setPage(0); @@ -321,7 +328,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { Assert.assertEquals(created, new Alarm(alarms.getData().get(0))); created.setPropagate(true); - created = alarmService.createOrUpdateAlarm(created); + result = alarmService.createOrUpdateAlarm(created); + created = result.getAlarm(); // Check child relation pageLink.setPage(0); @@ -402,7 +410,8 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { .severity(AlarmSeverity.CRITICAL).status(AlarmStatus.ACTIVE_UNACK) .startTs(ts).build(); - Alarm created = alarmService.createOrUpdateAlarm(alarm); + AlarmOperationResult result = alarmService.createOrUpdateAlarm(alarm); + Alarm created = result.getAlarm(); PageData alarms = alarmService.findAlarms(tenantId, AlarmQuery.builder() .affectedEntityId(childId) @@ -426,16 +435,16 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { Assert.assertEquals(created, alarms.getData().get(0)); List toAlarmRelations = relationService.findByTo(tenantId, created.getId(), RelationTypeGroup.ALARM); - Assert.assertEquals(8, toAlarmRelations.size()); + Assert.assertEquals(2, toAlarmRelations.size()); List fromChildRelations = relationService.findByFrom(tenantId, childId, RelationTypeGroup.ALARM); - Assert.assertEquals(4, fromChildRelations.size()); + Assert.assertEquals(1, fromChildRelations.size()); - List fromParentRelations = relationService.findByFrom(tenantId, childId, RelationTypeGroup.ALARM); - Assert.assertEquals(4, fromParentRelations.size()); + List fromParentRelations = relationService.findByFrom(tenantId, parentId, RelationTypeGroup.ALARM); + Assert.assertEquals(1, fromParentRelations.size()); - Assert.assertTrue("Alarm was not deleted when expected", alarmService.deleteAlarm(tenantId, created.getId())); + Assert.assertTrue("Alarm was not deleted when expected", alarmService.deleteAlarm(tenantId, created.getId()).isSuccessful()); Alarm fetched = alarmService.findAlarmByIdAsync(tenantId, created.getId()).get(); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineAlarmService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineAlarmService.java index 4195bf61dc..3ba6945544 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineAlarmService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineAlarmService.java @@ -54,4 +54,11 @@ public interface RuleEngineAlarmService { ListenableFuture findLatestByOriginatorAndType(TenantId tenantId, EntityId originator, String type); + ListenableFuture findAlarmInfoByIdAsync(TenantId tenantId, AlarmId alarmId); + + ListenableFuture> findAlarms(TenantId tenantId, AlarmQuery query); + + AlarmSeverity findHighestAlarmSeverity(TenantId tenantId, EntityId entityId, AlarmSearchStatus alarmSearchStatus, AlarmStatus alarmStatus); + + PageData findAlarmDataByQueryForEntities(TenantId tenantId, CustomerId customerId, AlarmDataPageLink pageLink, Collection orderedEntityIds); }